feat(contracts): add typed Forge runtime probes (#8)

- feat(contracts): add typed Forge runtime probes (d89267a3c4bac6bce619e74b3b280a1b0623524a)
- docs(provenance): bound provider build timestamp semantics (813cf06b4255cca3f4e649d24e96e6e2f5e7f615)

README.md

@@ -4,6 +4,7 @@ emoji: "⚒️"
 colorFrom: green
 colorTo: gray
 sdk: gradio
+sdk_version: 6.20.0
 app_file: app.py
 pinned: false
 short_description: Evidence console for SZL Forge models and formulas.
@@ -34,6 +35,12 @@ The Gradio API exposes `/status`, `/integrity`, `/evaluation`, `/receipt`,
 `/curriculum-stage`. Exact inputs and outputs are visible in the Space's
 **Use via API** panel and its **Callable API** tab.
 
+Operational probes that require ordinary HTTP GET JSON use three separate,
+typed compatibility routes: `/.well-known/szl-source.json`, `/healthz`, and
+`/api/status`. They report transport, evidence, revision measurement, source
+alignment, and authority independently. In particular,
+`PENDING_GITHUB_SYNC` never implies GitHub/Hugging Face byte parity.
+
 ```python
 from gradio_client import Client
 
@@ -46,5 +53,5 @@ formulas = client.predict("lambda", "ALL", 10, api_name="/formulas")
 
 ```bash
 python -m unittest discover -s tests -v
-python -m py_compile forge_lab.py app.py
+python -m py_compile forge_lab.py forge_runtime_contract.py app.py
 ```

app.py

@@ -27,6 +27,7 @@ from forge_lab import (
     metrics,
     source_rows,
 )
+from forge_runtime_contract import register_contract_routes
 
 
 CSS = """
@@ -396,14 +397,21 @@ with gr.Blocks(title="SZL Forge Lab") as demo:
 
 
 if __name__ == "__main__":
-    demo.launch(
-        server_name="0.0.0.0",
-        server_port=int(os.environ.get("PORT", "7860")),
-        show_error=False,
-        # The managed SSR shell only packages Brotli variants for its /_app CSS.
-        # Use Gradio's client shell so assets remain reachable without relying on
-        # proxy content-encoding negotiation.
-        ssr_mode=False,
-        css=CSS,
-        theme=gr.themes.Base(),
-    )
+    try:
+        server_app, _, _ = demo.launch(
+            server_name="0.0.0.0",
+            server_port=int(os.environ.get("PORT", "7860")),
+            show_error=False,
+            prevent_thread_lock=True,
+            # The managed SSR shell only packages Brotli variants for its /_app CSS.
+            # Use Gradio's client shell so assets remain reachable without relying on
+            # proxy content-encoding negotiation.
+            ssr_mode=False,
+            css=CSS,
+            theme=gr.themes.Base(),
+        )
+        register_contract_routes(server_app, get_status)
+    except Exception:
+        demo.close()
+        raise
+    demo.block_thread()

forge_runtime_contract.py

@@ -0,0 +1,256 @@
+"""Typed, read-only HTTP compatibility contracts for the managed Forge Space.
+
+The Gradio event API remains the product API.  These routes are deliberately
+small operational contracts for probes that require ordinary HTTP GET JSON.
+They do not claim GitHub/Hugging Face byte parity or a reproducible build.
+"""
+from __future__ import annotations
+
+import json
+import os
+import re
+import threading
+import time
+import urllib.request
+from collections.abc import Callable
+from datetime import datetime, timezone
+from typing import Any
+
+from fastapi.responses import JSONResponse
+
+
+SPACE_ID = "SZLHOLDINGS/szl-forge-lab"
+SOURCE_REPOSITORY = "szl-holdings/szl-forge"
+SOURCE_COMMIT = "43e0aedb176fbc8868557e56e30501d8de24578e"
+SOURCE_PATH = ""
+ALIGNMENT_STATE = "PENDING_GITHUB_SYNC"
+_SHA = re.compile(r"^[0-9a-f]{40}$")
+_CACHE_LOCK = threading.Lock()
+_CACHE: dict[str, Any] = {"stored_at": 0.0, "measurement": None}
+
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
+
+
+def _valid_sha(value: object) -> str | None:
+    candidate = str(value or "").strip().lower()
+    return candidate if _SHA.fullmatch(candidate) else None
+
+
+def measure_hf_repository_head(force: bool = False) -> dict[str, Any]:
+    """Measure Hub repository-head evidence without calling it process identity."""
+
+    now = time.monotonic()
+    with _CACHE_LOCK:
+        cached = _CACHE.get("measurement")
+        if not force and cached and now - float(_CACHE["stored_at"]) < 60:
+            return dict(cached)
+
+    space_id = os.environ.get("SPACE_ID") or SPACE_ID
+    observed_at = _now_iso()
+    resolver = f"https://huggingface.co/api/spaces/{space_id}?expand[]=sha&expand[]=lastModified"
+    revision = None
+    last_modified = None
+    error = None
+    request = urllib.request.Request(
+        resolver,
+        headers={"Accept": "application/json", "User-Agent": "szl-forge-runtime-contract/1.0"},
+    )
+    try:
+        with urllib.request.urlopen(request, timeout=4) as response:
+            payload = json.load(response)
+        revision = _valid_sha(payload.get("sha"))
+        last_modified = payload.get("lastModified")
+    except Exception as exc:  # Network evidence is optional and reported honestly.
+        error = type(exc).__name__
+
+    measurement: dict[str, Any] = {
+        "hf_revision": revision,
+        "revision_state": "MEASURED_REPOSITORY_HEAD" if revision else "UNAVAILABLE",
+        "running_process_revision_state": "NOT_EXPOSED_IN_PROCESS",
+        "measurement_method": "HUGGINGFACE_HUB_API" if revision else "UNAVAILABLE",
+        "resolver": resolver,
+        "observed_at": observed_at,
+        "provider_last_modified": last_modified,
+    }
+    if error:
+        measurement["measurement_error"] = error
+    with _CACHE_LOCK:
+        _CACHE.update(stored_at=time.monotonic(), measurement=dict(measurement))
+    return measurement
+
+
+def build_source_attestation(force: bool = False) -> tuple[dict[str, Any], int]:
+    measurement = measure_hf_repository_head(force=force)
+    revision = measurement["hf_revision"]
+    if revision is None:
+        return (
+            {
+                "schema": "szl.deployment-source-unavailable/v1",
+                "observed_at": measurement["observed_at"],
+                "transport_state": "REACHABLE",
+                "evidence_state": "UNAVAILABLE",
+                "authority_state": "READ_ONLY",
+                "deployment": {"hf_space": SPACE_ID, **measurement},
+                "limits": [
+                    "Repository-head evidence was unavailable, so no deployment-source document was fabricated.",
+                    "Retry this GET with ?refresh=1 after Hub API reachability recovers.",
+                ],
+            },
+            503,
+        )
+
+    built_at = measurement.get("provider_last_modified") or measurement["observed_at"]
+    return (
+        {
+            "schema": "szl.deployment-source/v1",
+            "source": {
+                "repository": SOURCE_REPOSITORY,
+                "commit": SOURCE_COMMIT,
+                "path": SOURCE_PATH,
+            },
+            "deployment": {"hf_space": SPACE_ID, "hf_revision": revision},
+            "built_at": built_at,
+            "alignment_state": ALIGNMENT_STATE,
+            "extensions": {
+                "schema": "szl.forge.deployment-source-extension/v1",
+                "deployment_revision_evidence": measurement,
+                "source_observation": {
+                    "state": "VERIFIED_REFERENCE",
+                    "relation": "showcase-derived-from-source",
+                    "immutable_evidence": (
+                        "https://api.github.com/repos/szl-holdings/szl-forge/commits/"
+                        + SOURCE_COMMIT
+                    ),
+                },
+                "build_metadata": {
+                    "built_at_semantics": (
+                        "Provider-reported lastModified for deployment.hf_revision; "
+                        "not a verified process start time or reproducible-build attestation."
+                    )
+                },
+                "claims": {
+                    "github_parity": "NOT_CLAIMED",
+                    "reproducible_build": "NOT_CLAIMED",
+                    "build_provenance": "NOT_CLAIMED",
+                },
+            },
+            "limits": [
+                "PENDING_GITHUB_SYNC means GitHub/Hugging Face content parity is not claimed.",
+                "deployment.hf_revision is repository-head evidence, not exact serving-process identity.",
+                "source.commit is a verified immutable reference, not proof of deployed byte equivalence.",
+                "This unsigned document does not establish SLSA provenance or binary reproducibility.",
+            ],
+        },
+        200,
+    )
+
+
+def build_health(status_provider: Callable[[], dict[str, Any]]) -> dict[str, Any]:
+    observed_at = _now_iso()
+    try:
+        packaged_status = status_provider()
+        packaged_contract = "PASS"
+        evidence_state = str(packaged_status.get("evidence_state", "UNKNOWN"))
+    except Exception as exc:
+        packaged_contract = "FAIL"
+        evidence_state = "UNAVAILABLE"
+        packaged_status = {"error": type(exc).__name__}
+    return {
+        "schema": "szl.forge.health/v1",
+        "observed_at": observed_at,
+        "status": "OK" if packaged_contract == "PASS" else "DEGRADED",
+        "transport_state": "REACHABLE",
+        "evidence_state": evidence_state,
+        "authority_state": "READ_ONLY",
+        "checks": {
+            "packaged_status_contract": packaged_contract,
+            "external_dependencies": "NOT_REQUIRED_FOR_TRANSPORT_HEALTH",
+            "model_quality": "NOT_MEASURED",
+            "source_parity": ALIGNMENT_STATE,
+        },
+        "limits": [
+            "HTTP health proves this process can answer a bounded read-only contract only.",
+            "It does not prove training quality, freshness, source parity, or promotion approval.",
+        ],
+        "packaged_status_schema": packaged_status.get("schema"),
+    }
+
+
+def build_runtime_status(
+    status_provider: Callable[[], dict[str, Any]], force: bool = False
+) -> dict[str, Any]:
+    packaged = status_provider()
+    measurement = measure_hf_repository_head(force=force)
+    return {
+        "schema": "szl.forge.runtime-status/v1",
+        "observed_at": _now_iso(),
+        "transport_state": "REACHABLE",
+        "evidence_state": packaged.get("evidence_state", "UNKNOWN"),
+        "authority_state": "READ_ONLY",
+        "interface_mode": packaged.get("interface_mode", "READ_ONLY"),
+        "deployment": {"hf_space": SPACE_ID, **measurement},
+        "source": {
+            "repository": SOURCE_REPOSITORY,
+            "commit": SOURCE_COMMIT,
+            "alignment_state": ALIGNMENT_STATE,
+            "github_parity": "NOT_CLAIMED",
+        },
+        "application": packaged,
+        "limits": [
+            "Runtime reachability, repository-head evidence, training state, and model quality are independent.",
+            "The nested application object is packaged snapshot evidence, not a live benchmark.",
+        ],
+    }
+
+
+def _json_response(payload: dict[str, Any], status_code: int = 200) -> JSONResponse:
+    return JSONResponse(
+        payload,
+        status_code=status_code,
+        headers={
+            "Cache-Control": "no-store",
+            "X-SZL-Transport-State": str(payload.get("transport_state", "REACHABLE")),
+            "X-SZL-Evidence-State": str(payload.get("evidence_state", "SNAPSHOT")),
+            "X-SZL-Authority-State": str(payload.get("authority_state", "READ_ONLY")),
+        },
+    )
+
+
+def register_contract_routes(
+    app: Any, status_provider: Callable[[], dict[str, Any]]
+) -> dict[str, Any]:
+    """Add exact GET routes ahead of Gradio's fallback after launch."""
+
+    async def source_attestation(refresh: int = 0):  # noqa: ANN202
+        payload, status_code = build_source_attestation(force=refresh == 1)
+        return _json_response(payload, status_code=status_code)
+
+    async def health():  # noqa: ANN202
+        return _json_response(build_health(status_provider))
+
+    async def runtime_status(refresh: int = 0):  # noqa: ANN202
+        return _json_response(build_runtime_status(status_provider, force=refresh == 1))
+
+    contracts = (
+        ("/.well-known/szl-source.json", source_attestation),
+        ("/healthz", health),
+        ("/api/status", runtime_status),
+    )
+    existing_paths = {getattr(route, "path", None) for route in app.router.routes}
+    collisions = [path for path, _ in contracts if path in existing_paths]
+    if collisions:
+        raise RuntimeError(f"Refusing to shadow existing routes: {collisions}")
+
+    before = list(app.router.routes)
+    for path, endpoint in contracts:
+        app.add_api_route(path, endpoint, methods=["GET"], include_in_schema=True)
+    added = list(app.router.routes[len(before) :])
+    app.router.routes[:] = added + before
+    app.openapi_schema = None
+    return {
+        "ok": True,
+        "routes": [path for path, _ in contracts],
+        "position": "BEFORE_GRADIO_FALLBACK",
+    }

tests/test_managed_gradio_contract.py

@@ -0,0 +1,86 @@
+import json
+import socket
+import time
+import unittest
+import urllib.request
+import warnings
+from pathlib import Path
+from unittest.mock import patch
+
+import app as forge_app
+import forge_runtime_contract as contract
+
+
+warnings.filterwarnings("ignore", category=ResourceWarning)
+
+
+MEASUREMENT = {
+    "hf_revision": "d" * 40,
+    "revision_state": "MEASURED_REPOSITORY_HEAD",
+    "running_process_revision_state": "NOT_EXPOSED_IN_PROCESS",
+    "measurement_method": "HUGGINGFACE_HUB_API",
+    "resolver": "https://huggingface.co/api/spaces/SZLHOLDINGS/szl-forge-lab",
+    "observed_at": "2026-07-12T00:00:00Z",
+    "provider_last_modified": "2026-07-12T00:00:00Z",
+}
+
+
+def _free_port() -> int:
+    with socket.socket() as sock:
+        sock.bind(("127.0.0.1", 0))
+        return int(sock.getsockname()[1])
+
+
+class ManagedGradioCompatibilityTests(unittest.TestCase):
+    def test_space_pins_the_locally_verified_managed_gradio_version(self):
+        readme = (Path(__file__).resolve().parents[1] / "README.md").read_text(encoding="utf-8")
+        self.assertIn("sdk: gradio\n", readme)
+        self.assertIn("sdk_version: 6.20.0\n", readme)
+
+    def test_public_launch_lifecycle_preserves_ui_config_and_exact_get_routes(self):
+        port = _free_port()
+        server_app, local_url, _ = forge_app.demo.launch(
+            server_name="127.0.0.1",
+            server_port=port,
+            prevent_thread_lock=True,
+            show_error=False,
+            ssr_mode=False,
+            css=forge_app.CSS,
+            theme=forge_app.gr.themes.Base(),
+            quiet=True,
+        )
+        try:
+            result = contract.register_contract_routes(server_app, forge_app.get_status)
+            self.assertTrue(result["ok"])
+            with patch.object(contract, "measure_hf_repository_head", return_value=MEASUREMENT):
+                expected = {
+                    "/": "text/html",
+                    "/config": "application/json",
+                    "/.well-known/szl-source.json": "application/json",
+                    "/healthz": "application/json",
+                    "/api/status": "application/json",
+                    "/gradio_api/openapi.json": "application/json",
+                }
+                for path, content_type in expected.items():
+                    response = None
+                    for _ in range(20):
+                        try:
+                            response = urllib.request.urlopen(
+                                local_url.rstrip("/") + path, timeout=3
+                            )
+                            break
+                        except OSError:
+                            time.sleep(0.05)
+                    self.assertIsNotNone(response, path)
+                    with response:
+                        self.assertEqual(200, response.status, path)
+                        self.assertTrue(response.headers["content-type"].startswith(content_type), path)
+                        body = response.read()
+                    if path in result["routes"]:
+                        self.assertIn("schema", json.loads(body))
+        finally:
+            forge_app.demo.close()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_runtime_contract.py

@@ -0,0 +1,122 @@
+import os
+import unittest
+from unittest.mock import patch
+
+from fastapi import FastAPI
+from fastapi.responses import HTMLResponse
+from fastapi.testclient import TestClient
+
+import forge_runtime_contract as contract
+from forge_lab import get_status
+
+
+MEASUREMENT = {
+    "hf_revision": "b" * 40,
+    "revision_state": "MEASURED_REPOSITORY_HEAD",
+    "running_process_revision_state": "NOT_EXPOSED_IN_PROCESS",
+    "measurement_method": "HUGGINGFACE_HUB_API",
+    "resolver": "https://huggingface.co/api/spaces/SZLHOLDINGS/szl-forge-lab",
+    "observed_at": "2026-07-12T00:00:00Z",
+    "provider_last_modified": "2026-07-12T00:00:00Z",
+}
+
+
+class RuntimeContractTests(unittest.TestCase):
+    def test_source_attestation_separates_head_evidence_from_process_identity(self):
+        with patch.object(contract, "measure_hf_repository_head", return_value=MEASUREMENT):
+            payload, status_code = contract.build_source_attestation()
+        self.assertEqual(200, status_code)
+        self.assertEqual("szl.deployment-source/v1", payload["schema"])
+        self.assertEqual("b" * 40, payload["deployment"]["hf_revision"])
+        self.assertEqual("PENDING_GITHUB_SYNC", payload["alignment_state"])
+        evidence = payload["extensions"]["deployment_revision_evidence"]
+        self.assertEqual("NOT_EXPOSED_IN_PROCESS", evidence["running_process_revision_state"])
+        self.assertIn(
+            "Provider-reported lastModified",
+            payload["extensions"]["build_metadata"]["built_at_semantics"],
+        )
+        self.assertEqual("NOT_CLAIMED", payload["extensions"]["claims"]["github_parity"])
+
+    def test_source_attestation_fails_closed_when_hub_measurement_is_unavailable(self):
+        unavailable = dict(MEASUREMENT, hf_revision=None, revision_state="UNAVAILABLE")
+        with patch.object(contract, "measure_hf_repository_head", return_value=unavailable):
+            payload, status_code = contract.build_source_attestation()
+        self.assertEqual(503, status_code)
+        self.assertEqual("szl.deployment-source-unavailable/v1", payload["schema"])
+        self.assertEqual("UNAVAILABLE", payload["evidence_state"])
+
+    def test_health_does_not_conflate_reachability_with_quality_or_parity(self):
+        payload = contract.build_health(get_status)
+        self.assertEqual("OK", payload["status"])
+        self.assertEqual("REACHABLE", payload["transport_state"])
+        self.assertEqual("NOT_MEASURED", payload["checks"]["model_quality"])
+        self.assertEqual("PENDING_GITHUB_SYNC", payload["checks"]["source_parity"])
+
+    def test_runtime_status_wraps_existing_packaged_contract(self):
+        with patch.object(contract, "measure_hf_repository_head", return_value=MEASUREMENT):
+            payload = contract.build_runtime_status(get_status)
+        self.assertEqual("szl.forge.runtime-status/v1", payload["schema"])
+        self.assertEqual("szl.forge.lab-status/v1", payload["application"]["schema"])
+        self.assertEqual("READ_ONLY", payload["authority_state"])
+        self.assertEqual("NOT_CLAIMED", payload["source"]["github_parity"])
+
+    def test_exact_routes_precede_html_fallback_and_return_json(self):
+        app = FastAPI()
+
+        @app.get("/{full_path:path}")
+        async def fallback(full_path: str):  # noqa: ARG001
+            return HTMLResponse("<html>fallback</html>")
+
+        result = contract.register_contract_routes(app, get_status)
+        self.assertTrue(result["ok"])
+        paths = [getattr(route, "path", None) for route in app.router.routes]
+        for path in result["routes"]:
+            self.assertLess(paths.index(path), paths.index("/{full_path:path}"))
+
+        with patch.object(contract, "measure_hf_repository_head", return_value=MEASUREMENT):
+            client = TestClient(app)
+            for path, schema in (
+                ("/.well-known/szl-source.json", "szl.deployment-source/v1"),
+                ("/healthz", "szl.forge.health/v1"),
+                ("/api/status", "szl.forge.runtime-status/v1"),
+            ):
+                response = client.get(path)
+                self.assertEqual(200, response.status_code)
+                self.assertTrue(response.headers["content-type"].startswith("application/json"))
+                self.assertEqual("no-store", response.headers["cache-control"])
+                self.assertEqual(schema, response.json()["schema"])
+
+    def test_registration_refuses_route_collisions(self):
+        app = FastAPI()
+
+        @app.get("/healthz")
+        async def existing_health():
+            return {"status": "existing"}
+
+        with self.assertRaisesRegex(RuntimeError, "Refusing to shadow"):
+            contract.register_contract_routes(app, get_status)
+
+    def test_space_id_is_built_from_documented_environment_variable(self):
+        class Response:
+            def __enter__(self):
+                return self
+
+            def __exit__(self, *_args):
+                return False
+
+            def read(self, *_args):
+                return (
+                    b'{"sha":"cccccccccccccccccccccccccccccccccccccccc",'
+                    b'"lastModified":"2026-07-12T00:00:00Z"}'
+                )
+
+        with patch.dict(os.environ, {"SPACE_ID": "SZLHOLDINGS/szl-forge-lab"}):
+            with patch("urllib.request.urlopen", return_value=Response()) as mocked:
+                measurement = contract.measure_hf_repository_head(force=True)
+        request = mocked.call_args.args[0]
+        self.assertIn("SZLHOLDINGS/szl-forge-lab", request.full_url)
+        self.assertEqual("c" * 40, measurement["hf_revision"])
+
+
+if __name__ == "__main__":
+    unittest.main()