chore(sync): mirror backend .py + Dockerfile to Space (hf-sync-backend)

Automated backend sync from szl-holdings/a11oy main via hf-sync-backend.
Updated (differed from the Space): Dockerfile, gdw_attention.py, gdw_proofs.py, gdw_telemetry.py, gdw_workspace.py, routers/gdw_frontier.py, serve.py
Deleted (gone from the repo + Dockerfile COPY set): (none)

Keeps the Space-built backend (serve.py + the Dockerfile-COPY'd .py
modules) identical to GitHub main so the Space never rebuilds from a
stale backend, new endpoints don't 404 there, and orphaned modules
removed from the repo don't linger in the Space tree.

Dockerfile

@@ -612,6 +612,7 @@ COPY szl_spend_cap.py ./szl_spend_cap.py
 # WAVE R Dev 1 — boot-resilience env/secret preflight. Per-file COPY (this
 # … (full rationale: docs/DOCKERFILE_NOTES.md §91)
 COPY a11oy_model_intel.py a11oy_experimental_tier.py a11oy_markets.py szl_agent_tts.py szl_gated_delta.py szl_blocksparse.py szl_retrieval_attn.py szl_model_harness.py szl_agent_loop_governed.py szl_crypto_pipeline.py szl_confattest.py szl_agent_operate.py szl_agentloop_brain.py szl_governed_rag.py szl_sovereign_flywheel.py szl_brain_corpus.py szl_verify_transcript.py szl_frontier_index.py szl_whatsnew.py szl_honestywall.py szl_brainmemory.py szl_agentos.py szl_brainground.py szl_brainuncertainty.py szl_brainhealth.py szl_brainwatch.py szl_boot_preflight.py szl_guarded_surface.py szl_status_aggregate.py szl_brainconstitution.py szl_brainagent.py szl_surface_manifests.py szl_source_attestation.py szl_compute_pool_contract.py szl_estateconstitution.py ./
+COPY gdw_attention.py gdw_workspace.py gdw_telemetry.py gdw_proofs.py ./
 COPY static/3d/surfaces/gateddelta.js static/3d/surfaces/blocksparse.js static/3d/surfaces/retrievalattn.js static/3d/surfaces/governedagent.js static/3d/surfaces/cryptopipeline.js static/3d/surfaces/confattest.js static/3d/surfaces/agentops.js static/3d/surfaces/frontierindex.js static/3d/surfaces/whatsnew.js static/3d/surfaces/opsdash.js ./static/3d/surfaces/
 
 # FORGE-FAMILY WALL (2026-07-14): /api/forge/family — server-side ed25519

gdw_attention.py

@@ -0,0 +1,141 @@
+"""Governed Delta Workspace attention-route policy and eager Torch dispatcher.
+
+The HTTP service uses the deterministic policy below. The optional Torch router is
+an executable research hook; it is not loaded unless a caller explicitly uses it.
+"""
+
+from dataclasses import dataclass
+from typing import Any, Callable, Dict, Optional
+
+
+ROUTE_NAMES = ("kda_local", "laguna_hybrid", "mla_global")
+
+
+@dataclass(frozen=True)
+class AttentionFeatures:
+    novelty: float
+    disagreement: float
+    risk: float
+    context_tokens: int
+    active_tool_count: int
+    memory_pressure: float
+
+
+def _clamp(value: float) -> float:
+    return max(0.0, min(1.0, float(value)))
+
+
+def choose_attention_mode(
+    features: AttentionFeatures,
+    mode_hint: str = "auto",
+) -> Dict[str, Any]:
+    """Choose a governed local, mixed, or global attention budget."""
+    if mode_hint in ROUTE_NAMES:
+        probabilities = {name: 1.0 if name == mode_hint else 0.0 for name in ROUTE_NAMES}
+        return {
+            "mode": mode_hint,
+            "score": None,
+            "probabilities": probabilities,
+            "reason": "explicit_validated_hint",
+        }
+
+    context = _clamp(features.context_tokens / 32768.0)
+    tools = _clamp(features.active_tool_count / 16.0)
+    memory = _clamp(features.memory_pressure)
+    score = _clamp(
+        0.20 * _clamp(features.novelty)
+        + 0.25 * _clamp(features.disagreement)
+        + 0.30 * _clamp(features.risk)
+        + 0.15 * context
+        + 0.10 * tools
+        - 0.25 * memory
+    )
+
+    if memory >= 0.85:
+        mode = "kda_local"
+        reason = "memory_pressure_guard"
+    elif score < 0.33:
+        mode = "kda_local"
+        reason = "bounded_local_budget"
+    elif score < 0.66:
+        mode = "laguna_hybrid"
+        reason = "mixed_local_global_budget"
+    else:
+        mode = "mla_global"
+        reason = "global_context_budget"
+
+    centers = (0.16, 0.50, 0.84)
+    weights = [max(0.0, 1.0 - abs(score - center) / 0.5) for center in centers]
+    total = sum(weights) or 1.0
+    probabilities = {
+        name: round(weight / total, 6)
+        for name, weight in zip(ROUTE_NAMES, weights)
+    }
+    return {
+        "mode": mode,
+        "score": round(score, 6),
+        "probabilities": probabilities,
+        "reason": reason,
+    }
+
+
+try:
+    import torch
+    from torch import nn
+except Exception:
+    torch = None
+    nn = None
+
+
+if nn is not None:
+
+    class HybridAttentionRouter(nn.Module):
+        """Trainable eager-mode KDA/Laguna/MLA route selector."""
+
+        def __init__(self, d_model: int):
+            super().__init__()
+            if d_model <= 0:
+                raise ValueError("d_model must be positive")
+            self.mode_proj = nn.Linear(d_model, 3)
+
+        def forward(self, summary_state):
+            logits = self.mode_proj(summary_state)
+            probabilities = torch.softmax(logits, dim=-1)
+            return probabilities.argmax(dim=-1), probabilities
+
+else:
+
+    class HybridAttentionRouter:
+        def __init__(self, d_model: int):
+            raise RuntimeError("Torch is required for HybridAttentionRouter")
+
+
+def hybrid_attention_dispatch(
+    mode,
+    x,
+    kda_fn: Callable,
+    mla_fn: Callable,
+    laguna_fn: Callable,
+):
+    """Batch samples by route before dispatch to reduce per-sample divergence."""
+    if torch is None:
+        raise RuntimeError("Torch is required for hybrid_attention_dispatch")
+    if mode.ndim != 1 or mode.shape[0] != x.shape[0]:
+        raise ValueError("mode must be a one-dimensional tensor matching batch size")
+
+    output: Optional[Any] = None
+    for route_id, route_fn in ((0, kda_fn), (1, mla_fn), (2, laguna_fn)):
+        indexes = torch.nonzero(mode == route_id, as_tuple=False).flatten()
+        if indexes.numel() == 0:
+            continue
+        routed = route_fn(x.index_select(0, indexes))
+        if output is None:
+            output = torch.empty(
+                (x.shape[0],) + tuple(routed.shape[1:]),
+                dtype=routed.dtype,
+                device=routed.device,
+            )
+        output.index_copy_(0, indexes, routed)
+    if output is None:
+        raise ValueError("mode contains no valid route ids")
+    return output

gdw_proofs.py

@@ -0,0 +1,77 @@
+"""Structured theorem-input export for asynchronous Lean checking."""
+
+import hashlib
+import json
+import os
+import tempfile
+from pathlib import Path
+from typing import Any, Dict
+
+
+def canonical_json(value: Any) -> str:
+    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
+
+
+def sha256_json(value: Any) -> str:
+    return hashlib.sha256(canonical_json(value).encode("utf-8")).hexdigest()
+
+
+def build_proof_payload(
+    proposal_id: str,
+    request_id: str,
+    step: int,
+    before_hash: str,
+    after_hash: str,
+    decision: str,
+    scheduler_mode: str,
+    receipt_hash: str,
+    dry_run: bool,
+) -> Dict[str, Any]:
+    mutates = decision == "ACCEPT" and not dry_run
+    payload = {
+        "schema": "szl.gdw.proof-input/v1",
+        "proposal_id": proposal_id,
+        "request_id": request_id,
+        "step_id": step,
+        "state_before_hash": before_hash,
+        "state_after_hash": after_hash,
+        "decision": decision,
+        "scheduler_mode": scheduler_mode,
+        "delta_update_receipt_hash": receipt_hash,
+        "invariants": {
+            "step_nonnegative": step >= 0,
+            "accepted_write_has_receipt": (not mutates) or bool(receipt_hash),
+            "non_mutating_preserves_state": mutates or before_hash == after_hash,
+            "scheduler_mode_valid": scheduler_mode
+            in {"kda_local", "laguna_hybrid", "mla_global"},
+        },
+        "formal_status": "NOT_RUN",
+    }
+    payload["payload_sha256"] = sha256_json(payload)
+    return payload
+
+
+def export_proof_payload(payload: Dict[str, Any]) -> Dict[str, Any]:
+    root = Path(os.environ.get("GDW_PROOF_DIR", "output/proofs")).resolve()
+    root.mkdir(parents=True, exist_ok=True)
+    proposal_id = payload["proposal_id"]
+    if not proposal_id or any(ch not in "0123456789abcdef" for ch in proposal_id):
+        raise ValueError("proposal_id is not a canonical lowercase hexadecimal id")
+    destination = root / f"{proposal_id}.json"
+    encoded = (json.dumps(payload, indent=2, sort_keys=True) + "\n").encode("utf-8")
+    handle, temporary = tempfile.mkstemp(prefix=".gdw-proof-", suffix=".tmp", dir=root)
+    try:
+        with os.fdopen(handle, "wb") as stream:
+            stream.write(encoded)
+            stream.flush()
+            os.fsync(stream.fileno())
+        os.replace(temporary, destination)
+    finally:
+        if os.path.exists(temporary):
+            os.unlink(temporary)
+    return {
+        "status": "INPUT_EXPORTED",
+        "path": str(destination),
+        "sha256": hashlib.sha256(encoded).hexdigest(),
+        "formal_status": "NOT_RUN",
+    }

gdw_telemetry.py

@@ -0,0 +1,83 @@
+"""In-process GDW counters and Prometheus text export."""
+
+import math
+import threading
+from collections import Counter, deque
+from typing import Any, Dict, Iterable, List
+
+
+class GDWTelemetry:
+    def __init__(self):
+        self._lock = threading.Lock()
+        self._requests = 0
+        self._errors = 0
+        self._receipts = 0
+        self._decisions = Counter()
+        self._routes = Counter()
+        self._latencies_ms = deque(maxlen=10000)
+
+    def observe(
+        self,
+        latency_ms: float,
+        decision: str,
+        route: str,
+        receipt_emitted: bool,
+        error: bool = False,
+    ) -> None:
+        with self._lock:
+            self._requests += 1
+            self._errors += int(error)
+            self._receipts += int(receipt_emitted)
+            self._decisions[decision] += 1
+            self._routes[route] += 1
+            self._latencies_ms.append(float(latency_ms))
+
+    @staticmethod
+    def _percentile(values: Iterable[float], quantile: float) -> float:
+        ordered = sorted(values)
+        if not ordered:
+            return 0.0
+        position = (len(ordered) - 1) * quantile
+        lower = math.floor(position)
+        upper = math.ceil(position)
+        if lower == upper:
+            return ordered[lower]
+        return ordered[lower] + (ordered[upper] - ordered[lower]) * (position - lower)
+
+    def snapshot(self) -> Dict[str, Any]:
+        with self._lock:
+            latencies: List[float] = list(self._latencies_ms)
+            return {
+                "requests": self._requests,
+                "errors": self._errors,
+                "receipts": self._receipts,
+                "decisions": dict(self._decisions),
+                "routes": dict(self._routes),
+                "p50_ms": round(self._percentile(latencies, 0.50), 6),
+                "p95_ms": round(self._percentile(latencies, 0.95), 6),
+                "p99_ms": round(self._percentile(latencies, 0.99), 6),
+            }
+
+    def render(self) -> str:
+        snapshot = self.snapshot()
+        lines = [
+            "# HELP gdw_requests_total Governed Delta Workspace requests.",
+            "# TYPE gdw_requests_total counter",
+            f"gdw_requests_total {snapshot['requests']}",
+            "# HELP gdw_errors_total Governed Delta Workspace errors.",
+            "# TYPE gdw_errors_total counter",
+            f"gdw_errors_total {snapshot['errors']}",
+            "# HELP gdw_receipts_total Persisted transition receipts.",
+            "# TYPE gdw_receipts_total counter",
+            f"gdw_receipts_total {snapshot['receipts']}",
+        ]
+        for decision, value in sorted(snapshot["decisions"].items()):
+            lines.append(f'gdw_decisions_total{{decision="{decision}"}} {value}')
+        for route, value in sorted(snapshot["routes"].items()):
+            lines.append(f'gdw_routes_total{{route="{route}"}} {value}')
+        for quantile in ("p50", "p95", "p99"):
+            lines.append(
+                f'gdw_latency_milliseconds{{quantile="{quantile}"}} '
+                f"{snapshot[quantile + '_ms']}"
+            )
+        return "\n".join(lines) + "\n"

gdw_workspace.py

@@ -0,0 +1,336 @@
+"""Concurrency-safe SQLite state, idempotency, and receipt storage for GDW."""
+
+import contextlib
+import json
+import os
+import sqlite3
+import threading
+from pathlib import Path
+from typing import Any, Dict, Iterator, Optional, Tuple
+
+
+_SCHEMA_LOCK = threading.RLock()
+_PROCESS_WRITE_LOCK = threading.RLock()
+_INITIALISED_PATHS = set()
+
+
+class GDWWorkspace:
+    def __init__(self, path: Optional[str] = None):
+        configured = path or os.environ.get("GDW_DB_PATH", "output/gdw/gdw.sqlite3")
+        self.path = Path(configured).resolve()
+        self._initialise()
+
+    def _connect(self) -> sqlite3.Connection:
+        connection = sqlite3.connect(
+            str(self.path),
+            timeout=30.0,
+            isolation_level=None,
+            check_same_thread=False,
+        )
+        connection.row_factory = sqlite3.Row
+        connection.execute("PRAGMA busy_timeout=30000")
+        connection.execute("PRAGMA synchronous=NORMAL")
+        connection.execute("PRAGMA foreign_keys=ON")
+        return connection
+
+    def _initialise(self) -> None:
+        key = str(self.path)
+        if key in _INITIALISED_PATHS:
+            return
+        with _SCHEMA_LOCK:
+            if key in _INITIALISED_PATHS:
+                return
+            self.path.parent.mkdir(parents=True, exist_ok=True)
+            connection = sqlite3.connect(str(self.path), timeout=30.0)
+            try:
+                connection.execute("PRAGMA journal_mode=WAL")
+                connection.execute("PRAGMA synchronous=NORMAL")
+                connection.execute("PRAGMA foreign_keys=ON")
+                self._ensure_schema(connection)
+                connection.commit()
+                _INITIALISED_PATHS.add(key)
+            finally:
+                connection.close()
+
+    @staticmethod
+    def _ensure_schema(connection: sqlite3.Connection) -> None:
+        connection.executescript(
+            """
+            CREATE TABLE IF NOT EXISTS session_state (
+                session_id TEXT PRIMARY KEY,
+                step INTEGER NOT NULL CHECK(step >= 0),
+                state_json TEXT NOT NULL,
+                state_hash TEXT NOT NULL,
+                updated_at TEXT NOT NULL
+            );
+            CREATE TABLE IF NOT EXISTS requests (
+                request_id TEXT PRIMARY KEY,
+                request_digest TEXT NOT NULL,
+                session_id TEXT NOT NULL,
+                response_json TEXT NOT NULL,
+                response_hash TEXT NOT NULL,
+                created_at TEXT NOT NULL
+            );
+            CREATE TABLE IF NOT EXISTS receipts (
+                receipt_hash TEXT PRIMARY KEY,
+                request_id TEXT NOT NULL UNIQUE,
+                session_id TEXT NOT NULL,
+                step INTEGER NOT NULL,
+                receipt_json TEXT NOT NULL,
+                created_at TEXT NOT NULL,
+                FOREIGN KEY(request_id) REFERENCES requests(request_id)
+                    DEFERRABLE INITIALLY DEFERRED
+            );
+            CREATE TABLE IF NOT EXISTS proof_outbox (
+                proposal_id TEXT PRIMARY KEY,
+                payload_json TEXT NOT NULL,
+                payload_sha256 TEXT NOT NULL,
+                status TEXT NOT NULL CHECK(status IN ('PENDING', 'EXPORTED')),
+                artifact_json TEXT,
+                created_at TEXT NOT NULL,
+                exported_at TEXT
+            );
+            CREATE INDEX IF NOT EXISTS idx_requests_session ON requests(session_id);
+            CREATE INDEX IF NOT EXISTS idx_receipts_session ON receipts(session_id, step);
+            CREATE INDEX IF NOT EXISTS idx_proof_outbox_status
+                ON proof_outbox(status, created_at);
+            """
+        )
+
+    @contextlib.contextmanager
+    def transaction(self) -> Iterator[sqlite3.Connection]:
+        with _PROCESS_WRITE_LOCK:
+            connection = self._connect()
+            try:
+                connection.execute("BEGIN IMMEDIATE")
+                yield connection
+                connection.execute("COMMIT")
+            except Exception:
+                connection.execute("ROLLBACK")
+                raise
+            finally:
+                connection.close()
+
+    @staticmethod
+    def cached_request(
+        connection: sqlite3.Connection,
+        request_id: str,
+    ) -> Optional[Tuple[str, Dict[str, Any]]]:
+        row = connection.execute(
+            "SELECT request_digest, response_json FROM requests WHERE request_id = ?",
+            (request_id,),
+        ).fetchone()
+        if row is None:
+            return None
+        return row["request_digest"], json.loads(row["response_json"])
+
+    @staticmethod
+    def session_state(
+        connection: sqlite3.Connection,
+        session_id: str,
+    ) -> Optional[Dict[str, Any]]:
+        row = connection.execute(
+            "SELECT step, state_json, state_hash, updated_at "
+            "FROM session_state WHERE session_id = ?",
+            (session_id,),
+        ).fetchone()
+        if row is None:
+            return None
+        return {
+            "session_id": session_id,
+            "step": int(row["step"]),
+            "state": json.loads(row["state_json"]),
+            "state_hash": row["state_hash"],
+            "updated_at": row["updated_at"],
+        }
+
+    @staticmethod
+    def save_state(
+        connection: sqlite3.Connection,
+        session_id: str,
+        step: int,
+        state: Dict[str, Any],
+        state_hash: str,
+        updated_at: str,
+    ) -> None:
+        connection.execute(
+            """
+            INSERT INTO session_state(session_id, step, state_json, state_hash, updated_at)
+            VALUES (?, ?, ?, ?, ?)
+            ON CONFLICT(session_id) DO UPDATE SET
+                step = excluded.step,
+                state_json = excluded.state_json,
+                state_hash = excluded.state_hash,
+                updated_at = excluded.updated_at
+            """,
+            (
+                session_id,
+                step,
+                json.dumps(state, sort_keys=True, separators=(",", ":")),
+                state_hash,
+                updated_at,
+            ),
+        )
+
+    @staticmethod
+    def save_request(
+        connection: sqlite3.Connection,
+        request_id: str,
+        request_digest: str,
+        session_id: str,
+        response: Dict[str, Any],
+        response_hash: str,
+        created_at: str,
+    ) -> None:
+        connection.execute(
+            """
+            INSERT INTO requests(
+                request_id, request_digest, session_id, response_json,
+                response_hash, created_at
+            ) VALUES (?, ?, ?, ?, ?, ?)
+            """,
+            (
+                request_id,
+                request_digest,
+                session_id,
+                json.dumps(response, sort_keys=True, separators=(",", ":")),
+                response_hash,
+                created_at,
+            ),
+        )
+
+    @staticmethod
+    def save_receipt(
+        connection: sqlite3.Connection,
+        receipt_hash: str,
+        request_id: str,
+        session_id: str,
+        step: int,
+        receipt: Dict[str, Any],
+        created_at: str,
+    ) -> None:
+        connection.execute(
+            """
+            INSERT INTO receipts(
+                receipt_hash, request_id, session_id, step, receipt_json, created_at
+            ) VALUES (?, ?, ?, ?, ?, ?)
+            """,
+            (
+                receipt_hash,
+                request_id,
+                session_id,
+                step,
+                json.dumps(receipt, sort_keys=True, separators=(",", ":")),
+                created_at,
+            ),
+        )
+
+    @staticmethod
+    def save_proof_outbox(
+        connection: sqlite3.Connection,
+        proposal_id: str,
+        payload: Dict[str, Any],
+        payload_sha256: str,
+        created_at: str,
+    ) -> None:
+        connection.execute(
+            """
+            INSERT INTO proof_outbox(
+                proposal_id, payload_json, payload_sha256, status, created_at
+            ) VALUES (?, ?, ?, 'PENDING', ?)
+            """,
+            (
+                proposal_id,
+                json.dumps(payload, sort_keys=True, separators=(",", ":")),
+                payload_sha256,
+                created_at,
+            ),
+        )
+
+    def pending_proofs(self, limit: int = 100) -> list:
+        bounded = max(1, min(int(limit), 10000))
+        connection = self._connect()
+        try:
+            rows = connection.execute(
+                """
+                SELECT proposal_id, payload_json, payload_sha256
+                FROM proof_outbox
+                WHERE status = 'PENDING'
+                ORDER BY created_at, proposal_id
+                LIMIT ?
+                """,
+                (bounded,),
+            ).fetchall()
+            return [
+                {
+                    "proposal_id": row["proposal_id"],
+                    "payload": json.loads(row["payload_json"]),
+                    "payload_sha256": row["payload_sha256"],
+                }
+                for row in rows
+            ]
+        finally:
+            connection.close()
+
+    def mark_proof_exported(
+        self,
+        proposal_id: str,
+        artifact: Dict[str, Any],
+        exported_at: str,
+    ) -> None:
+        with self.transaction() as connection:
+            connection.execute(
+                """
+                UPDATE proof_outbox
+                SET status = 'EXPORTED', artifact_json = ?, exported_at = ?
+                WHERE proposal_id = ? AND status = 'PENDING'
+                """,
+                (
+                    json.dumps(artifact, sort_keys=True, separators=(",", ":")),
+                    exported_at,
+                    proposal_id,
+                ),
+            )
+
+    def read_session(self, session_id: str) -> Optional[Dict[str, Any]]:
+        connection = self._connect()
+        try:
+            return self.session_state(connection, session_id)
+        finally:
+            connection.close()
+
+    def integrity(self) -> Dict[str, Any]:
+        connection = self._connect()
+        try:
+            check = connection.execute("PRAGMA integrity_check").fetchone()[0]
+            counts = {}
+            for table in ("session_state", "requests", "receipts", "proof_outbox"):
+                counts[table] = int(
+                    connection.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
+                )
+            pending_proofs = int(
+                connection.execute(
+                    "SELECT COUNT(*) FROM proof_outbox WHERE status = 'PENDING'"
+                ).fetchone()[0]
+            )
+            orphan_receipts = int(
+                connection.execute(
+                    """
+                    SELECT COUNT(*) FROM receipts r
+                    LEFT JOIN requests q ON q.request_id = r.request_id
+                    WHERE q.request_id IS NULL
+                    """
+                ).fetchone()[0]
+            )
+            return {
+                "ok": check == "ok" and orphan_receipts == 0,
+                "sqlite_integrity": check,
+                "orphan_receipts": orphan_receipts,
+                "pending_proofs": pending_proofs,
+                "counts": counts,
+                "path": str(self.path),
+                "wal": True,
+                "synchronous": "NORMAL",
+            }
+        finally:
+            connection.close()

routers/gdw_frontier.py

@@ -0,0 +1,399 @@
+"""Authenticated Governed Delta Workspace API and benchmark surfaces."""
+
+import hashlib
+import hmac
+import json
+import os
+import re
+import time
+from datetime import datetime, timezone
+from typing import List, Literal, Optional
+
+from fastapi import Header, HTTPException
+from fastapi.responses import PlainTextResponse
+from pydantic import BaseModel, Field
+
+from gdw_attention import AttentionFeatures, choose_attention_mode
+from gdw_proofs import build_proof_payload, export_proof_payload, sha256_json
+from gdw_telemetry import GDWTelemetry
+from gdw_workspace import GDWWorkspace
+from szl_receipt_substrate import append_receipt
+from szl_sgh_scheduler import build_plan
+
+
+_TELEMETRY = GDWTelemetry()
+_ID_PATTERN = re.compile(r"^[A-Za-z0-9._:-]{1,128}$")
+_EXPERTS = {"planner", "retriever", "auditor", "verifier", "operator"}
+
+
+class GDWStepRequest(BaseModel):
+    session_id: str = Field(min_length=1, max_length=128)
+    request: str = Field(min_length=1, max_length=4096)
+    allowed_experts: List[str] = Field(default_factory=list)
+    risk_budget: float = Field(default=0.35, ge=0.0, le=1.0)
+    mode_hint: Literal[
+        "auto", "kda_local", "laguna_hybrid", "mla_global"
+    ] = "auto"
+    dry_run: bool = False
+    novelty: Optional[float] = Field(default=None, ge=0.0, le=1.0)
+    disagreement: Optional[float] = Field(default=None, ge=0.0, le=1.0)
+    context_tokens: int = Field(default=0, ge=0, le=1000000)
+    active_tool_count: int = Field(default=0, ge=0, le=64)
+    memory_pressure: Optional[float] = Field(default=None, ge=0.0, le=1.0)
+
+    class Config:
+        extra = "forbid"
+
+
+def _dump_model(model):
+    if hasattr(model, "model_dump"):
+        return model.model_dump(mode="json")
+    return model.dict()
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _sha(value) -> str:
+    encoded = json.dumps(
+        value, sort_keys=True, separators=(",", ":"), ensure_ascii=True
+    ).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
+def _authorise(authorization: Optional[str]) -> None:
+    token = os.environ.get("GDW_AUTH_TOKEN")
+    if not token:
+        raise HTTPException(
+            status_code=503,
+            detail="GDW_AUTH_TOKEN is not configured; write surface is unavailable",
+        )
+    supplied = authorization or ""
+    expected = "Bearer " + token
+    if not hmac.compare_digest(supplied, expected):
+        raise HTTPException(status_code=401, detail="invalid bearer token")
+
+
+def _validate_identifiers(payload: GDWStepRequest, request_id: Optional[str]) -> str:
+    if not request_id or not _ID_PATTERN.fullmatch(request_id):
+        raise HTTPException(
+            status_code=422,
+            detail="X-Request-Id must be 1-128 canonical identifier characters",
+        )
+    if not _ID_PATTERN.fullmatch(payload.session_id):
+        raise HTTPException(status_code=422, detail="invalid session_id")
+    if len(payload.allowed_experts) > 16:
+        raise HTTPException(status_code=422, detail="too many allowed_experts")
+    return request_id
+
+
+def _decision(payload: GDWStepRequest) -> str:
+    experts = set(payload.allowed_experts)
+    if not experts or not experts.issubset(_EXPERTS):
+        return "QUARANTINE"
+    if payload.risk_budget >= 0.90:
+        return "REJECT"
+    if payload.risk_budget >= 0.75:
+        return "QUARANTINE"
+    return "ACCEPT"
+
+
+def register(app, ns: str = "a11oy"):
+    prefix = f"/api/{ns}/v1/gdw"
+
+    @app.get(prefix + "/healthz")
+    @app.get("/v1/gdw/healthz")
+    def gdw_healthz():
+        return {
+            "service": "gdw-frontier",
+            "status": "REAL",
+            "write_ready": bool(os.environ.get("GDW_AUTH_TOKEN")),
+            "persistence": "SQLITE_WAL",
+            "benchmark_claim": "UNMEASURED",
+        }
+
+    @app.get(prefix + "/bench/meta")
+    @app.get("/v1/gdw/bench/meta")
+    def gdw_bench_meta(
+        authorization: Optional[str] = Header(default=None, alias="Authorization"),
+    ):
+        _authorise(authorization)
+        return {
+            "service": "gdw-frontier",
+            "implementation_status": "REAL",
+            "benchmark_status": "UNMEASURED",
+            "recommended_burst": 10000,
+            "metrics_path": prefix + "/metrics",
+            "notes": [
+                "A 10k-request run is harness evidence, not a production guarantee.",
+                "Use p95, p99, error rate, receipt integrity, and SQLite integrity.",
+            ],
+        }
+
+    @app.get(prefix + "/metrics", response_class=PlainTextResponse)
+    @app.get("/v1/gdw/metrics", response_class=PlainTextResponse)
+    def gdw_metrics(
+        authorization: Optional[str] = Header(default=None, alias="Authorization"),
+    ):
+        _authorise(authorization)
+        return PlainTextResponse(
+            _TELEMETRY.render(),
+            media_type="text/plain; version=0.0.4; charset=utf-8",
+        )
+
+    @app.get(prefix + "/integrity")
+    @app.get("/v1/gdw/integrity")
+    def gdw_integrity(
+        authorization: Optional[str] = Header(default=None, alias="Authorization"),
+    ):
+        _authorise(authorization)
+        return GDWWorkspace().integrity()
+
+    @app.get(prefix + "/sessions/{session_id}")
+    @app.get("/v1/gdw/sessions/{session_id}")
+    def gdw_session(
+        session_id: str,
+        authorization: Optional[str] = Header(default=None, alias="Authorization"),
+    ):
+        _authorise(authorization)
+        if not _ID_PATTERN.fullmatch(session_id):
+            raise HTTPException(status_code=422, detail="invalid session_id")
+        state = GDWWorkspace().read_session(session_id)
+        if state is None:
+            raise HTTPException(status_code=404, detail="session not found")
+        return state
+
+    @app.post(prefix + "/step")
+    @app.post("/v1/gdw/step")
+    def gdw_step(
+        payload: GDWStepRequest,
+        authorization: Optional[str] = Header(default=None, alias="Authorization"),
+        x_request_id: Optional[str] = Header(default=None, alias="X-Request-Id"),
+    ):
+        started = time.perf_counter()
+        _authorise(authorization)
+        request_id = _validate_identifiers(payload, x_request_id)
+        payload_data = _dump_model(payload)
+        request_digest = _sha(payload_data)
+        workspace = GDWWorkspace()
+        selected_mode = "unresolved"
+        decision = "ERROR"
+        receipt_hash = ""
+
+        try:
+            with workspace.transaction() as connection:
+                cached = workspace.cached_request(connection, request_id)
+                if cached is not None:
+                    cached_digest, cached_response = cached
+                    if cached_digest != request_digest:
+                        raise HTTPException(
+                            status_code=409,
+                            detail="X-Request-Id was already used with different content",
+                        )
+                    cached_response["replayed"] = True
+                    selected_mode = cached_response["scheduler_mode"]
+                    decision = cached_response["decision"]
+                    receipt_hash = cached_response.get("receipt_hash") or ""
+                    _TELEMETRY.observe(
+                        (time.perf_counter() - started) * 1000.0,
+                        decision,
+                        selected_mode,
+                        bool(receipt_hash),
+                    )
+                    return cached_response
+
+                previous = workspace.session_state(connection, payload.session_id)
+                if previous is None:
+                    before_step = 0
+                    before_hash = _sha(
+                        {
+                            "session_id": payload.session_id,
+                            "step": 0,
+                            "state": "GENESIS",
+                        }
+                    )
+                else:
+                    before_step = previous["step"]
+                    before_hash = previous["state_hash"]
+
+                features = AttentionFeatures(
+                    novelty=payload.novelty
+                    if payload.novelty is not None
+                    else min(1.0, len(payload.request) / 1024.0),
+                    disagreement=payload.disagreement
+                    if payload.disagreement is not None
+                    else min(1.0, max(0, len(set(payload.allowed_experts)) - 1) / 5.0),
+                    risk=payload.risk_budget,
+                    context_tokens=payload.context_tokens
+                    or max(1, len(payload.request) // 4),
+                    active_tool_count=payload.active_tool_count,
+                    memory_pressure=payload.memory_pressure or 0.0,
+                )
+                routing = choose_attention_mode(features, payload.mode_hint)
+                selected_mode = routing["mode"]
+                decision = _decision(payload)
+                mutates = decision == "ACCEPT" and not payload.dry_run
+                step = before_step + 1 if mutates else before_step
+                proposal_id = hashlib.sha256(
+                    (request_id + ":" + request_digest).encode("utf-8")
+                ).hexdigest()[:32]
+                timestamp = _now()
+
+                if mutates:
+                    state = {
+                        "session_id": payload.session_id,
+                        "step": step,
+                        "previous_state_hash": before_hash,
+                        "request_digest": request_digest,
+                        "scheduler_mode": selected_mode,
+                        "allowed_experts": sorted(set(payload.allowed_experts)),
+                    }
+                    after_hash = _sha(state)
+                    workspace.save_state(
+                        connection,
+                        payload.session_id,
+                        step,
+                        state,
+                        after_hash,
+                        timestamp,
+                    )
+                    receipt = append_receipt(
+                        actor_id="gdw-frontier",
+                        tool_name="gdw.step",
+                        payload={
+                            "proposal_id": proposal_id,
+                            "request_id": request_id,
+                            "session_id": payload.session_id,
+                            "step": step,
+                            "state_before_hash": before_hash,
+                            "state_after_hash": after_hash,
+                            "scheduler_mode": selected_mode,
+                        },
+                    )
+                    receipt_hash = receipt.get("receipt_hash") or sha256_json(receipt)
+                else:
+                    state = previous["state"] if previous else {"state": "GENESIS"}
+                    after_hash = before_hash
+                    receipt = None
+
+                proof_payload = build_proof_payload(
+                    proposal_id=proposal_id,
+                    request_id=request_id,
+                    step=step,
+                    before_hash=before_hash,
+                    after_hash=after_hash,
+                    decision=decision,
+                    scheduler_mode=selected_mode,
+                    receipt_hash=receipt_hash,
+                    dry_run=payload.dry_run,
+                )
+                proof_mode = os.environ.get(
+                    "GDW_PROOF_EXPORT_MODE", "sync"
+                ).strip().lower()
+                if proof_mode == "outbox":
+                    workspace.save_proof_outbox(
+                        connection,
+                        proposal_id,
+                        proof_payload,
+                        proof_payload["payload_sha256"],
+                        timestamp,
+                    )
+                    proof_artifact = {
+                        "status": "OUTBOX_PERSISTED",
+                        "payload_sha256": proof_payload["payload_sha256"],
+                        "formal_status": "NOT_RUN",
+                    }
+                elif proof_mode == "sync":
+                    proof_artifact = export_proof_payload(proof_payload)
+                else:
+                    raise ValueError(
+                        "GDW_PROOF_EXPORT_MODE must be 'sync' or 'outbox'"
+                    )
+                plan = build_plan(
+                    tasks=("governance", "attention_route", "state_transition", "verify"),
+                    meta={"proposal_id": proposal_id},
+                )
+                response = {
+                    "service": "gdw-frontier",
+                    "implementation_status": "REAL",
+                    "benchmark_status": "UNMEASURED",
+                    "proposal_id": proposal_id,
+                    "request_id": request_id,
+                    "session_id": payload.session_id,
+                    "decision": decision,
+                    "step": step,
+                    "state_hash": after_hash,
+                    "state_before_hash": before_hash,
+                    "receipt_hash": receipt_hash or None,
+                    "scheduler_mode": selected_mode,
+                    "routing": routing,
+                    "kernel_execution": "NOT_EXECUTED_BY_CONTROL_API",
+                    "dry_run": payload.dry_run,
+                    "replayed": False,
+                    "audit": {
+                        "governance": "DENY_BY_DEFAULT",
+                        "allowed_experts": sorted(set(payload.allowed_experts)),
+                        "plan_version": plan["plan_version"],
+                        "receipt_substrate": "szl_receipt_substrate"
+                        if receipt is not None
+                        else None,
+                        "proof_export_mode": proof_mode,
+                    },
+                    "proof": proof_artifact,
+                }
+                response_hash = _sha(response)
+                workspace.save_request(
+                    connection,
+                    request_id,
+                    request_digest,
+                    payload.session_id,
+                    response,
+                    response_hash,
+                    timestamp,
+                )
+                if receipt is not None:
+                    workspace.save_receipt(
+                        connection,
+                        receipt_hash,
+                        request_id,
+                        payload.session_id,
+                        step,
+                        receipt,
+                        timestamp,
+                    )
+
+            _TELEMETRY.observe(
+                (time.perf_counter() - started) * 1000.0,
+                decision,
+                selected_mode,
+                bool(receipt_hash),
+            )
+            return response
+        except HTTPException:
+            raise
+        except Exception as exc:
+            _TELEMETRY.observe(
+                (time.perf_counter() - started) * 1000.0,
+                "ERROR",
+                selected_mode,
+                False,
+                error=True,
+            )
+            raise HTTPException(
+                status_code=500,
+                detail=f"GDW transition failed closed: {type(exc).__name__}",
+            ) from exc
+
+    return {
+        "ok": True,
+        "state": "REAL",
+        "routes": [
+            prefix + "/healthz",
+            prefix + "/bench/meta",
+            prefix + "/metrics",
+            prefix + "/integrity",
+            prefix + "/sessions/{session_id}",
+            prefix + "/step",
+        ],
+    }

serve.py

@@ -550,6 +550,24 @@ async def quantum_utility_receipt_replay(request: Request) -> JSONResponse:
         )
 
 
+# Governed Delta Workspace (wave 28): authenticated, deny-by-default state
+# transitions over SQLite WAL, idempotency keys, the shared receipt substrate,
+# structured theorem inputs, and honest benchmark metadata. Runtime route
+# decisions are REAL; benchmark claims remain UNMEASURED until a captured run.
+# Registered before both catch-alls.
+try:
+    from routers import gdw_frontier as _gdw_frontier
+
+    _GDW_FRONTIER_STATUS = _gdw_frontier.register(app, ns="a11oy")
+except Exception as _gdw_frontier_error:  # pragma: no cover - fail one surface closed
+    _GDW_FRONTIER_STATUS = {
+        "ok": False,
+        "state": "UNAVAILABLE",
+        "reason": type(_gdw_frontier_error).__name__,
+        "routes": [],
+    }
+
+
 # External numerical-engine frontier (wave 18). The module accepts only fixed
 # matrix solve, symmetric-eigenvalue, and reference-vector operations. GNU
 # Octave and MATLAB remain external installations; missing engine, offline