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_auth.py, gdw_drain.py, gdw_proofs.py, gdw_runtime.py, gdw_workspace.py, routers/gdw_frontier.py, serve.py, szl_colang_policy.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

@@ -521,7 +521,7 @@ COPY static/shared/szl_label_engine.js static/shared/szl_receipt_cosign.js stati
 
 # --- GOVERNANCE / EVAL / CALIBRATION layer (Dev B, 2026-06): ADDITIVE ---
 # … (full rationale: docs/DOCKERFILE_NOTES.md §67)
-COPY policy/colang/roe_core.co policy/colang/killinchu_threat.co policy/colang/gdw_enforcement_contract.json ./policy/colang/
+COPY policy/colang/roe_core.co policy/colang/killinchu_threat.co policy/colang/gdw_enforcement_contract.json policy/colang/enforcement-contract.json ./policy/colang/
 # GOVERNED AUTO-REVIEW (Integration I2) — keystone autonomy layer: governed +
 # … (full rationale: docs/DOCKERFILE_NOTES.md §68)
 COPY scripts/check_tau_eval.py ./scripts/check_tau_eval.py
@@ -612,7 +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 gdw_drain.py ./
+COPY gdw_attention.py gdw_auth.py gdw_workspace.py gdw_telemetry.py gdw_proofs.py gdw_runtime.py gdw_drain.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
@@ -654,7 +654,7 @@ ENV SZL_GIT_SHA=${SZL_GIT_SHA} \
 # The Second Brain's SQLite index is rebuildable, but a mounted /app/data keeps
 # … (full rationale: docs/DOCKERFILE_NOTES.md §97)
 VOLUME ["/app/data"]
-CMD ["python", "serve.py"]
+CMD ["python", "gdw_runtime.py"]
 
 
 # Build cache-bust 2026-06-05T00:00Z (Orchestrator Squad):

gdw_auth.py

@@ -0,0 +1,527 @@
+"""Stable principal authentication for the Governed Delta Workspace.
+
+This services-layer module accepts a secret-managed JSON registry, reduces raw
+bearer tokens to fixed-length digests during parsing, accepts exact pre-hashed
+token bindings, and returns immutable principal identities. Raw tokens are
+never retained by the resulting registry or included in errors and
+representations.
+"""
+
+import hashlib
+import hmac
+import json
+import re
+from dataclasses import dataclass
+from typing import Any, FrozenSet, Iterable, Optional, Tuple, Union
+
+
+_IDENTIFIER_PATTERN = re.compile(r"^[a-z0-9][a-z0-9._:-]{0,127}$")
+_REGISTRY_KEYS = frozenset({"version", "credentials"})
+_CREDENTIAL_KEYS = frozenset(
+    {
+        "owner_id",
+        "namespace",
+        "key_id",
+        "token",
+        "token_sha256",
+        "scopes",
+        "revoked",
+    }
+)
+_REQUIRED_CREDENTIAL_KEYS = frozenset(
+    {"owner_id", "namespace", "key_id", "scopes"}
+)
+_TOKEN_KEYS = frozenset({"token", "token_sha256"})
+_LEGACY_PRINCIPAL_KEYS = frozenset({"token_sha256", "roles"})
+_LEGACY_ROLE_SCOPES = {
+    "user": frozenset({"session:read", "step:write"}),
+    "admin": frozenset(
+        {
+            "bench:read",
+            "integrity:global",
+            "integrity:read",
+            "metrics:read",
+            "session:read",
+            "step:write",
+        }
+    ),
+}
+_SAFE_AUTH_MESSAGES = {
+    "missing_authorization": "bearer authorization is required",
+    "invalid_authorization": "bearer authorization is invalid",
+    "invalid_bearer_token": "bearer credential is invalid",
+    "credential_revoked": "bearer credential is revoked",
+    "foreign_namespace": "bearer credential is not valid for this namespace",
+    "missing_scopes": "bearer credential lacks required scopes",
+}
+
+
+class AuthConfigurationError(ValueError):
+    """Raised when authentication configuration is absent or malformed."""
+
+
+class AuthenticationError(ValueError):
+    """A token-safe authentication failure with a stable machine code."""
+
+    def __init__(self, code: str):
+        if code not in _SAFE_AUTH_MESSAGES:
+            raise ValueError("unknown authentication error code")
+        self.code = code
+        super().__init__(_SAFE_AUTH_MESSAGES[code])
+
+
+@dataclass(frozen=True, slots=True)
+class Principal:
+    """Stable caller identity, independent of the active credential key."""
+
+    owner_id: str
+    namespace: str
+    key_id: str
+    scopes: Tuple[str, ...]
+
+
+@dataclass(frozen=True, slots=True, repr=False)
+class _Credential:
+    owner_id: str
+    namespace: str
+    key_id: str
+    token_digest: bytes
+    scopes: FrozenSet[str]
+    revoked: bool
+
+
+class CredentialRegistry:
+    """Immutable, token-redacted credential registry."""
+
+    __slots__ = ("_credentials",)
+
+    def __init__(self, credentials: Iterable[_Credential]):
+        values = tuple(credentials)
+        if not values:
+            raise AuthConfigurationError(
+                "credential registry must contain at least one credential"
+            )
+        self._credentials = values
+
+    @property
+    def credential_count(self) -> int:
+        return len(self._credentials)
+
+    def __repr__(self) -> str:
+        return f"CredentialRegistry(credential_count={self.credential_count})"
+
+    def authenticate(
+        self,
+        authorization: Optional[str],
+        *,
+        namespace: str,
+        required_scopes: Iterable[str] = (),
+    ) -> Principal:
+        """Authenticate a bearer header without short-circuiting credential scans."""
+        canonical_namespace = _validate_identifier("namespace", namespace)
+        required = _normalize_scopes(
+            required_scopes,
+            field_name="required_scopes",
+            allow_empty=True,
+            require_list=False,
+        )
+        token = _parse_bearer_header(authorization)
+        try:
+            supplied_digest = hashlib.sha256(token.encode("utf-8")).digest()
+        finally:
+            token = None
+
+        matched: Optional[_Credential] = None
+        for credential in self._credentials:
+            is_match = hmac.compare_digest(
+                supplied_digest,
+                credential.token_digest,
+            )
+            if is_match:
+                matched = credential
+
+        if matched is None:
+            raise AuthenticationError("invalid_bearer_token")
+        if matched.revoked:
+            raise AuthenticationError("credential_revoked")
+        if matched.namespace != canonical_namespace:
+            raise AuthenticationError("foreign_namespace")
+        if not required.issubset(matched.scopes):
+            raise AuthenticationError("missing_scopes")
+        return Principal(
+            owner_id=matched.owner_id,
+            namespace=matched.namespace,
+            key_id=matched.key_id,
+            scopes=tuple(sorted(matched.scopes)),
+        )
+
+
+def _reject_duplicate_object_keys(pairs):
+    result = {}
+    for key, value in pairs:
+        if key in result:
+            raise AuthConfigurationError("credential registry has duplicate object keys")
+        result[key] = value
+    return result
+
+
+def _validate_identifier(field_name: str, value: Any) -> str:
+    if type(value) is not str or not _IDENTIFIER_PATTERN.fullmatch(value):
+        raise AuthConfigurationError(
+            f"{field_name} must be a canonical lowercase identifier"
+        )
+    return value
+
+
+def _normalize_scopes(
+    value: Any,
+    *,
+    field_name: str,
+    allow_empty: bool,
+    require_list: bool,
+) -> FrozenSet[str]:
+    if require_list:
+        if type(value) is not list:
+            raise AuthConfigurationError(f"{field_name} must be a JSON array")
+        values = value
+    else:
+        if isinstance(value, (str, bytes)) or value is None:
+            raise AuthConfigurationError(f"{field_name} must be an iterable of scopes")
+        try:
+            values = list(value)
+        except TypeError as exc:
+            raise AuthConfigurationError(
+                f"{field_name} must be an iterable of scopes"
+            ) from exc
+    if not values and not allow_empty:
+        raise AuthConfigurationError(f"{field_name} must not be empty")
+    normalized = tuple(
+        _validate_identifier(f"{field_name} item", item) for item in values
+    )
+    if len(set(normalized)) != len(normalized):
+        raise AuthConfigurationError(f"{field_name} must not contain duplicates")
+    return frozenset(normalized)
+
+
+def _digest_registry_token(value: Any) -> bytes:
+    token = value
+    try:
+        if type(token) is not str or not token:
+            raise AuthConfigurationError("credential token must be a non-empty string")
+        if len(token) > 4096 or any(character.isspace() for character in token):
+            raise AuthConfigurationError(
+                "credential token must be a bounded bearer token"
+            )
+        return hashlib.sha256(token.encode("utf-8")).digest()
+    finally:
+        token = None
+
+
+def _parse_token_sha256(value: Any) -> bytes:
+    if (
+        type(value) is not str
+        or not re.fullmatch(r"[0-9a-f]{64}", value)
+        or value == hashlib.sha256(b"").hexdigest()
+    ):
+        raise AuthConfigurationError(
+            "credential token_sha256 must be a lowercase SHA-256 digest"
+        )
+    return bytes.fromhex(value)
+
+
+def _credential_from_mapping(raw: Any, index: int) -> _Credential:
+    if type(raw) is not dict:
+        raise AuthConfigurationError(
+            f"credential registry entry {index} must be an object"
+        )
+
+    keys = frozenset(raw)
+    unknown = keys - _CREDENTIAL_KEYS
+    missing = _REQUIRED_CREDENTIAL_KEYS - keys
+    token_keys = keys & _TOKEN_KEYS
+    if unknown or missing or len(token_keys) != 1:
+        raise AuthConfigurationError(
+            f"credential registry entry {index} has an invalid shape"
+        )
+
+    token_key = next(iter(token_keys))
+    token_value = raw.pop(token_key)
+    try:
+        token_digest = (
+            _digest_registry_token(token_value)
+            if token_key == "token"
+            else _parse_token_sha256(token_value)
+        )
+    finally:
+        token_value = None
+
+    owner_id = _validate_identifier("owner_id", raw["owner_id"])
+    namespace = _validate_identifier("namespace", raw["namespace"])
+    key_id = _validate_identifier("key_id", raw["key_id"])
+    scopes = _normalize_scopes(
+        raw["scopes"],
+        field_name="scopes",
+        allow_empty=False,
+        require_list=True,
+    )
+    revoked = raw.get("revoked", False)
+    if type(revoked) is not bool:
+        raise AuthConfigurationError("revoked must be a JSON boolean")
+    return _Credential(
+        owner_id=owner_id,
+        namespace=namespace,
+        key_id=key_id,
+        token_digest=token_digest,
+        scopes=scopes,
+        revoked=revoked,
+    )
+
+
+def _registry_from_credentials(
+    credentials: Iterable[_Credential],
+) -> CredentialRegistry:
+    values = []
+    token_digests = set()
+    key_ids = set()
+    for credential in credentials:
+        if credential.token_digest in token_digests:
+            raise AuthConfigurationError(
+                "credential registry contains a duplicate token"
+            )
+        key_identity = (credential.namespace, credential.key_id)
+        if key_identity in key_ids:
+            raise AuthConfigurationError(
+                "credential registry contains a duplicate namespace/key_id"
+            )
+        token_digests.add(credential.token_digest)
+        key_ids.add(key_identity)
+        values.append(credential)
+    return CredentialRegistry(values)
+
+
+def parse_credential_registry(
+    registry_json: Union[str, bytes, bytearray],
+) -> CredentialRegistry:
+    """Parse a strict version-1 JSON registry and immediately redact raw tokens."""
+    raw_registry = registry_json
+    try:
+        if not isinstance(raw_registry, (str, bytes, bytearray)):
+            raise AuthConfigurationError("credential registry must be JSON text")
+        if not raw_registry:
+            raise AuthConfigurationError("credential registry must not be empty")
+        try:
+            decoded = json.loads(
+                raw_registry,
+                object_pairs_hook=_reject_duplicate_object_keys,
+            )
+        except AuthConfigurationError:
+            raise
+        except (TypeError, UnicodeDecodeError, json.JSONDecodeError) as exc:
+            raise AuthConfigurationError("credential registry is not valid JSON") from exc
+    finally:
+        raw_registry = None
+        registry_json = None
+
+    if type(decoded) is not dict or frozenset(decoded) != _REGISTRY_KEYS:
+        raise AuthConfigurationError("credential registry has an invalid top-level shape")
+    if type(decoded["version"]) is not int or decoded["version"] != 1:
+        raise AuthConfigurationError("credential registry version must be 1")
+    raw_credentials = decoded["credentials"]
+    if type(raw_credentials) is not list or not raw_credentials:
+        raise AuthConfigurationError(
+            "credential registry must contain a non-empty credentials array"
+        )
+
+    credentials = []
+    for index, raw_credential in enumerate(raw_credentials):
+        credentials.append(_credential_from_mapping(raw_credential, index))
+    return _registry_from_credentials(credentials)
+
+
+def parse_legacy_principal_registry(
+    registry_json: Union[str, bytes, bytearray],
+    *,
+    namespace: str,
+) -> CredentialRegistry:
+    """Map the former digest-only principal registry to stable scoped credentials."""
+    canonical_namespace = _validate_identifier("namespace", namespace)
+    raw_registry = registry_json
+    try:
+        if not isinstance(raw_registry, (str, bytes, bytearray)):
+            raise AuthConfigurationError("principal registry must be JSON text")
+        if not raw_registry:
+            raise AuthConfigurationError("principal registry must not be empty")
+        try:
+            decoded = json.loads(
+                raw_registry,
+                object_pairs_hook=_reject_duplicate_object_keys,
+            )
+        except AuthConfigurationError:
+            raise
+        except (TypeError, UnicodeDecodeError, json.JSONDecodeError) as exc:
+            raise AuthConfigurationError("principal registry is not valid JSON") from exc
+    finally:
+        raw_registry = None
+        registry_json = None
+
+    if type(decoded) is not dict or not decoded:
+        raise AuthConfigurationError(
+            "principal registry must contain at least one principal"
+        )
+
+    credentials = []
+    for index, (principal_id, raw_record) in enumerate(decoded.items()):
+        owner_id = _validate_identifier("principal_id", principal_id)
+        if (
+            type(raw_record) is not dict
+            or frozenset(raw_record) != _LEGACY_PRINCIPAL_KEYS
+        ):
+            raise AuthConfigurationError(
+                f"principal registry entry {index} has an invalid shape"
+            )
+        roles = raw_record["roles"]
+        if type(roles) is not list or not roles:
+            raise AuthConfigurationError("principal roles must be a non-empty array")
+        if (
+            any(type(role) is not str or role not in _LEGACY_ROLE_SCOPES for role in roles)
+            or len(set(roles)) != len(roles)
+        ):
+            raise AuthConfigurationError("principal roles are invalid")
+        scopes = frozenset().union(
+            *(_LEGACY_ROLE_SCOPES[role] for role in roles)
+        )
+        key_id = "legacy:" + hashlib.sha256(
+            owner_id.encode("utf-8")
+        ).hexdigest()[:24]
+        credentials.append(
+            _Credential(
+                owner_id=owner_id,
+                namespace=canonical_namespace,
+                key_id=key_id,
+                token_digest=_parse_token_sha256(raw_record["token_sha256"]),
+                scopes=scopes,
+                revoked=False,
+            )
+        )
+    return _registry_from_credentials(credentials)
+
+
+def _legacy_registry(
+    *,
+    token: Optional[str],
+    owner_id: Optional[str],
+    namespace: Optional[str],
+    key_id: str,
+    scopes: Iterable[str],
+) -> CredentialRegistry:
+    if token is None or owner_id is None or namespace is None:
+        raise AuthConfigurationError(
+            "legacy authentication requires token, owner_id, and namespace"
+        )
+    credential = _Credential(
+        owner_id=_validate_identifier("legacy owner_id", owner_id),
+        namespace=_validate_identifier("legacy namespace", namespace),
+        key_id=_validate_identifier("legacy key_id", key_id),
+        token_digest=_digest_registry_token(token),
+        scopes=_normalize_scopes(
+            scopes,
+            field_name="legacy scopes",
+            allow_empty=False,
+            require_list=False,
+        ),
+        revoked=False,
+    )
+    token = None
+    return CredentialRegistry((credential,))
+
+
+def load_credential_registry(
+    registry_json: Optional[Union[str, bytes, bytearray]],
+    *,
+    principal_registry_json: Optional[Union[str, bytes, bytearray]] = None,
+    principal_registry_namespace: Optional[str] = None,
+    legacy_enabled: bool = False,
+    legacy_token: Optional[str] = None,
+    legacy_owner_id: Optional[str] = None,
+    legacy_namespace: Optional[str] = None,
+    legacy_key_id: str = "legacy",
+    legacy_scopes: Iterable[str] = (),
+) -> CredentialRegistry:
+    """Load registry JSON or an explicitly enabled, fully bound legacy credential."""
+    legacy_scopes = tuple(legacy_scopes)
+    configured_registries = sum(
+        value is not None for value in (registry_json, principal_registry_json)
+    )
+    if configured_registries > 1:
+        raise AuthConfigurationError(
+            "credential registries cannot be configured together"
+        )
+    legacy_values_present = any(
+        value is not None
+        for value in (legacy_token, legacy_owner_id, legacy_namespace)
+    ) or legacy_key_id != "legacy" or bool(legacy_scopes)
+    if registry_json is not None:
+        if legacy_enabled is True or legacy_values_present:
+            raise AuthConfigurationError(
+                "registry and legacy authentication cannot be configured together"
+            )
+        return parse_credential_registry(registry_json)
+    if principal_registry_json is not None:
+        if legacy_enabled is True or legacy_values_present:
+            raise AuthConfigurationError(
+                "registry and legacy authentication cannot be configured together"
+            )
+        if principal_registry_namespace is None:
+            raise AuthConfigurationError(
+                "principal registry requires a namespace binding"
+            )
+        return parse_legacy_principal_registry(
+            principal_registry_json,
+            namespace=principal_registry_namespace,
+        )
+    if legacy_enabled is not True:
+        if legacy_values_present:
+            raise AuthConfigurationError("legacy authentication is not enabled")
+        raise AuthConfigurationError("credential registry is not configured")
+    return _legacy_registry(
+        token=legacy_token,
+        owner_id=legacy_owner_id,
+        namespace=legacy_namespace,
+        key_id=legacy_key_id,
+        scopes=legacy_scopes,
+    )
+
+
+def _parse_bearer_header(authorization: Optional[str]) -> str:
+    if authorization is None:
+        raise AuthenticationError("missing_authorization")
+    if type(authorization) is not str:
+        raise AuthenticationError("invalid_authorization")
+    scheme, separator, token = authorization.partition(" ")
+    if (
+        not separator
+        or scheme.lower() != "bearer"
+        or not token
+        or len(token) > 4096
+        or any(character.isspace() for character in token)
+    ):
+        token = None
+        raise AuthenticationError("invalid_authorization")
+    return token
+
+
+def authenticate_bearer(
+    authorization: Optional[str],
+    registry: CredentialRegistry,
+    *,
+    namespace: str,
+    required_scopes: Iterable[str] = (),
+) -> Principal:
+    """Authenticate through a parsed registry and return an immutable principal."""
+    if not isinstance(registry, CredentialRegistry):
+        raise AuthConfigurationError("registry must be a CredentialRegistry")
+    return registry.authenticate(
+        authorization,
+        namespace=namespace,
+        required_scopes=required_scopes,
+    )

gdw_drain.py

@@ -1,13 +1,10 @@
-"""Token-fenced, idempotent drain for durable GDW effect-outbox rows."""
+"""Compatibility entry point for the single generation-fenced GDW drain."""
 
 from __future__ import annotations
 
-import os
-import uuid
-from datetime import datetime, timezone
 from typing import Any, Dict, Optional
 
-from gdw_proofs import export_proof_payload, export_receipt_projection
+from gdw_runtime import drain_once
 from gdw_workspace import GDWWorkspace
 
 
@@ -17,62 +14,18 @@ def drain_effects(
     limit: int = 100,
     worker_id: Optional[str] = None,
 ) -> Dict[str, Any]:
-    bounded = int(limit)
-    if bounded < 1 or bounded > 1000:
-        raise ValueError("drain limit must be between 1 and 1000")
-    worker = worker_id or f"gdw-runtime-{os.getpid()}-{uuid.uuid4().hex[:12]}"
-    exported = 0
-    failed = 0
-    error_classes = []
+    """Delegate to the runtime's only claim, export, and completion path."""
 
-    while exported + failed < bounded:
-        rows = workspace.claim_effects(worker, limit=1)
-        if not rows:
-            break
-        row = rows[0]
-        try:
-            workspace.validate_claimed_effect(row)
-            if row["kind"] == "proof_export":
-                artifact = export_proof_payload(
-                    row["payload"],
-                    artifact_identity=row["idempotency_key"],
-                    owner_id=row["owner_id"],
-                )
-            elif row["kind"] == "receipt_projection":
-                artifact = export_receipt_projection(
-                    row["payload"],
-                    row["idempotency_key"],
-                    owner_id=row["owner_id"],
-                )
-            else:
-                raise ValueError("unsupported effect kind")
-            workspace.mark_effect_exported(
-                row["idempotency_key"],
-                worker,
-                row["claim_token"],
-                artifact,
-                datetime.now(timezone.utc).isoformat(),
-            )
-            exported += 1
-        except Exception as exc:
-            workspace.release_effect(
-                row["idempotency_key"],
-                worker,
-                row["claim_token"],
-                f"{type(exc).__name__}: {exc}",
-            )
-            failed += 1
-            error_classes.append(type(exc).__name__)
-            break
-
-    integrity = workspace.integrity()
+    report = drain_once(
+        limit=limit,
+        worker_id=worker_id,
+        workspace=workspace,
+    )
+    integrity = workspace.integrity(global_scope=True)
     return {
-        "schema": "szl.gdw-effect-drain/v1",
-        "exported": exported,
-        "failed": failed,
-        "error_classes": error_classes,
-        "pending_effects": integrity["pending_effects"],
+        "schema": "szl.gdw-effect-drain/v2",
+        **report,
         "integrity_ok": integrity["ok"],
-        "generation_id": integrity["generation_id"],
+        "database_generation_id": integrity["database_generation_id"],
         "credential_values_recorded": False,
     }

gdw_proofs.py

@@ -4,10 +4,14 @@ import hashlib
 import json
 import os
 import tempfile
+import threading
 from pathlib import Path
 from typing import Any, Dict
 
 
+_ARTIFACT_QUOTA_LOCK = threading.RLock()
+
+
 def canonical_json(value: Any) -> str:
     return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
 
@@ -20,8 +24,9 @@ def build_proof_payload(
     proposal_id: str,
     request_id: str,
     request_digest: str,
+    namespace: str,
     owner_id: str,
-    generation_id: str,
+    database_generation_id: str,
     step: int,
     before_hash: str,
     after_hash: str,
@@ -38,8 +43,9 @@ def build_proof_payload(
         "proposal_id": proposal_id,
         "request_id": request_id,
         "request_digest": request_digest,
+        "namespace": namespace,
         "owner_id": owner_id,
-        "generation_id": generation_id,
+        "database_generation_id": database_generation_id,
         "step_id": step,
         "state_before_hash": before_hash,
         "state_after_hash": after_hash,
@@ -64,46 +70,52 @@ def build_proof_payload(
     return payload
 
 
-def _export_json_artifact(
+def _export_json_artifact_unlocked(
     root: Path,
     filename: str,
     payload: Dict[str, Any],
     owner_id: str,
 ) -> Dict[str, Any]:
-    if not owner_id:
+    if type(owner_id) is not str or not owner_id:
         raise ValueError("owner_id is required for artifact isolation")
+    root.mkdir(parents=True, exist_ok=True)
+    root = root.resolve()
     owner_scope = hashlib.sha256(owner_id.encode("utf-8")).hexdigest()[:32]
-    owner_root = root / owner_scope
-    owner_root.mkdir(parents=True, exist_ok=True)
+    owner_candidate = root / owner_scope
+    owner_candidate.mkdir(parents=True, exist_ok=True)
+    owner_root = owner_candidate.resolve()
+    if owner_root.parent != root or owner_root.name != owner_scope:
+        raise ValueError("artifact owner scope escapes the configured root")
     destination = owner_root / filename
     encoded = (json.dumps(payload, indent=2, sort_keys=True) + "\n").encode("utf-8")
     expected_sha256 = hashlib.sha256(encoded).hexdigest()
-
     if destination.exists():
-        existing = destination.read_bytes()
-        if existing != encoded:
+        if destination.read_bytes() != encoded:
             raise FileExistsError(
-                f"immutable artifact identity collision: {destination.name}"
+                "refusing to overwrite an existing non-identical GDW artifact"
             )
         return {
-            "status": "EXISTS_IDENTICAL",
+            "status": "EXPORTED",
             "path": str(destination),
             "sha256": expected_sha256,
+            "reused": True,
             "immutable": True,
             "owner_scope": owner_scope,
         }
-
-    owner_limit = int(os.environ.get("GDW_OWNER_MAX_ARTIFACTS", "10000"))
-    global_limit = int(os.environ.get("GDW_GLOBAL_MAX_ARTIFACTS", "100000"))
-    if owner_limit < 1 or owner_limit > 100000:
-        raise RuntimeError("GDW owner artifact quota is invalid")
-    if global_limit < owner_limit or global_limit > 1000000:
-        raise RuntimeError("GDW global artifact quota is invalid")
+    owner_limit = _bounded_artifact_limit(
+        "GDW_OWNER_MAX_ARTIFACTS", default=10_000, maximum=100_000
+    )
+    global_limit = _bounded_artifact_limit(
+        "GDW_GLOBAL_MAX_ARTIFACTS", default=100_000, maximum=1_000_000
+    )
+    if global_limit < owner_limit:
+        raise ValueError(
+            "GDW_GLOBAL_MAX_ARTIFACTS must be at least GDW_OWNER_MAX_ARTIFACTS"
+        )
     if sum(1 for _ in owner_root.glob("*.json")) >= owner_limit:
         raise RuntimeError("per-owner artifact quota exceeded")
     if sum(1 for _ in root.glob("*/*.json")) >= global_limit:
         raise RuntimeError("global artifact quota exceeded")
-
     handle, temporary = tempfile.mkstemp(
         prefix=".gdw-artifact-", suffix=".tmp", dir=owner_root
     )
@@ -115,10 +127,10 @@ def _export_json_artifact(
         try:
             os.link(temporary, destination)
         except FileExistsError:
-            existing = destination.read_bytes()
-            if existing != encoded:
+            if destination.read_bytes() != encoded:
                 raise FileExistsError(
-                    f"immutable artifact identity collision: {destination.name}"
+                    "refusing to overwrite a concurrently created "
+                    "non-identical GDW artifact"
                 )
     finally:
         if os.path.exists(temporary):
@@ -127,14 +139,50 @@ def _export_json_artifact(
         "status": "EXPORTED",
         "path": str(destination),
         "sha256": expected_sha256,
+        "reused": False,
         "immutable": True,
         "owner_scope": owner_scope,
     }
 
 
+def _bounded_artifact_limit(name: str, *, default: int, maximum: int) -> int:
+    raw = os.environ.get(name)
+    try:
+        value = int(raw) if raw not in (None, "") else default
+    except (TypeError, ValueError) as exc:
+        raise ValueError(f"{name} must be an integer") from exc
+    if value < 1 or value > maximum:
+        raise ValueError(f"{name} must be between 1 and {maximum}")
+    return value
+
+
+def _export_json_artifact(
+    root: Path,
+    filename: str,
+    payload: Dict[str, Any],
+    owner_id: str,
+) -> Dict[str, Any]:
+    with _ARTIFACT_QUOTA_LOCK:
+        return _export_json_artifact_unlocked(
+            root,
+            filename,
+            payload,
+            owner_id,
+        )
+
+
+def _validate_artifact_id(artifact_id: str) -> None:
+    if (
+        len(artifact_id) != 64
+        or any(ch not in "0123456789abcdef" for ch in artifact_id)
+    ):
+        raise ValueError("artifact_id must be a lowercase SHA-256 digest")
+
+
 def export_proof_payload(
     payload: Dict[str, Any],
-    artifact_identity: str | None = None,
+    *,
+    artifact_id: str | None = None,
     owner_id: str | None = None,
 ) -> Dict[str, Any]:
     root = Path(os.environ.get("GDW_PROOF_DIR", "output/proofs")).resolve()
@@ -146,36 +194,28 @@ def export_proof_payload(
     unsigned_payload.pop("payload_sha256", None)
     if claimed_digest != sha256_json(unsigned_payload):
         raise ValueError("proof payload_sha256 does not match canonical payload")
-    identity = artifact_identity or proposal_id
-    if (
-        len(identity) != 64
-        or any(ch not in "0123456789abcdef" for ch in identity)
-    ):
-        raise ValueError("artifact_identity must be a lowercase SHA-256 digest")
+    resolved_artifact_id = artifact_id or claimed_digest
+    _validate_artifact_id(resolved_artifact_id)
     artifact = _export_json_artifact(
         root,
-        f"{identity}.json",
+        f"{resolved_artifact_id}.json",
         payload,
         owner_id or str(payload.get("owner_id") or ""),
     )
-    artifact["artifact_identity"] = identity
+    artifact["artifact_identity"] = resolved_artifact_id
     artifact.update({"status": "INPUT_EXPORTED", "formal_status": "NOT_RUN"})
     return artifact
 
 
 def export_receipt_projection(
     payload: Dict[str, Any],
-    idempotency_key: str,
+    artifact_id: str,
     owner_id: str | None = None,
 ) -> Dict[str, Any]:
     root = Path(
         os.environ.get("GDW_RECEIPT_PROJECTION_DIR", "output/gdw/receipts")
     ).resolve()
-    if (
-        len(idempotency_key) != 64
-        or any(ch not in "0123456789abcdef" for ch in idempotency_key)
-    ):
-        raise ValueError("idempotency_key must be a lowercase SHA-256 digest")
+    _validate_artifact_id(artifact_id)
     claimed_digest = payload.get("receipt_hash")
     unsigned_payload = dict(payload)
     unsigned_payload.pop("receipt_hash", None)
@@ -183,11 +223,11 @@ def export_receipt_projection(
         raise ValueError("receipt_hash does not match canonical receipt")
     artifact = _export_json_artifact(
         root,
-        f"{idempotency_key}.json",
+        f"{artifact_id}.json",
         payload,
         owner_id or str(payload.get("owner_id") or ""),
     )
-    artifact["artifact_identity"] = idempotency_key
+    artifact["artifact_identity"] = artifact_id
     artifact.update(
         {
             "status": "RECEIPT_PROJECTED",

gdw_runtime.py

@@ -0,0 +1,601 @@
+"""Fail-closed GDW storage preparation and supervised outbox draining."""
+
+import json
+import os
+import runpy
+import sqlite3
+import sys
+import tempfile
+import threading
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Mapping, Optional
+
+from gdw_proofs import (
+    export_proof_payload,
+    export_receipt_projection,
+)
+from gdw_workspace import GDWWorkspace
+
+
+if __name__ == "__main__":
+    # Let route modules import the same stateful module while this file is the
+    # process entry point.
+    sys.modules.setdefault("gdw_runtime", sys.modules[__name__])
+
+
+ALLOWED_JOURNAL_MODES = {"DELETE", "WAL"}
+ALLOWED_SYNCHRONOUS_MODES = {"FULL", "NORMAL"}
+_STATE_LOCK = threading.RLock()
+_STATE: dict[str, Any] = {
+    "startup_state": "NOT_RUN",
+    "evidence_label": "UNAVAILABLE",
+    "drain": {
+        "enabled": False,
+        "running": False,
+        "last_outcome": "NOT_RUN",
+        "last_attempt_at": None,
+        "last_success_at": None,
+        "last_error": None,
+        "last_report": None,
+        "run_generation_id": None,
+        "success_run_generation_id": None,
+        "success_database_generation_id": None,
+        "max_staleness_seconds": None,
+    },
+}
+
+
+class GDWRuntimeError(RuntimeError):
+    """Fail-closed GDW production-runtime configuration error."""
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _enabled(value: Optional[str]) -> bool:
+    return (value or "").strip().lower() in {"1", "true", "yes", "on"}
+
+
+def _bounded_int(
+    value: Optional[str],
+    *,
+    default: int,
+    minimum: int,
+    maximum: int,
+    name: str,
+) -> int:
+    try:
+        parsed = int(value) if value not in (None, "") else default
+    except (TypeError, ValueError) as exc:
+        raise GDWRuntimeError(f"{name} must be an integer") from exc
+    if parsed < minimum or parsed > maximum:
+        raise GDWRuntimeError(
+            f"{name} must be between {minimum} and {maximum}"
+        )
+    return parsed
+
+
+def _path_within(path: Path, root: Path, *, name: str) -> Path:
+    candidate = path.resolve()
+    try:
+        candidate.relative_to(root)
+    except ValueError as exc:
+        raise GDWRuntimeError(
+            f"{name} must be contained by required mount {root}"
+        ) from exc
+    return candidate
+
+
+def _verify_writable_directory(path: Path) -> None:
+    path.mkdir(parents=True, exist_ok=True)
+    handle, probe = tempfile.mkstemp(prefix=".gdw-write-probe-", dir=path)
+    try:
+        with os.fdopen(handle, "wb") as stream:
+            stream.write(b"ok")
+            stream.flush()
+            os.fsync(stream.fileno())
+    finally:
+        Path(probe).unlink(missing_ok=True)
+
+
+def storage_contract(
+    environ: Optional[Mapping[str, str]] = None,
+) -> dict[str, Any]:
+    """Resolve and validate the declared GDW storage contract without writes."""
+
+    values = os.environ if environ is None else environ
+    database = Path(
+        values.get("GDW_DB_PATH", "output/gdw/gdw.sqlite3")
+    ).resolve()
+    proof_dir = Path(
+        values.get("GDW_PROOF_DIR", "output/proofs")
+    ).resolve()
+    receipt_dir = Path(
+        values.get("GDW_RECEIPT_PROJECTION_DIR", "output/gdw/receipts")
+    ).resolve()
+    required_mount_text = (values.get("GDW_REQUIRED_MOUNT") or "").strip()
+    persistent_required = _enabled(
+        values.get("GDW_REQUIRE_PERSISTENT_STORAGE")
+    )
+    journal_mode = (
+        values.get("GDW_SQLITE_JOURNAL") or "WAL"
+    ).strip().upper()
+    synchronous = (
+        values.get("GDW_SQLITE_SYNCHRONOUS") or "NORMAL"
+    ).strip().upper()
+    proof_export_mode = (
+        values.get("GDW_PROOF_EXPORT_MODE") or "outbox"
+    ).strip().lower()
+
+    if journal_mode not in ALLOWED_JOURNAL_MODES:
+        raise GDWRuntimeError(
+            "GDW_SQLITE_JOURNAL must be one of "
+            + ",".join(sorted(ALLOWED_JOURNAL_MODES))
+        )
+    if synchronous not in ALLOWED_SYNCHRONOUS_MODES:
+        raise GDWRuntimeError(
+            "GDW_SQLITE_SYNCHRONOUS must be one of "
+            + ",".join(sorted(ALLOWED_SYNCHRONOUS_MODES))
+        )
+    if proof_export_mode != "outbox":
+        raise GDWRuntimeError(
+            "GDW_PROOF_EXPORT_MODE must be 'outbox'; synchronous export "
+            "is not transaction-safe"
+        )
+    if persistent_required and not required_mount_text:
+        raise GDWRuntimeError(
+            "GDW_REQUIRED_MOUNT is required when persistent storage is required"
+        )
+
+    mount: Optional[Path] = None
+    mount_verified = False
+    if required_mount_text:
+        mount = Path(required_mount_text).resolve()
+        database = _path_within(database, mount, name="GDW_DB_PATH")
+        proof_dir = _path_within(proof_dir, mount, name="GDW_PROOF_DIR")
+        receipt_dir = _path_within(
+            receipt_dir,
+            mount,
+            name="GDW_RECEIPT_PROJECTION_DIR",
+        )
+        mount_verified = os.path.ismount(str(mount))
+        if not mount_verified:
+            raise GDWRuntimeError(
+                f"required GDW storage mount is not attached: {mount}"
+            )
+    if persistent_required and journal_mode != "DELETE":
+        raise GDWRuntimeError(
+            "persistent GDW storage requires GDW_SQLITE_JOURNAL=DELETE"
+        )
+
+    return {
+        "persistence_required": persistent_required,
+        "required_mount": str(mount) if mount else None,
+        "mount_verified": mount_verified,
+        "database_path": str(database),
+        "proof_dir": str(proof_dir),
+        "receipt_projection_dir": str(receipt_dir),
+        "journal_mode_requested": journal_mode,
+        "synchronous_requested": synchronous,
+        "proof_export_mode": proof_export_mode,
+    }
+
+
+def prepare_runtime(
+    environ: Optional[Mapping[str, str]] = None,
+) -> dict[str, Any]:
+    """Verify durable paths, initialise SQLite, and select the declared journal."""
+
+    contract = storage_contract(environ)
+    database = Path(contract["database_path"])
+    proof_dir = Path(contract["proof_dir"])
+    receipt_dir = Path(contract["receipt_projection_dir"])
+
+    try:
+        _verify_writable_directory(database.parent)
+        _verify_writable_directory(proof_dir)
+        _verify_writable_directory(receipt_dir)
+        workspace = GDWWorkspace(
+            str(database),
+            namespace=(os.environ.get("GDW_NAMESPACE") or "a11oy"),
+            owner_id=(
+                os.environ.get("GDW_SERVICE_OWNER_ID") or "gdw-runtime"
+            ),
+            production=False,
+        )
+        connection = sqlite3.connect(str(database), timeout=30)
+        try:
+            selected = connection.execute(
+                f"PRAGMA journal_mode={contract['journal_mode_requested']}"
+            ).fetchone()[0]
+            selected = str(selected).upper()
+            if selected != contract["journal_mode_requested"]:
+                raise GDWRuntimeError(
+                    "SQLite journal mode mismatch: requested "
+                    f"{contract['journal_mode_requested']}, observed {selected}"
+                )
+            connection.execute(
+                f"PRAGMA synchronous={contract['synchronous_requested']}"
+            )
+            observed_synchronous = int(
+                connection.execute("PRAGMA synchronous").fetchone()[0]
+            )
+            integrity = str(
+                connection.execute("PRAGMA integrity_check").fetchone()[0]
+            )
+            if integrity != "ok":
+                raise GDWRuntimeError(
+                    f"GDW SQLite integrity check failed: {integrity}"
+                )
+        finally:
+            connection.close()
+        observed = {
+            **contract,
+            "journal_mode_observed": selected,
+            "synchronous_observed": observed_synchronous,
+            "sqlite_integrity": integrity,
+            "schema_version": workspace.schema_version(),
+            "database_generation_id": workspace.database_generation_id,
+            "workspace_path": str(workspace.path),
+        }
+    except GDWRuntimeError:
+        raise
+    except Exception as exc:
+        raise GDWRuntimeError(
+            f"GDW persistent runtime preparation failed: {type(exc).__name__}"
+        ) from exc
+
+    with _STATE_LOCK:
+        _STATE.update(
+            {
+                "startup_state": "READY",
+                "evidence_label": "VERIFIED",
+                "storage": observed,
+                "prepared_at": _now(),
+                "error": None,
+            }
+        )
+    return observed
+
+
+def _verify_effect_binding(
+    workspace: GDWWorkspace,
+    row: Mapping[str, Any],
+) -> None:
+    errors = workspace.effect_binding_errors_for_row(dict(row))
+    if errors:
+        raise ValueError("invalid effect binding: " + ",".join(errors))
+
+
+def _export_effect(
+    workspace: GDWWorkspace,
+    row: Mapping[str, Any],
+) -> dict[str, Any]:
+    _verify_effect_binding(workspace, row)
+    artifact_id = str(row["intent_sha256"])
+    if row["kind"] == "proof_export":
+        return export_proof_payload(
+            row["payload"],
+            artifact_id=artifact_id,
+            owner_id=str(row["owner_id"]),
+        )
+    if row["kind"] == "receipt_projection":
+        return export_receipt_projection(
+            row["payload"],
+            artifact_id,
+            owner_id=str(row["owner_id"]),
+        )
+    raise ValueError(f"unsupported effect kind: {row['kind']}")
+
+
+def drain_once(
+    *,
+    limit: int = 100,
+    lease_seconds: int = 300,
+    worker_id: Optional[str] = None,
+    workspace: Optional[GDWWorkspace] = None,
+) -> dict[str, Any]:
+    """Run one bounded drain pass and leave failed rows retryable."""
+
+    bounded = _bounded_int(
+        str(limit), default=100, minimum=1, maximum=1000, name="limit"
+    )
+    lease = _bounded_int(
+        str(lease_seconds),
+        default=300,
+        minimum=1,
+        maximum=3600,
+        name="lease_seconds",
+    )
+    store = workspace or GDWWorkspace(
+        namespace=(os.environ.get("GDW_NAMESPACE") or "a11oy"),
+        owner_id=(os.environ.get("GDW_SERVICE_OWNER_ID") or "gdw-runtime"),
+    )
+    owner = worker_id or f"gdw-drain-{os.getpid()}-{uuid.uuid4().hex[:12]}"
+    exported = 0
+    failed = 0
+    errors = []
+    identities = (
+        [(store.namespace, store.owner_id)]
+        if workspace is not None
+        else store.pending_effect_identities()
+    )
+
+    for namespace, owner_id in identities:
+        remaining = bounded - exported - failed
+        if remaining <= 0:
+            break
+        if not store.production:
+            for row in store.pending_proofs(
+                remaining,
+                namespace=namespace,
+                owner_id=owner_id,
+            ):
+                try:
+                    artifact = export_proof_payload(
+                        row["payload"],
+                        owner_id=owner_id,
+                    )
+                    store.mark_proof_exported(
+                        row["proposal_id"],
+                        artifact,
+                        _now(),
+                        namespace=namespace,
+                        owner_id=owner_id,
+                    )
+                    exported += 1
+                except Exception as exc:
+                    failed += 1
+                    errors.append(f"legacy:{type(exc).__name__}")
+                if exported + failed >= bounded:
+                    break
+
+        remaining = bounded - exported - failed
+        if remaining <= 0:
+            break
+        rows = store.claim_effects(
+            owner,
+            limit=remaining,
+            lease_seconds=lease,
+            namespace=namespace,
+            owner_id=owner_id,
+        )
+        for row in rows:
+            try:
+                store.assert_effect_claim(
+                    row["idempotency_key"],
+                    owner,
+                    row["claim_generation"],
+                    namespace=namespace,
+                    owner_id=owner_id,
+                )
+                artifact = _export_effect(store, row)
+                store.mark_effect_exported(
+                    row["idempotency_key"],
+                    owner,
+                    row["claim_generation"],
+                    artifact,
+                    _now(),
+                    namespace=namespace,
+                    owner_id=owner_id,
+                )
+                exported += 1
+            except Exception as exc:
+                try:
+                    store.release_effect(
+                        row["idempotency_key"],
+                        owner,
+                        row["claim_generation"],
+                        f"{type(exc).__name__}: {str(exc)[:240]}",
+                        namespace=namespace,
+                        owner_id=owner_id,
+                    )
+                except RuntimeError:
+                    errors.append(f"{row['kind']}:CLAIM_LOST")
+                failed += 1
+                errors.append(f"{row['kind']}:{type(exc).__name__}")
+
+    integrity = store.integrity(global_scope=True)
+    return {
+        "attempted": exported + failed,
+        "exported": exported,
+        "failed": failed,
+        "pending_effects": integrity["pending_effects"],
+        "legacy_pending_proofs": integrity["pending_proofs"],
+        "sqlite_integrity": integrity["sqlite_integrity"],
+        "errors": errors,
+    }
+
+
+def _set_drain_state(**values: Any) -> None:
+    with _STATE_LOCK:
+        drain = dict(_STATE["drain"])
+        drain.update(values)
+        _STATE["drain"] = drain
+
+
+class OutboxSupervisor:
+    """Single-process bounded outbox worker with retry backoff."""
+
+    def __init__(
+        self,
+        *,
+        enabled: bool,
+        interval_seconds: int,
+        retry_max_seconds: int,
+        batch_size: int,
+        lease_seconds: int,
+    ) -> None:
+        self.enabled = enabled
+        self.interval_seconds = interval_seconds
+        self.retry_max_seconds = retry_max_seconds
+        self.batch_size = batch_size
+        self.lease_seconds = lease_seconds
+        self.worker_id = f"gdw-supervisor-{os.getpid()}-{uuid.uuid4().hex[:12]}"
+        self._stop = threading.Event()
+        self._thread: Optional[threading.Thread] = None
+
+    @classmethod
+    def from_environment(cls) -> "OutboxSupervisor":
+        return cls(
+            enabled=_enabled(os.environ.get("GDW_OUTBOX_ENABLED")),
+            interval_seconds=_bounded_int(
+                os.environ.get("GDW_OUTBOX_INTERVAL_SECONDS"),
+                default=5,
+                minimum=1,
+                maximum=3600,
+                name="GDW_OUTBOX_INTERVAL_SECONDS",
+            ),
+            retry_max_seconds=_bounded_int(
+                os.environ.get("GDW_OUTBOX_RETRY_MAX_SECONDS"),
+                default=60,
+                minimum=1,
+                maximum=3600,
+                name="GDW_OUTBOX_RETRY_MAX_SECONDS",
+            ),
+            batch_size=_bounded_int(
+                os.environ.get("GDW_OUTBOX_BATCH_SIZE"),
+                default=100,
+                minimum=1,
+                maximum=1000,
+                name="GDW_OUTBOX_BATCH_SIZE",
+            ),
+            lease_seconds=_bounded_int(
+                os.environ.get("GDW_OUTBOX_LEASE_SECONDS"),
+                default=300,
+                minimum=1,
+                maximum=3600,
+                name="GDW_OUTBOX_LEASE_SECONDS",
+            ),
+        )
+
+    def start(self) -> None:
+        _set_drain_state(enabled=self.enabled)
+        if not self.enabled:
+            _set_drain_state(last_outcome="DISABLED")
+            return
+        if self._thread and self._thread.is_alive():
+            return
+        self._thread = threading.Thread(
+            target=self._run,
+            name="gdw-outbox-supervisor",
+            daemon=True,
+        )
+        self._thread.start()
+
+    def _run(self) -> None:
+        run_generation_id = uuid.uuid4().hex
+        with _STATE_LOCK:
+            database_generation_id = str(
+                (_STATE.get("storage") or {}).get("database_generation_id") or ""
+            )
+        _set_drain_state(
+            running=True,
+            worker_id=self.worker_id,
+            last_outcome="STARTING",
+            last_attempt_at=None,
+            last_success_at=None,
+            last_error=None,
+            last_report=None,
+            run_generation_id=run_generation_id,
+            success_run_generation_id=None,
+            success_database_generation_id=None,
+            max_staleness_seconds=max(30, self.interval_seconds * 3),
+        )
+        delay = 0
+        retry_delay = self.interval_seconds
+        try:
+            while not self._stop.wait(delay):
+                attempted_at = _now()
+                _set_drain_state(last_attempt_at=attempted_at)
+                try:
+                    report = drain_once(
+                        limit=self.batch_size,
+                        lease_seconds=self.lease_seconds,
+                        worker_id=self.worker_id,
+                    )
+                    if report["failed"] or report["legacy_pending_proofs"]:
+                        retry_delay = min(
+                            self.retry_max_seconds,
+                            max(self.interval_seconds, retry_delay * 2),
+                        )
+                        delay = retry_delay
+                        _set_drain_state(
+                            last_outcome="RETRY_SCHEDULED",
+                            last_error=(
+                                "bounded drain pass reported failures or "
+                                "unmigrated legacy proofs"
+                            ),
+                            last_report=report,
+                        )
+                    else:
+                        retry_delay = self.interval_seconds
+                        delay = self.interval_seconds
+                        _set_drain_state(
+                            last_outcome="SUCCEEDED",
+                            last_success_at=_now(),
+                            last_error=None,
+                            last_report=report,
+                            success_run_generation_id=run_generation_id,
+                            success_database_generation_id=(
+                                database_generation_id
+                            ),
+                        )
+                except Exception as exc:
+                    retry_delay = min(
+                        self.retry_max_seconds,
+                        max(self.interval_seconds, retry_delay * 2),
+                    )
+                    delay = retry_delay
+                    _set_drain_state(
+                        last_outcome="RETRY_SCHEDULED",
+                        last_error=f"{type(exc).__name__}: {str(exc)[:240]}",
+                    )
+        finally:
+            _set_drain_state(running=False)
+
+    def stop(self, timeout_seconds: float = 10.0) -> None:
+        self._stop.set()
+        if self._thread:
+            self._thread.join(timeout=max(0.0, timeout_seconds))
+        _set_drain_state(running=False)
+
+
+def runtime_health() -> dict[str, Any]:
+    """Return secret-free observed inputs for the GDW health route."""
+
+    with _STATE_LOCK:
+        return json.loads(json.dumps(_STATE))
+
+
+def main() -> int:
+    try:
+        prepare_runtime()
+    except Exception as exc:
+        with _STATE_LOCK:
+            _STATE.update(
+                {
+                    "startup_state": "BLOCKED",
+                    "evidence_label": "VERIFIED",
+                    "error": f"{type(exc).__name__}: {str(exc)[:240]}",
+                }
+            )
+        raise
+
+    supervisor = OutboxSupervisor.from_environment()
+    supervisor.start()
+    try:
+        runpy.run_path(
+            str(Path(__file__).with_name("serve.py")),
+            run_name="__main__",
+        )
+    finally:
+        supervisor.stop()
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

routers/gdw_frontier.py

@@ -1,12 +1,12 @@
 """Authenticated Governed Delta Workspace API and benchmark surfaces."""
 
 import hashlib
-import hmac
 import json
 import os
 import re
+import threading
 import time
-from datetime import datetime, timedelta, timezone
+from datetime import datetime, timezone
 from typing import List, Literal, Optional
 
 from fastapi import Header, HTTPException, Request
@@ -14,17 +14,31 @@ from fastapi.responses import PlainTextResponse
 from pydantic import BaseModel, Field, ValidationError
 
 from gdw_attention import AttentionFeatures, choose_attention_mode
-from gdw_drain import drain_effects
+from gdw_auth import (
+    AuthConfigurationError,
+    AuthenticationError,
+    Principal,
+    authenticate_bearer,
+    load_credential_registry,
+)
 from gdw_proofs import build_proof_payload, sha256_json
+from gdw_runtime import drain_once, runtime_health
 from gdw_telemetry import GDWTelemetry
-from gdw_workspace import GDWWorkspace
+from gdw_workspace import (
+    GDWConfigurationError,
+    GDWLifecycleError,
+    GDWQuotaExceeded,
+    GDWWorkspace,
+)
 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"}
-_PRINCIPAL_ROLES = {"user", "admin"}
+_AUTH_LOCK = threading.RLock()
+_AUTH_REGISTRY = None
+_AUTH_FINGERPRINT = None
 
 
 class GDWStepRequest(BaseModel):
@@ -52,6 +66,24 @@ def _dump_model(model):
     return model.dict()
 
 
+def _model_schema(model):
+    if hasattr(model, "model_json_schema"):
+        return model.model_json_schema()
+    return model.schema()
+
+
+def _validate_step_payload(value) -> GDWStepRequest:
+    try:
+        if hasattr(GDWStepRequest, "model_validate"):
+            return GDWStepRequest.model_validate(value)
+        return GDWStepRequest.parse_obj(value)
+    except ValidationError as exc:
+        raise HTTPException(
+            status_code=422,
+            detail="invalid GDW step request",
+        ) from exc
+
+
 def _now() -> str:
     return datetime.now(timezone.utc).isoformat()
 
@@ -63,171 +95,241 @@ def _sha(value) -> str:
     return hashlib.sha256(encoded).hexdigest()
 
 
-def _principal_registry() -> dict:
-    configured = os.environ.get("GDW_PRINCIPALS_JSON", "")
-    try:
-        registry = json.loads(configured)
-    except Exception as exc:
-        raise RuntimeError("GDW principal registry is invalid") from exc
-    if not isinstance(registry, dict) or not registry:
-        raise RuntimeError("GDW principal registry is unavailable")
-
-    normalized = {}
-    token_digests = set()
-    for principal_id, record in registry.items():
-        if not isinstance(principal_id, str) or not _ID_PATTERN.fullmatch(
-            principal_id
-        ):
-            raise RuntimeError("GDW principal identifier is invalid")
-        if not isinstance(record, dict):
-            raise RuntimeError("GDW principal record is invalid")
-        token_sha256 = str(record.get("token_sha256") or "")
-        if (
-            len(token_sha256) != 64
-            or any(ch not in "0123456789abcdef" for ch in token_sha256)
-            or token_sha256 == hashlib.sha256(b"").hexdigest()
-            or token_sha256 in token_digests
-        ):
-            raise RuntimeError("GDW principal token binding is invalid")
-        roles = record.get("roles")
-        if (
-            not isinstance(roles, list)
-            or not roles
-            or not set(roles).issubset(_PRINCIPAL_ROLES)
-        ):
-            raise RuntimeError("GDW principal roles are invalid")
-        token_digests.add(token_sha256)
-        normalized[principal_id] = {
-            "principal_id": principal_id,
-            "token_sha256": token_sha256,
-            "roles": sorted(set(roles)),
+def _credential_registry():
+    global _AUTH_FINGERPRINT, _AUTH_REGISTRY
+    registry_json = os.environ.get("GDW_CREDENTIALS_JSON")
+    principal_registry_json = os.environ.get("GDW_PRINCIPALS_JSON")
+    legacy_enabled = os.environ.get(
+        "GDW_ALLOW_LEGACY_AUTH", ""
+    ).strip().lower() in {"1", "true", "yes", "on"}
+    legacy_scopes = tuple(
+        value.strip()
+        for value in os.environ.get("GDW_LEGACY_SCOPES", "").split(",")
+        if value.strip()
+    )
+    fingerprint = _sha(
+        {
+            "registry": registry_json,
+            "principal_registry": principal_registry_json,
+            "legacy_enabled": legacy_enabled,
+            "legacy_token": os.environ.get("GDW_AUTH_TOKEN"),
+            "legacy_owner": os.environ.get("GDW_OWNER_ID"),
+            "legacy_namespace": os.environ.get("GDW_NAMESPACE"),
+            "legacy_scopes": legacy_scopes,
         }
-    return normalized
+    )
+    with _AUTH_LOCK:
+        if _AUTH_REGISTRY is not None and _AUTH_FINGERPRINT == fingerprint:
+            return _AUTH_REGISTRY
+        registry = load_credential_registry(
+            registry_json,
+            principal_registry_json=principal_registry_json,
+            principal_registry_namespace=os.environ.get("GDW_NAMESPACE") or "a11oy",
+            legacy_enabled=legacy_enabled,
+            legacy_token=os.environ.get("GDW_AUTH_TOKEN"),
+            legacy_owner_id=os.environ.get("GDW_OWNER_ID"),
+            legacy_namespace=os.environ.get("GDW_NAMESPACE"),
+            legacy_scopes=legacy_scopes,
+        )
+        _AUTH_REGISTRY = registry
+        _AUTH_FINGERPRINT = fingerprint
+        return registry
 
 
-def _authenticate(
+def _authorise(
     authorization: Optional[str],
-    required_role: Optional[str] = None,
-) -> dict:
+    *,
+    namespace: str,
+    required_scopes=(),
+) -> Principal:
     try:
-        registry = _principal_registry()
-    except Exception as exc:
+        return authenticate_bearer(
+            authorization,
+            _credential_registry(),
+            namespace=namespace,
+            required_scopes=required_scopes,
+        )
+    except AuthConfigurationError as exc:
         raise HTTPException(
             status_code=503,
-            detail="GDW principal registry is unavailable",
+            detail="GDW credential registry is unavailable",
         ) from exc
-    supplied = authorization or ""
-    if not supplied.startswith("Bearer "):
-        raise HTTPException(status_code=401, detail="invalid bearer token")
-    token = supplied[len("Bearer ") :]
-    if not token:
-        raise HTTPException(status_code=401, detail="invalid bearer token")
-    digest = hashlib.sha256(token.encode("utf-8")).hexdigest()
-    principal = None
-    for record in registry.values():
-        if hmac.compare_digest(digest, record["token_sha256"]):
-            principal = record
-    if principal is None:
-        raise HTTPException(status_code=401, detail="invalid bearer token")
-    if required_role and required_role not in principal["roles"]:
-        raise HTTPException(status_code=403, detail="principal role is insufficient")
-    return principal
-
-
-def _bounded_config(name: str, default: int, maximum: int) -> int:
-    try:
-        value = int(os.environ.get(name, str(default)))
-    except Exception as exc:
-        raise RuntimeError(f"{name} must be an integer") from exc
-    if value < 1 or value > maximum:
-        raise RuntimeError(f"{name} must be between 1 and {maximum}")
-    return value
-
-
-def _admission_limits() -> dict:
-    owner_requests = _bounded_config("GDW_OWNER_MAX_REQUESTS", 1000, 10000)
-    owner_sessions = _bounded_config("GDW_OWNER_MAX_SESSIONS", 100, 1000)
-    global_requests = _bounded_config(
-        "GDW_GLOBAL_MAX_REQUESTS", 100000, 1000000
+    except AuthenticationError as exc:
+        status = 403 if exc.code in {
+            "credential_revoked",
+            "foreign_namespace",
+            "missing_scopes",
+        } else 401
+        raise HTTPException(status_code=status, detail=exc.code) from exc
+
+
+def _workspace(principal: Principal) -> GDWWorkspace:
+    return GDWWorkspace(
+        namespace=principal.namespace,
+        owner_id=principal.owner_id,
     )
-    global_sessions = _bounded_config(
-        "GDW_GLOBAL_MAX_SESSIONS", 10000, 100000
-    )
-    if global_requests < owner_requests or global_sessions < owner_sessions:
-        raise RuntimeError("GDW global quotas cannot be lower than owner quotas")
-    return {
-        "owner_requests": owner_requests,
-        "owner_sessions": owner_sessions,
-        "global_requests": global_requests,
-        "global_sessions": global_sessions,
-    }
 
 
-def _retention_seconds() -> int:
-    return _bounded_config("GDW_RETENTION_SECONDS", 604800, 31536000)
-
+def _governance_ready() -> bool:
+    try:
+        import szl_codename_gate
+        import szl_colang_policy
 
-def _effect_limits() -> dict:
-    owner_artifacts = _bounded_config(
-        "GDW_OWNER_MAX_ARTIFACTS", 10000, 100000
-    )
-    global_artifacts = _bounded_config(
-        "GDW_GLOBAL_MAX_ARTIFACTS", 100000, 1000000
-    )
-    if global_artifacts < owner_artifacts:
-        raise RuntimeError(
-            "GDW global artifact quota cannot be lower than owner quota"
+        policy = szl_colang_policy.get_policy()
+        return bool(
+            policy.loaded
+            and policy.enforcement_ready
+            and callable(getattr(szl_codename_gate, "scan_text", None))
         )
-    return {
-        "owner_artifacts": owner_artifacts,
-        "global_artifacts": global_artifacts,
-        "max_attempts": _bounded_config("GDW_MAX_EFFECT_ATTEMPTS", 20, 100),
-    }
+    except Exception:
+        return False
 
 
-def _strict_policy():
-    import szl_colang_policy
-
-    policy = szl_colang_policy.get_policy(reload=True)
-    status = policy.enforcement_contract_status()
-    if not policy.loaded or not status["valid"]:
-        raise RuntimeError("strict file-backed governance is unavailable")
-    return policy
-
-
-def _runtime_workspace() -> GDWWorkspace:
-    _principal_registry()
-    _admission_limits()
-    _retention_seconds()
-    _effect_limits()
-    _strict_policy()
-    workspace = GDWWorkspace()
-    integrity = workspace.integrity()
-    if not integrity["ok"]:
-        raise RuntimeError("GDW workspace integrity gate is closed")
-    return workspace
+def _policy_bundle_sha256() -> Optional[str]:
+    try:
+        import szl_colang_policy
+
+        policy = szl_colang_policy.get_policy()
+        if not policy.enforcement_ready:
+            return None
+        return policy.bundle_sha256
+    except Exception:
+        return None
+
+
+def _write_readiness(
+    namespace: str,
+) -> tuple[bool, list[str], int, dict, bool]:
+    runtime = runtime_health()
+    production = os.environ.get(
+        "GDW_PRODUCTION_MODE", ""
+    ).strip().lower() in {"1", "true", "yes", "on"}
+    blockers = []
+    if production:
+        storage = runtime.get("storage") or {}
+        drain = runtime.get("drain") or {}
+        if runtime.get("evidence_label") != "VERIFIED":
+            blockers.append("RUNTIME_EVIDENCE_UNVERIFIED")
+        if runtime.get("startup_state") != "READY":
+            blockers.append("RUNTIME_NOT_READY")
+        if storage.get("sqlite_integrity") != "ok":
+            blockers.append("SQLITE_INTEGRITY_UNVERIFIED")
+        if storage.get("schema_version") != GDWWorkspace.schema_version():
+            blockers.append("SCHEMA_VERSION_UNVERIFIED")
+        if not re.fullmatch(
+            r"[0-9a-f]{32}",
+            str(storage.get("database_generation_id") or ""),
+        ):
+            blockers.append("DATABASE_GENERATION_UNVERIFIED")
+        if storage.get("proof_export_mode") != "outbox":
+            blockers.append("OUTBOX_MODE_UNVERIFIED")
+        if storage.get("journal_mode_observed") != storage.get(
+            "journal_mode_requested"
+        ):
+            blockers.append("JOURNAL_MODE_MISMATCH")
+        if storage.get("persistence_required") is not True:
+            blockers.append("PERSISTENCE_NOT_REQUIRED")
+        if storage.get("mount_verified") is not True:
+            blockers.append("PERSISTENT_MOUNT_UNVERIFIED")
+        expected_synchronous = {"FULL": 2, "NORMAL": 1}.get(
+            storage.get("synchronous_requested")
+        )
+        if storage.get("synchronous_observed") != expected_synchronous:
+            blockers.append("SYNCHRONOUS_MODE_MISMATCH")
+        if not drain.get("enabled") or not drain.get("running"):
+            blockers.append("OUTBOX_SUPERVISOR_NOT_RUNNING")
+        if drain.get("last_outcome") != "SUCCEEDED":
+            blockers.append("OUTBOX_SUPERVISOR_NOT_HEALTHY")
+        if drain.get("success_run_generation_id") != drain.get(
+            "run_generation_id"
+        ):
+            blockers.append("OUTBOX_SUPERVISOR_SUCCESS_STALE")
+        if drain.get("success_database_generation_id") != storage.get(
+            "database_generation_id"
+        ):
+            blockers.append("OUTBOX_SUPERVISOR_DATABASE_STALE")
+        try:
+            success_at = datetime.fromisoformat(
+                str(drain.get("last_success_at") or "").replace("Z", "+00:00")
+            )
+            age = (datetime.now(timezone.utc) - success_at).total_seconds()
+            max_age = int(drain.get("max_staleness_seconds") or 0)
+            if max_age < 1 or age < 0 or age > max_age:
+                raise ValueError
+        except (TypeError, ValueError):
+            blockers.append("OUTBOX_SUPERVISOR_HEARTBEAT_STALE")
+    try:
+        credentials = _credential_registry()
+        credential_count = credentials.credential_count
+    except AuthConfigurationError:
+        credential_count = 0
+        blockers.append("CREDENTIAL_REGISTRY_UNAVAILABLE")
+    governance_ready = _governance_ready()
+    if not governance_ready:
+        blockers.append("GOVERNANCE_SOURCE_UNREADY")
+    return (
+        not blockers,
+        sorted(set(blockers)),
+        credential_count,
+        runtime,
+        governance_ready,
+    )
 
 
-def _available_workspace() -> GDWWorkspace:
-    try:
-        return _runtime_workspace()
-    except Exception as exc:
+def _require_write_ready(namespace: str) -> None:
+    ready, blockers, _, _, _ = _write_readiness(namespace)
+    if not ready:
         raise HTTPException(
             status_code=503,
-            detail=f"GDW semantic gate is closed: {type(exc).__name__}",
-        ) from exc
+            detail={
+                "reason": "GDW_WRITE_SURFACE_UNAVAILABLE",
+                "write_blockers": blockers,
+            },
+        )
 
 
-def _step_openapi() -> dict:
-    if hasattr(GDWStepRequest, "model_json_schema"):
-        schema = GDWStepRequest.model_json_schema()
-    else:
-        schema = GDWStepRequest.schema()
-    return {
-        "requestBody": {
-            "required": True,
-            "content": {"application/json": {"schema": schema}},
+def _public_runtime_health(runtime: dict) -> dict:
+    storage = runtime.get("storage")
+    public_storage = None
+    if isinstance(storage, dict):
+        public_storage = {
+            key: storage.get(key)
+            for key in (
+                "persistence_required",
+                "mount_verified",
+                "journal_mode_requested",
+                "synchronous_requested",
+                "proof_export_mode",
+                "journal_mode_observed",
+                "synchronous_observed",
+                "sqlite_integrity",
+                "schema_version",
+                "database_generation_id",
+            )
+            if key in storage
+        }
+    drain = runtime.get("drain")
+    public_drain = None
+    if isinstance(drain, dict):
+        public_drain = {
+            key: drain.get(key)
+            for key in (
+                "enabled",
+                "running",
+                "last_outcome",
+                "last_attempt_at",
+                "last_success_at",
+                "last_error",
+            )
+            if key in drain
         }
+    return {
+        "startup_state": runtime.get("startup_state"),
+        "evidence_label": runtime.get("evidence_label"),
+        "storage": public_storage,
+        "drain": public_drain,
+        "prepared_at": runtime.get("prepared_at"),
+        "error": runtime.get("error"),
     }
 
 
@@ -259,7 +361,7 @@ def _governance_gate(
     payload_data: dict,
     request_id: str,
     request_digest: str,
-    principal_id: str,
+    principal: Principal,
 ) -> dict:
     action = {
         "tool": "execute",
@@ -269,17 +371,22 @@ def _governance_gate(
         "target": payload_data["session_id"],
         "request_id": request_id,
         "request_digest": request_digest,
-        "principal_id": principal_id,
-        "target_owner_id": principal_id,
+        "principal": principal.owner_id,
+        "namespace": principal.namespace,
+        "credential_key_id": principal.key_id,
         "text": payload_data["request"],
         "high_impact": float(payload_data["risk_budget"]) >= 0.75,
         "irreversible": False,
     }
     try:
-        policy = _strict_policy()
-        colang = policy.evaluate_strict(action)
-        if not colang.get("enforcement_contract", {}).get("valid"):
-            raise RuntimeError("strict policy enforcement contract is invalid")
+        import szl_colang_policy
+
+        policy = szl_colang_policy.get_policy()
+        if not policy.enforcement_ready:
+            raise RuntimeError("exact file-backed Colang policy is not ready")
+        colang = policy.evaluate(action)
+        if not colang.get("enforcement_ready"):
+            raise RuntimeError("Colang policy evaluation failed exact-source checks")
     except Exception as exc:
         return {
             "allowed": False,
@@ -287,7 +394,11 @@ def _governance_gate(
             "reason_codes": ["DOCTRINE_GATE_UNAVAILABLE"],
             "detail": type(exc).__name__,
             "writer_is_judge": True,
-            "enforcement_mode": "IN_PROCESS_STRICT_FILE_LOCK",
+            "principal": {
+                "owner_id": principal.owner_id,
+                "namespace": principal.namespace,
+                "key_id": principal.key_id,
+            },
         }
 
     try:
@@ -306,13 +417,17 @@ def _governance_gate(
             "reason_codes": ["CODENAME_GATE_UNAVAILABLE"],
             "detail": type(exc).__name__,
             "writer_is_judge": True,
-            "enforcement_mode": "IN_PROCESS_STRICT_FILE_LOCK",
+            "principal": {
+                "owner_id": principal.owner_id,
+                "namespace": principal.namespace,
+                "key_id": principal.key_id,
+            },
             "colang": {
                 "decision": colang.get("decision"),
                 "fired_flows": colang.get("fired_flows", []),
                 "flows_evaluated": colang.get("flows_evaluated", []),
                 "policy_files": colang.get("policy_files", []),
-                "enforcement_contract": colang.get("enforcement_contract", {}),
+                "bundle_sha256": colang.get("bundle_sha256"),
             },
         }
 
@@ -324,15 +439,19 @@ def _governance_gate(
     return {
         "allowed": not reasons,
         "decision": "ALLOW" if not reasons else "DENY",
-        "reason_codes": reasons or ["STRICT_FILE_BACKED_GOVERNANCE_PASS"],
+        "reason_codes": reasons or ["FILE_BACKED_GOVERNANCE_PASS"],
         "writer_is_judge": True,
-        "enforcement_mode": "IN_PROCESS_STRICT_FILE_LOCK",
+        "principal": {
+            "owner_id": principal.owner_id,
+            "namespace": principal.namespace,
+            "key_id": principal.key_id,
+        },
         "colang": {
             "decision": colang.get("decision"),
             "fired_flows": colang.get("fired_flows", []),
             "flows_evaluated": colang.get("flows_evaluated", []),
             "policy_files": colang.get("policy_files", []),
-            "enforcement_contract": colang.get("enforcement_contract", {}),
+            "bundle_sha256": colang.get("bundle_sha256"),
         },
         "codename_gate": {
             "clean": not codename_hits,
@@ -346,14 +465,14 @@ def _atomic_receipt(
     proposal_id: str,
     request_id: str,
     request_digest: str,
-    owner_id: str,
-    generation_id: str,
     session_id: str,
     step: int,
     before_hash: str,
     after_hash: str,
     scheduler_mode: str,
     governance: dict,
+    principal: Principal,
+    database_generation_id: str,
     timestamp: str,
 ) -> dict:
     receipt = {
@@ -362,9 +481,11 @@ def _atomic_receipt(
         "proposal_id": proposal_id,
         "request_id": request_id,
         "request_digest": request_digest,
-        "owner_id": owner_id,
-        "generation_id": generation_id,
         "session_id": session_id,
+        "owner_id": principal.owner_id,
+        "namespace": principal.namespace,
+        "database_generation_id": database_generation_id,
+        "credential_key_id": principal.key_id,
         "step": step,
         "state_before_hash": before_hash,
         "state_after_hash": after_hash,
@@ -377,81 +498,41 @@ def _atomic_receipt(
     return receipt
 
 
-def _effect_key(
-    *,
-    generation_id: str,
-    owner_id: str,
-    request_id: str,
-    request_digest: str,
-    kind: str,
-    canonical_identity: str,
-    payload_sha256: str,
-) -> str:
-    return hashlib.sha256(
-        (
-            f"{generation_id}:{owner_id}:{request_id}:{request_digest}:"
-            f"{kind}:{canonical_identity}:{payload_sha256}"
-        ).encode("utf-8")
-    ).hexdigest()
-
-
 def register(app, ns: str = "a11oy"):
     prefix = f"/api/{ns}/v1/gdw"
 
     @app.get(prefix + "/healthz")
     @app.get("/v1/gdw/healthz")
     def gdw_healthz():
-        try:
-            workspace = _runtime_workspace()
-            return {
-                "service": "gdw-frontier",
-                "status": "REAL",
-                "write_ready": True,
-                "persistence": f"SQLITE_{workspace.journal_mode}",
-                "generation_id": workspace.generation_id(),
-                "external_effects": "OUTBOX_ONLY",
-                "benchmark_claim": "UNMEASURED",
-            }
-        except Exception as exc:
-            return {
-                "service": "gdw-frontier",
-                "status": "UNAVAILABLE",
-                "label": "UNAVAILABLE",
-                "write_ready": False,
-                "persistence": "SQLITE_CONFIGURATION_GATED",
-                "external_effects": "DISABLED",
-                "reason": f"semantic gate closed: {type(exc).__name__}",
-                "benchmark_claim": "UNMEASURED",
-            }
-
-    @app.post(prefix + "/drain")
-    @app.post("/v1/gdw/drain")
-    def gdw_drain(
-        limit: int = 100,
-        authorization: Optional[str] = Header(default=None, alias="Authorization"),
-    ):
-        _authenticate(authorization, "admin")
-        workspace = _available_workspace()
-        try:
-            result = drain_effects(workspace, limit=limit)
-        except ValueError as exc:
-            raise HTTPException(status_code=422, detail=str(exc)) from exc
-        except Exception as exc:
-            raise HTTPException(
-                status_code=503,
-                detail=f"GDW effect drain failed closed: {type(exc).__name__}",
-            ) from exc
-        if result["failed"] or not result["integrity_ok"]:
-            raise HTTPException(status_code=503, detail=result)
-        return result
+        (
+            write_ready,
+            blockers,
+            credential_count,
+            runtime,
+            governance_ready,
+        ) = _write_readiness(ns)
+        public_runtime = _public_runtime_health(runtime)
+        return {
+            "service": "gdw-frontier",
+            "status": "REAL" if write_ready else "UNAVAILABLE",
+            "write_ready": write_ready,
+            "credential_count": credential_count,
+            "governance_ready": governance_ready,
+            "write_blockers": blockers,
+            "persistence": public_runtime,
+            "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"),
     ):
-        _authenticate(authorization, "admin")
-        _available_workspace()
+        _authorise(
+            authorization,
+            namespace=ns,
+            required_scopes=("bench:read",),
+        )
         return {
             "service": "gdw-frontier",
             "implementation_status": "REAL",
@@ -469,8 +550,11 @@ def register(app, ns: str = "a11oy"):
     def gdw_metrics(
         authorization: Optional[str] = Header(default=None, alias="Authorization"),
     ):
-        _authenticate(authorization, "admin")
-        _available_workspace()
+        _authorise(
+            authorization,
+            namespace=ns,
+            required_scopes=("metrics:read",),
+        )
         return PlainTextResponse(
             _TELEMETRY.render(),
             media_type="text/plain; version=0.0.4; charset=utf-8",
@@ -481,9 +565,45 @@ def register(app, ns: str = "a11oy"):
     def gdw_integrity(
         authorization: Optional[str] = Header(default=None, alias="Authorization"),
     ):
-        _authenticate(authorization, "admin")
-        workspace = _available_workspace()
-        return workspace.integrity()
+        principal = _authorise(
+            authorization,
+            namespace=ns,
+            required_scopes=("integrity:read",),
+        )
+        return _workspace(principal).integrity()
+
+    @app.post(prefix + "/drain")
+    @app.post("/v1/gdw/drain")
+    def gdw_drain(
+        limit: int = 100,
+        authorization: Optional[str] = Header(default=None, alias="Authorization"),
+    ):
+        principal = _authorise(
+            authorization,
+            namespace=ns,
+            required_scopes=("integrity:global",),
+        )
+        _require_write_ready(ns)
+        report = drain_once(limit=limit)
+        integrity = _workspace(principal).integrity(global_scope=True)
+        return {
+            "schema": "szl.gdw.drain-report/v1",
+            **report,
+            "integrity_ok": integrity["ok"],
+            "database_generation_id": integrity["database_generation_id"],
+        }
+
+    @app.get(prefix + "/integrity/global")
+    @app.get("/v1/gdw/integrity/global")
+    def gdw_global_integrity(
+        authorization: Optional[str] = Header(default=None, alias="Authorization"),
+    ):
+        principal = _authorise(
+            authorization,
+            namespace=ns,
+            required_scopes=("integrity:global",),
+        )
+        return _workspace(principal).integrity(global_scope=True)
 
     @app.get(prefix + "/sessions/{session_id}")
     @app.get("/v1/gdw/sessions/{session_id}")
@@ -491,57 +611,62 @@ def register(app, ns: str = "a11oy"):
         session_id: str,
         authorization: Optional[str] = Header(default=None, alias="Authorization"),
     ):
-        principal = _authenticate(authorization)
-        workspace = _available_workspace()
+        principal = _authorise(
+            authorization,
+            namespace=ns,
+            required_scopes=("session:read",),
+        )
         if not _ID_PATTERN.fullmatch(session_id):
             raise HTTPException(status_code=422, detail="invalid session_id")
-        try:
-            state = workspace.read_session(session_id, principal["principal_id"])
-        except PermissionError as exc:
-            raise HTTPException(status_code=403, detail=str(exc)) from exc
+        state = _workspace(principal).read_session(session_id)
         if state is None:
             raise HTTPException(status_code=404, detail="session not found")
         return state
 
-    @app.post(prefix + "/step", openapi_extra=_step_openapi())
-    @app.post("/v1/gdw/step", openapi_extra=_step_openapi())
+    request_body_contract = {
+        "requestBody": {
+            "required": True,
+            "content": {
+                "application/json": {
+                    "schema": _model_schema(GDWStepRequest),
+                }
+            },
+        }
+    }
+
+    @app.post(prefix + "/step", openapi_extra=request_body_contract)
+    @app.post("/v1/gdw/step", openapi_extra=request_body_contract)
     async def gdw_step(
         request: Request,
         authorization: Optional[str] = Header(default=None, alias="Authorization"),
         x_request_id: Optional[str] = Header(default=None, alias="X-Request-Id"),
     ):
         started = time.perf_counter()
-        principal = _authenticate(authorization)
-        workspace = _available_workspace()
-        principal_id = principal["principal_id"]
+        principal = _authorise(
+            authorization,
+            namespace=ns,
+            required_scopes=("step:write",),
+        )
+        _require_write_ready(ns)
         try:
             raw_payload = await request.json()
-            if hasattr(GDWStepRequest, "model_validate"):
-                payload = GDWStepRequest.model_validate(raw_payload)
-            else:
-                payload = GDWStepRequest.parse_obj(raw_payload)
-        except (ValueError, TypeError, ValidationError) as exc:
-            detail = (
-                exc.errors()
-                if isinstance(exc, ValidationError)
-                else "request body must be valid JSON"
-            )
-            raise HTTPException(status_code=422, detail=detail) from exc
+        except Exception as exc:
+            raise HTTPException(
+                status_code=422,
+                detail="invalid GDW step request",
+            ) from exc
+        payload = _validate_step_payload(raw_payload)
         request_id = _validate_identifiers(payload, x_request_id)
         payload_data = _dump_model(payload)
         request_digest = _sha(payload_data)
-        generation_id = workspace.generation_id()
-        limits = _admission_limits()
-        retention_seconds = _retention_seconds()
         selected_mode = "unresolved"
         decision = "ERROR"
         receipt_hash = ""
 
         try:
+            workspace = _workspace(principal)
             with workspace.transaction() as connection:
-                cached = workspace.cached_request(
-                    connection, request_id, principal_id
-                )
+                cached = workspace.cached_request(connection, request_id)
                 if cached is not None:
                     cached_digest, cached_response = cached
                     if cached_digest != request_digest:
@@ -549,6 +674,18 @@ def register(app, ns: str = "a11oy"):
                             status_code=409,
                             detail="X-Request-Id was already used with different content",
                         )
+                    current_bundle = _policy_bundle_sha256()
+                    cached_bundle = (
+                        cached_response.get("audit", {})
+                        .get("governance", {})
+                        .get("colang", {})
+                        .get("bundle_sha256")
+                    )
+                    if not current_bundle or cached_bundle != current_bundle:
+                        raise HTTPException(
+                            status_code=409,
+                            detail="policy snapshot changed; replay refused",
+                        )
                     cached_response["replayed"] = True
                     selected_mode = cached_response["scheduler_mode"]
                     decision = cached_response["decision"]
@@ -566,18 +703,14 @@ def register(app, ns: str = "a11oy"):
                     before_step = 0
                     before_hash = _sha(
                         {
+                            "namespace": principal.namespace,
+                            "owner_id": principal.owner_id,
                             "session_id": payload.session_id,
                             "step": 0,
                             "state": "GENESIS",
                         }
                     )
                 else:
-                    workspace.require_object_owner(
-                        connection,
-                        "session",
-                        payload.session_id,
-                        principal_id,
-                    )
                     before_step = previous["step"]
                     before_hash = previous["state_hash"]
 
@@ -598,48 +731,34 @@ def register(app, ns: str = "a11oy"):
                 selected_mode = routing["mode"]
                 precondition_decision = _decision(payload)
                 governance = _governance_gate(
-                    payload_data,
-                    request_id,
-                    request_digest,
-                    principal_id,
+                    payload_data, request_id, request_digest, principal
                 )
                 decision = precondition_decision
                 if decision == "ACCEPT" and not governance["allowed"]:
                     decision = "REJECT"
                 mutates = decision == "ACCEPT" and not payload.dry_run
                 step = before_step + 1 if mutates else before_step
-                timestamp = _now()
-                expires_at = (
-                    datetime.now(timezone.utc)
-                    + timedelta(seconds=retention_seconds)
-                ).isoformat()
-                workspace.admit_request(
-                    connection,
-                    owner_id=principal_id,
-                    request_id=request_id,
-                    session_id=payload.session_id,
-                    mutates=mutates,
-                    created_at=timestamp,
-                    expires_at=expires_at,
-                    limits=limits,
-                )
+                database_generation_id = workspace.database_generation_id
                 proposal_id = sha256_json(
                     {
                         "schema": "szl.gdw.proposal-identity/v1",
-                        "generation_id": generation_id,
-                        "owner_id": principal_id,
+                        "database_generation_id": database_generation_id,
+                        "namespace": principal.namespace,
+                        "owner_id": principal.owner_id,
                         "request_id": request_id,
                         "request_digest": request_digest,
                         "state_before_hash": before_hash,
                         "governance_evidence_sha256": sha256_json(governance),
                     }
                 )
+                timestamp = _now()
 
                 if mutates:
                     state = {
+                        "namespace": principal.namespace,
+                        "owner_id": principal.owner_id,
                         "session_id": payload.session_id,
-                        "owner_id": principal_id,
-                        "generation_id": generation_id,
+                        "database_generation_id": database_generation_id,
                         "step": step,
                         "previous_state_hash": before_hash,
                         "request_digest": request_digest,
@@ -659,14 +778,14 @@ def register(app, ns: str = "a11oy"):
                         proposal_id=proposal_id,
                         request_id=request_id,
                         request_digest=request_digest,
-                        owner_id=principal_id,
-                        generation_id=generation_id,
                         session_id=payload.session_id,
                         step=step,
                         before_hash=before_hash,
                         after_hash=after_hash,
                         scheduler_mode=selected_mode,
                         governance=governance,
+                        principal=principal,
+                        database_generation_id=database_generation_id,
                         timestamp=timestamp,
                     )
                     receipt_hash = receipt["receipt_hash"]
@@ -679,8 +798,9 @@ def register(app, ns: str = "a11oy"):
                     proposal_id=proposal_id,
                     request_id=request_id,
                     request_digest=request_digest,
-                    owner_id=principal_id,
-                    generation_id=generation_id,
+                    namespace=principal.namespace,
+                    owner_id=principal.owner_id,
+                    database_generation_id=database_generation_id,
                     step=step,
                     before_hash=before_hash,
                     after_hash=after_hash,
@@ -698,21 +818,16 @@ def register(app, ns: str = "a11oy"):
                         "GDW_PROOF_EXPORT_MODE must be 'outbox'; "
                         "synchronous external effects are not transaction-safe"
                     )
-                proof_payload_digest = sha256_json(proof_payload)
-                proof_effect_key = _effect_key(
-                    generation_id=generation_id,
-                    owner_id=principal_id,
-                    request_id=request_id,
-                    request_digest=request_digest,
-                    kind="proof_export",
-                    canonical_identity=proof_payload["payload_sha256"],
-                    payload_sha256=proof_payload_digest,
-                )
                 proof_artifact = {
                     "status": "OUTBOX_PENDING",
                     "kind": "proof_export",
-                    "idempotency_key": proof_effect_key,
-                    "canonical_identity": proof_payload["payload_sha256"],
+                    "idempotency_key": workspace.scoped_effect_key(
+                        principal.namespace,
+                        principal.owner_id,
+                        request_id,
+                        "proof_export",
+                        proof_payload["payload_sha256"],
+                    ),
                     "payload_sha256": proof_payload["payload_sha256"],
                     "formal_status": "NOT_RUN",
                 }
@@ -726,9 +841,14 @@ def register(app, ns: str = "a11oy"):
                     "benchmark_status": "UNMEASURED",
                     "proposal_id": proposal_id,
                     "request_id": request_id,
-                    "owner_id": principal_id,
-                    "generation_id": generation_id,
+                    "request_digest": request_digest,
+                    "database_generation_id": database_generation_id,
                     "session_id": payload.session_id,
+                    "principal": {
+                        "owner_id": principal.owner_id,
+                        "namespace": principal.namespace,
+                        "key_id": principal.key_id,
+                    },
                     "decision": decision,
                     "step": step,
                     "state_hash": after_hash,
@@ -740,10 +860,6 @@ def register(app, ns: str = "a11oy"):
                     "kernel_execution": "NOT_EXECUTED_BY_CONTROL_API",
                     "dry_run": payload.dry_run,
                     "replayed": False,
-                    "retention": {
-                        "expires_at": expires_at,
-                        "seconds": retention_seconds,
-                    },
                     "audit": {
                         "governance": governance,
                         "precondition_decision": precondition_decision,
@@ -767,6 +883,7 @@ def register(app, ns: str = "a11oy"):
                     timestamp,
                 )
                 if receipt is not None:
+                    receipt_payload_sha256 = sha256_json(receipt)
                     workspace.save_receipt(
                         connection,
                         receipt_hash,
@@ -776,38 +893,34 @@ def register(app, ns: str = "a11oy"):
                         receipt,
                         timestamp,
                     )
-                    receipt_payload_digest = sha256_json(receipt)
-                    receipt_effect_key = _effect_key(
-                        generation_id=generation_id,
-                        owner_id=principal_id,
-                        request_id=request_id,
-                        request_digest=request_digest,
-                        kind="receipt_projection",
-                        canonical_identity=receipt_hash,
-                        payload_sha256=receipt_payload_digest,
-                    )
                     workspace.save_effect_outbox(
                         connection,
                         request_id,
                         "receipt_projection",
-                        generation_id,
-                        principal_id,
-                        receipt_hash,
                         receipt,
-                        receipt_payload_digest,
-                        receipt_effect_key,
+                        receipt_payload_sha256,
+                        workspace.scoped_effect_key(
+                            principal.namespace,
+                            principal.owner_id,
+                            request_id,
+                            "receipt_projection",
+                            receipt_payload_sha256,
+                        ),
                         timestamp,
                     )
                 workspace.save_effect_outbox(
                     connection,
                     request_id,
                     "proof_export",
-                    generation_id,
-                    principal_id,
-                    proof_payload["payload_sha256"],
                     proof_payload,
-                    proof_payload_digest,
-                    proof_effect_key,
+                    proof_payload["payload_sha256"],
+                    workspace.scoped_effect_key(
+                        principal.namespace,
+                        principal.owner_id,
+                        request_id,
+                        "proof_export",
+                        proof_payload["payload_sha256"],
+                    ),
                     timestamp,
                 )
 
@@ -820,10 +933,21 @@ def register(app, ns: str = "a11oy"):
             return response
         except HTTPException:
             raise
-        except PermissionError as exc:
-            raise HTTPException(status_code=403, detail=str(exc)) from exc
-        except OverflowError as exc:
-            raise HTTPException(status_code=429, detail=str(exc)) from exc
+        except GDWQuotaExceeded as exc:
+            raise HTTPException(
+                status_code=429,
+                detail="GDW quota exceeded",
+            ) from exc
+        except GDWLifecycleError as exc:
+            raise HTTPException(
+                status_code=409,
+                detail="GDW object is outside its active lifecycle",
+            ) from exc
+        except GDWConfigurationError as exc:
+            raise HTTPException(
+                status_code=503,
+                detail="GDW durable workspace is unavailable",
+            ) from exc
         except Exception as exc:
             _TELEMETRY.observe(
                 (time.perf_counter() - started) * 1000.0,
@@ -839,12 +963,13 @@ def register(app, ns: str = "a11oy"):
 
     return {
         "ok": True,
-        "state": "CONFIGURATION_GATED",
+        "state": "REAL",
         "routes": [
             prefix + "/healthz",
             prefix + "/bench/meta",
             prefix + "/metrics",
             prefix + "/integrity",
+            prefix + "/integrity/global",
             prefix + "/drain",
             prefix + "/sessions/{session_id}",
             prefix + "/step",

serve.py

@@ -552,8 +552,9 @@ 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.
+# structured theorem inputs, and honest benchmark metadata. Runtime writes are
+# REAL only when the route's full readiness predicate passes; otherwise the
+# surface is UNAVAILABLE. Benchmark claims remain UNMEASURED until a captured run.
 # Registered before both catch-alls.
 try:
     from routers import gdw_frontier as _gdw_frontier

szl_colang_policy.py

@@ -36,7 +36,7 @@ import json
 import os
 import re
 from pathlib import Path
-from typing import Any, Optional
+from typing import Optional
 
 # Candidate locations for the policy dir (image: /app/policy/colang; repo-relative).
 _POLICY_DIRS = [
@@ -49,7 +49,19 @@ _POLICY_DIRS = [
 _FLOW_RE = re.compile(r"^define\s+flow\s+([A-Za-z0-9_]+)\s*$")
 _REASON_RE = re.compile(r'with\s+reason\s+"([^"]+)"')
 _PRED_RE = re.compile(r"\b([a-z_][a-z0-9_]*)\s*\(\s*\$action\s*\)")
-
+_CONTRACT_NAME = "enforcement-contract.json"
+_EVALUATOR_REGION_BEGIN = "# BEGIN EXACT POLICY EVALUATOR CONTRACT"
+_EVALUATOR_REGION_END = "# END EXACT POLICY EVALUATOR CONTRACT"
+
+# BEGIN EXACT POLICY EVALUATOR CONTRACT
+_CANONICAL_POLICY_FILES = {
+    "killinchu_threat.co": (
+        "3c47dfd730d738f1ce0b722b5ccc21fb12dda70ccb288043ffe10b78479df9c2"
+    ),
+    "roe_core.co": (
+        "77598deb89c4bebeb3682b30a0a01fcebc73a089183fa0a9a2877ad1fd732984"
+    ),
+}
 # Threat / injection signatures the policy predicates actually scan for. Mirrors
 # the in-image arena gate so policy enforcement is consistent across surfaces.
 _INJECTION_SIGS = ["ignore previous", "ignore policy", "exfiltrate", "override",
@@ -61,9 +73,6 @@ _PII_SIGS = ["ssn", "social security", "card number", "full card", "pan ",
 _EFFECTOR_TOOLS = {"issue_refund", "send_email", "reset_password", "assign_seat",
                    "apply_change", "run_sql", "engage", "release", "execute"}
 _PAYLOAD_CEILING = 1_000_000
-_ENFORCEMENT_CONTRACT_SCHEMA = "szl.colang-enforcement-contract/v1"
-_ENFORCEMENT_EVALUATOR = "szl.colang-python-evaluator/v1"
-_ENFORCEMENT_CONTRACT_FILENAME = "gdw_enforcement_contract.json"
 
 
 def _resolve_dir() -> Optional[Path]:
@@ -213,6 +222,57 @@ _FLOW_LOGIC = {
     "require_sensor_quorum":
         lambda a: _is_threat_decision(a) and not _sensor_quorum_met(a),
 }
+_FLOW_GUARDS = {
+    "refuse_destructive_actions": [
+        "is_destructive",
+        "has_operator_authorization",
+    ],
+    "refuse_pii_exfiltration": ["requests_pii_exfiltration"],
+    "refuse_prompt_injection": ["matches_injection_signature"],
+    "require_operator_approval_high_impact": [
+        "is_high_impact",
+        "has_operator_approval_event",
+    ],
+    "enforce_payload_ceiling": ["payload_exceeds_ceiling"],
+    "policy_before_effect": ["is_effecting", "policy_evaluated_before"],
+    "no_autonomous_engagement": [
+        "is_engagement",
+        "has_human_authorization",
+    ],
+    "require_calibrated_classifier": [
+        "is_automated_response",
+        "classifier_calibration_gate_pass",
+    ],
+    "require_singleton_conformal_set": [
+        "is_automated_response",
+        "conformal_set_ambiguous",
+    ],
+    "require_sensor_quorum": ["is_threat_decision", "sensor_quorum_met"],
+}
+
+
+def _load_enforcement_contract(
+    directory: Path,
+    trusted_policy_files: dict[str, str],
+) -> dict:
+    path = directory / _CONTRACT_NAME
+    try:
+        contract = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError) as exc:
+        raise ValueError("policy enforcement contract is unavailable") from exc
+    if contract.get("schema") != "szl.colang-enforcement-contract/v1":
+        raise ValueError("policy enforcement contract schema is unsupported")
+    expected_files = contract.get("policy_files")
+    if not isinstance(expected_files, dict) or not expected_files:
+        raise ValueError("policy enforcement contract has no policy file bindings")
+    expected_contract = {
+        "schema": "szl.colang-enforcement-contract/v1",
+        "evaluator_region_sha256": _LOADED_EVALUATOR_SHA256,
+        "policy_files": trusted_policy_files,
+    }
+    if contract != expected_contract:
+        raise ValueError("policy enforcement contract is not rooted in trusted source")
+    return contract
 
 
 def _parse_flows(text: str) -> list[dict]:
@@ -233,13 +293,41 @@ def _parse_flows(text: str) -> list[dict]:
         rm = _REASON_RE.search(line)
         if rm and not current["reason"]:
             current["reason"] = rm.group(1)
-        for pm in _PRED_RE.finditer(line):
-            g = pm.group(1)
-            if g not in current["guards"]:
-                current["guards"].append(g)
+        if line.strip().startswith("if "):
+            for pm in _PRED_RE.finditer(line):
+                g = pm.group(1)
+                if g not in current["guards"]:
+                    current["guards"].append(g)
     if current:
         flows.append(current)
     return flows
+# END EXACT POLICY EVALUATOR CONTRACT
+
+
+def _evaluator_region_sha256() -> str:
+    """Hash the exact reviewed evaluator/parser region from loaded source."""
+
+    source = Path(__file__).read_text(encoding="utf-8").replace("\r\n", "\n")
+    pattern = re.compile(
+        r"^# BEGIN EXACT POLICY EVALUATOR CONTRACT\r?\n"
+        r"(?P<region>.*?)"
+        r"^# END EXACT POLICY EVALUATOR CONTRACT$",
+        re.MULTILINE | re.DOTALL,
+    )
+    match = pattern.search(source)
+    if match is None:
+        raise ValueError("policy evaluator contract markers are missing")
+    region = match.group("region")
+    if (
+        "_FLOW_LOGIC" not in region
+        or "def _matches_injection_signature" not in region
+        or "def _parse_flows" not in region
+    ):
+        raise ValueError("policy evaluator contract region is incomplete")
+    return hashlib.sha256(region.encode("utf-8")).hexdigest()
+
+
+_LOADED_EVALUATOR_SHA256 = _evaluator_region_sha256()
 
 
 _NEMO_AVAILABLE = False
@@ -253,19 +341,43 @@ except Exception:
 class ColangPolicy:
     """Loaded, file-backed, auditable Colang policy set."""
 
-    def __init__(self, directory: Optional[Path] = None) -> None:
+    def __init__(
+        self,
+        directory: Optional[Path] = None,
+        *,
+        trusted_policy_files: Optional[dict[str, str]] = None,
+    ) -> None:
         self.directory = directory or _resolve_dir()
+        self.trusted_policy_files = dict(
+            trusted_policy_files or _CANONICAL_POLICY_FILES
+        )
         self.files: list[dict] = []
+        self.contract: Optional[dict] = None
+        self.contract_errors: list[str] = []
+        self.validation_errors: list[str] = []
+        self._bundle_sha256: Optional[str] = None
         self._load()
 
     def _load(self) -> None:
         self.files = []
+        self.contract = None
+        self.contract_errors = []
+        self.validation_errors = []
+        self._bundle_sha256 = None
         if not self.directory:
+            self.contract_errors = ["policy_directory_unavailable"]
             return
+        try:
+            self.contract = _load_enforcement_contract(
+                self.directory,
+                self.trusted_policy_files,
+            )
+        except ValueError:
+            self.contract_errors = ["enforcement_contract_unavailable"]
         for p in sorted(self.directory.glob("*.co")):
             try:
                 raw = p.read_bytes()
-                text = raw.decode("utf-8", "replace")
+                text = raw.decode("utf-8", "strict")
                 flows = _parse_flows(text)
                 pid = None
                 pver = None
@@ -285,13 +397,91 @@ class ColangPolicy:
                     "flows": flows,
                     "content": text,
                 })
-            except Exception:
-                continue
+            except (OSError, UnicodeDecodeError):
+                self.validation_errors.append(f"policy_file_unreadable:{p.name}")
+        self.validation_errors.extend(self._flow_validation_errors())
+        self.validation_errors = sorted(set(self.validation_errors))
+        self.contract_errors.extend(self._snapshot_contract_errors())
+        self.contract_errors = sorted(set(self.contract_errors))
+        if not self.contract_errors and not self.validation_errors and self.contract:
+            self._bundle_sha256 = hashlib.sha256(
+                json.dumps(
+                    self.contract,
+                    sort_keys=True,
+                    separators=(",", ":"),
+                ).encode("utf-8")
+            ).hexdigest()
+
+    def _flow_validation_errors(self) -> list[str]:
+        errors = []
+        observed_names = []
+        for source in self.files:
+            if not source["flows"]:
+                errors.append(f"policy_file_has_no_flows:{source['name']}")
+            if not source["policy_id"]:
+                errors.append(f"policy_id_missing:{source['name']}")
+            if not source["policy_version"]:
+                errors.append(f"policy_version_missing:{source['name']}")
+            for flow in source["flows"]:
+                name = flow["name"]
+                observed_names.append(name)
+                if not flow["reason"]:
+                    errors.append(f"flow_reason_missing:{name}")
+                expected_guards = _FLOW_GUARDS.get(name)
+                if expected_guards is None:
+                    errors.append(f"unsupported_flow:{name}")
+                elif flow["guards"] != expected_guards:
+                    errors.append(f"flow_guard_mismatch:{name}")
+        duplicates = {
+            name for name in observed_names if observed_names.count(name) > 1
+        }
+        errors.extend(f"duplicate_flow:{name}" for name in sorted(duplicates))
+        return errors
+
+    def _snapshot_contract_errors(self) -> list[str]:
+        """Validate the immutable bytes loaded into this policy snapshot."""
+        if not self.directory or not self.contract:
+            return self.contract_errors or ["enforcement_contract_unavailable"]
+        errors: list[str] = []
+        expected_files = self.contract["policy_files"]
+        observed_names = {source["name"] for source in self.files}
+        if observed_names != set(expected_files):
+            errors.append("policy_file_set_mismatch")
+        observed_hashes = {
+            source["name"]: source["sha256"] for source in self.files
+        }
+        for name, expected_sha in sorted(expected_files.items()):
+            if observed_hashes.get(name) != expected_sha:
+                errors.append(f"policy_file_digest_mismatch:{name}")
+        return sorted(set(errors))
 
     @property
     def loaded(self) -> bool:
         return bool(self.files)
 
+    @property
+    def unsupported_flows(self) -> list[str]:
+        return sorted(
+            {
+                flow["name"]
+                for flow in self.all_flows()
+                if flow["name"] not in _FLOW_LOGIC
+            }
+        )
+
+    @property
+    def enforcement_ready(self) -> bool:
+        return (
+            self.loaded
+            and not self.unsupported_flows
+            and not self.contract_errors
+            and not self.validation_errors
+        )
+
+    @property
+    def bundle_sha256(self) -> Optional[str]:
+        return self._bundle_sha256
+
     def all_flows(self) -> list[dict]:
         out = []
         for f in self.files:
@@ -300,165 +490,90 @@ class ColangPolicy:
                             "policy_version": f["policy_version"]})
         return out
 
-    def enforcement_contract_status(self) -> dict:
-        """Bind the hard-coded evaluator to exact reviewed Colang source bytes.
-
-        The legacy evaluator dispatches by flow name. It therefore cannot safely
-        interpret a new flow or changed Colang expression. Strict callers must
-        fail closed unless the loaded files exactly match this reviewed contract
-        and every declared flow has a local evaluator.
-        """
-        reasons: list[str] = []
-        contract_path = (
-            self.directory / _ENFORCEMENT_CONTRACT_FILENAME
-            if self.directory
-            else None
-        )
-        contract: dict[str, Any] = {}
-        if contract_path is None or not contract_path.is_file():
-            reasons.append("ENFORCEMENT_CONTRACT_MISSING")
-        else:
-            try:
-                contract = json.loads(contract_path.read_text(encoding="utf-8"))
-            except Exception:
-                reasons.append("ENFORCEMENT_CONTRACT_INVALID")
-
-        if contract.get("schema") != _ENFORCEMENT_CONTRACT_SCHEMA:
-            reasons.append("ENFORCEMENT_CONTRACT_SCHEMA_MISMATCH")
-        if contract.get("evaluator") != _ENFORCEMENT_EVALUATOR:
-            reasons.append("ENFORCEMENT_EVALUATOR_MISMATCH")
-
-        expected_files = contract.get("files")
-        if not isinstance(expected_files, dict):
-            expected_files = {}
-            reasons.append("ENFORCEMENT_FILE_LOCK_INVALID")
-        actual_files = {item["name"]: item["sha256"] for item in self.files}
-        if expected_files != actual_files:
-            reasons.append("ENFORCEMENT_FILE_LOCK_MISMATCH")
-
-        unsupported_flows = sorted(
-            flow["name"]
-            for flow in self.all_flows()
-            if flow["name"] not in _FLOW_LOGIC
+    def evaluate(self, action: dict) -> dict:
+        """Evaluate a proposed action against EVERY loaded flow. Returns which
+        flows fired (rule violated -> refuse) and the overall allow/deny. This is
+        the policy layer; serve.py calls it before signing an action receipt."""
+        action = action or {}
+        fired: list[dict] = []
+        evaluated: list[str] = []
+        unsupported: list[str] = []
+        evaluation_errors: list[str] = []
+        source_errors = sorted(
+            set(self.contract_errors + self.validation_errors)
         )
-        if unsupported_flows:
-            reasons.append("UNSUPPORTED_POLICY_FLOW")
-
-        return {
-            "valid": not reasons,
-            "schema": _ENFORCEMENT_CONTRACT_SCHEMA,
-            "evaluator": _ENFORCEMENT_EVALUATOR,
-            "contract_path": str(contract_path) if contract_path else None,
-            "contract_sha256": (
-                hashlib.sha256(contract_path.read_bytes()).hexdigest()
-                if contract_path and contract_path.is_file()
-                else None
-            ),
-            "reason_codes": sorted(set(reasons)),
-            "unsupported_flows": unsupported_flows,
-            "files": actual_files,
-        }
-
-    def evaluate_strict(self, action: dict) -> dict:
-        """Evaluate only an exact, supported, byte-locked policy generation."""
-        enforcement = self.enforcement_contract_status()
-        if not enforcement["valid"]:
+        if source_errors:
             return {
                 "allow": False,
                 "decision": "deny",
                 "fired_flows": [
                     {
-                        "flow": "strict_enforcement_contract",
-                        "reason": reason,
-                        "file": _ENFORCEMENT_CONTRACT_FILENAME,
-                        "policy_id": "GDW_STRICT",
-                        "policy_version": _ENFORCEMENT_EVALUATOR,
+                        "flow": "exact_source_contract",
+                        "reason": (
+                            "POLICY_SOURCE_DRIFT"
+                            if self.contract_errors
+                            else "POLICY_SOURCE_INVALID"
+                        ),
+                        "file": _CONTRACT_NAME,
+                        "policy_id": "szl-colang-exact-source",
+                        "policy_version": "1",
                     }
-                    for reason in enforcement["reason_codes"]
                 ],
-                "fired_count": len(enforcement["reason_codes"]),
-                "flows_evaluated": [
-                    flow["name"] for flow in self.all_flows()
-                ],
-                "matched_count": len(enforcement["reason_codes"]),
+                "fired_count": 1,
+                "flows_evaluated": [],
+                "unsupported_flows": self.unsupported_flows,
+                "evaluation_errors": [],
+                "validation_errors": list(self.validation_errors),
+                "source_contract_errors": source_errors,
+                "bundle_sha256": None,
+                "enforcement_ready": False,
+                "matched_count": 1,
                 "policy_files": [
                     {
-                        "name": item["name"],
-                        "sha256": item["sha256"],
-                        "policy_id": item["policy_id"],
-                        "policy_version": item["policy_version"],
+                        "name": f["name"],
+                        "sha256": f["sha256"],
+                        "policy_id": f["policy_id"],
+                        "policy_version": f["policy_version"],
                     }
-                    for item in self.files
+                    for f in self.files
                 ],
-                "enforcement_contract": enforcement,
+                "honesty": (
+                    "The exact reviewed policy/evaluator source contract did not "
+                    "match current on-disk bytes; enforcement failed closed."
+                ),
             }
-        action = action or {}
-        fired: list[dict] = []
-        evaluated: list[str] = []
-        evaluator_errors: list[str] = []
-        for flow in self.all_flows():
-            name = flow["name"]
-            logic = _FLOW_LOGIC[name]
-            evaluated.append(name)
-            try:
-                violated = bool(logic(action))
-            except Exception:
-                violated = True
-                evaluator_errors.append(name)
-            if violated:
-                fired.append({
-                    "flow": name,
-                    "reason": (
-                        "POLICY_EVALUATOR_ERROR"
-                        if name in evaluator_errors
-                        else flow.get("reason") or name
-                    ),
-                    "file": flow["file"],
-                    "policy_id": flow["policy_id"],
-                    "policy_version": flow["policy_version"],
-                })
-        allow = not fired
-        return {
-            "allow": allow,
-            "decision": "allow" if allow else "deny",
-            "fired_flows": fired,
-            "fired_count": len(fired),
-            "flows_evaluated": evaluated,
-            "matched_count": len(fired),
-            "policy_files": [
-                {
-                    "name": item["name"],
-                    "sha256": item["sha256"],
-                    "policy_id": item["policy_id"],
-                    "policy_version": item["policy_version"],
-                }
-                for item in self.files
-            ],
-            "evaluator_errors": evaluator_errors,
-            "enforcement_contract": enforcement,
-        }
-
-    def evaluate(self, action: dict) -> dict:
-        """Evaluate a proposed action against EVERY loaded flow. Returns which
-        flows fired (rule violated -> refuse) and the overall allow/deny. This is
-        the policy layer; serve.py calls it before signing an action receipt."""
-        action = action or {}
-        fired: list[dict] = []
-        evaluated: list[str] = []
         for fl in self.all_flows():
             name = fl["name"]
             logic = _FLOW_LOGIC.get(name)
             evaluated.append(name)
             if logic is None:
+                unsupported.append(name)
+                fired.append({
+                    "flow": name,
+                    "reason": "UNSUPPORTED_FLOW_FAIL_CLOSED",
+                    "file": fl["file"],
+                    "policy_id": fl["policy_id"],
+                    "policy_version": fl["policy_version"],
+                })
                 continue
             try:
                 violated = bool(logic(action))
             except Exception:
-                violated = False
+                violated = True
+                evaluation_errors.append(name)
             if violated:
-                fired.append({"flow": name, "reason": fl.get("reason") or name,
-                              "file": fl["file"], "policy_id": fl["policy_id"],
-                              "policy_version": fl["policy_version"]})
+                reason = (
+                    "FLOW_EVALUATION_ERROR"
+                    if name in evaluation_errors
+                    else fl.get("reason") or name
+                )
+                fired.append({
+                    "flow": name,
+                    "reason": reason,
+                    "file": fl["file"],
+                    "policy_id": fl["policy_id"],
+                    "policy_version": fl["policy_version"],
+                })
         allow = len(fired) == 0
         return {
             "allow": allow,
@@ -466,6 +581,11 @@ class ColangPolicy:
             "fired_flows": fired,
             "fired_count": len(fired),
             "flows_evaluated": evaluated,
+            "unsupported_flows": sorted(unsupported),
+            "evaluation_errors": sorted(evaluation_errors),
+            "source_contract_errors": [],
+            "bundle_sha256": self.bundle_sha256,
+            "enforcement_ready": not unsupported and not evaluation_errors,
             "matched_count": len(fired),
             "policy_files": [{"name": f["name"], "sha256": f["sha256"],
                               "policy_id": f["policy_id"],
@@ -484,6 +604,18 @@ class ColangPolicy:
             "nemoguardrails_runtime_present": _NEMO_AVAILABLE,
             "file_count": len(self.files),
             "flow_count": len(self.all_flows()),
+            "enforcement_ready": self.enforcement_ready,
+            "source_contract_errors": list(self.contract_errors),
+            "validation_errors": list(self.validation_errors),
+            "bundle_sha256": self.bundle_sha256,
+            "enforcement_contract": {
+                "schema": self.contract.get("schema"),
+                "evaluator_region_sha256": self.contract.get(
+                    "evaluator_region_sha256"
+                ),
+            }
+            if self.contract
+            else None,
             "files": self.files,
             "honesty": (
                 "Policy is FILE-BACKED and version-controlled: each rule is a "
@@ -498,14 +630,11 @@ class ColangPolicy:
         }
 
 
-_SINGLETON: Optional[ColangPolicy] = None
-
-
 def get_policy(reload: bool = False) -> ColangPolicy:
-    global _SINGLETON
-    if _SINGLETON is None or reload:
-        _SINGLETON = ColangPolicy()
-    return _SINGLETON
+    # Policy files are tiny. A fresh immutable snapshot avoids stale singleton
+    # state and binds evaluation, evidence, and bundle identity to one read.
+    del reload
+    return ColangPolicy()
 
 
 if __name__ == "__main__":  # pragma: no cover