"""Authenticated Governed Delta Workspace API and benchmark surfaces.""" import asyncio import hashlib import json import os import re import threading import time from datetime import datetime, timezone from typing import List, Literal, Optional from urllib.request import Request as UrlRequest from urllib.request import urlopen from fastapi import Header, HTTPException, Query, Request from fastapi.responses import PlainTextResponse from pydantic import BaseModel, Field, ValidationError from gdw_attention import AttentionFeatures, choose_attention_mode 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 ( 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"} _AUTH_LOCK = threading.RLock() _STEP_WRITE_LOCKS = tuple(threading.Lock() for _ in range(1024)) _POLICY_READINESS_LOCK = threading.Lock() _POLICY_READINESS_CACHE = { "origin": None, "checked_at": 0.0, "ready": False, } _AUTH_REGISTRY = None _AUTH_FINGERPRINT = None def _step_write_locks( namespace: str, owner_id: str, session_id: str, request_id: str, ) -> tuple[threading.Lock, ...]: indexes = set() for scope, value in (("session", session_id), ("request", request_id)): key = "\x00".join((scope, namespace, owner_id, value)).encode("utf-8") digest = hashlib.sha256(key).digest() indexes.add(int.from_bytes(digest[:4], "big") % len(_STEP_WRITE_LOCKS)) return tuple(_STEP_WRITE_LOCKS[index] for index in sorted(indexes)) async def _acquire_step_write_locks( locks: tuple[threading.Lock, ...], ) -> None: acquired = [] try: for lock in locks: while not lock.acquire(blocking=False): await asyncio.sleep(0.01) acquired.append(lock) except BaseException: for lock in reversed(acquired): lock.release() raise class GDWStepRequest(BaseModel): session_id: str = Field(min_length=1, max_length=128) request: str = Field(min_length=1, max_length=4096) allowed_experts: List[str] = Field(default_factory=list) risk_budget: float = Field(default=0.35, ge=0.0, le=1.0) mode_hint: Literal[ "auto", "kda_local", "laguna_hybrid", "mla_global" ] = "auto" dry_run: bool = False novelty: Optional[float] = Field(default=None, ge=0.0, le=1.0) disagreement: Optional[float] = Field(default=None, ge=0.0, le=1.0) context_tokens: int = Field(default=0, ge=0, le=1000000) active_tool_count: int = Field(default=0, ge=0, le=64) memory_pressure: Optional[float] = Field(default=None, ge=0.0, le=1.0) class Config: extra = "forbid" def _dump_model(model): if hasattr(model, "model_dump"): return model.model_dump(mode="json") return model.dict() def _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() def _sha(value) -> str: encoded = json.dumps( value, sort_keys=True, separators=(",", ":"), ensure_ascii=True ).encode("utf-8") return hashlib.sha256(encoded).hexdigest() 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() ) principal_namespace = os.environ.get("GDW_NAMESPACE") or "a11oy" legacy_token = os.environ.get("GDW_AUTH_TOKEN") if legacy_enabled else None legacy_owner_id = os.environ.get("GDW_OWNER_ID") if legacy_enabled else None legacy_namespace = os.environ.get("GDW_NAMESPACE") if legacy_enabled else None if not legacy_enabled: legacy_scopes = () fingerprint = _sha( { "registry": registry_json, "principal_registry": principal_registry_json, "principal_namespace": principal_namespace, "legacy_enabled": legacy_enabled, "legacy_token": legacy_token, "legacy_owner": legacy_owner_id, "legacy_namespace": legacy_namespace, "legacy_scopes": legacy_scopes, } ) 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=principal_namespace, legacy_enabled=legacy_enabled, legacy_token=legacy_token, legacy_owner_id=legacy_owner_id, legacy_namespace=legacy_namespace, legacy_scopes=legacy_scopes, ) _AUTH_REGISTRY = registry _AUTH_FINGERPRINT = fingerprint return registry def _authorise( authorization: Optional[str], *, namespace: str, required_scopes=(), ) -> Principal: try: return authenticate_bearer( authorization, _credential_registry(), namespace=namespace, required_scopes=required_scopes, ) except AuthConfigurationError as exc: raise HTTPException( status_code=503, detail="GDW credential registry is unavailable", ) from exc 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, ) def _governance_ready() -> bool: try: import szl_codename_gate import szl_colang_policy policy = szl_colang_policy.get_policy() return bool( policy.loaded and policy.enforcement_ready and callable(getattr(szl_codename_gate, "scan_text", None)) ) except Exception: return False 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") last_report = drain.get("last_report") if isinstance(last_report, dict) and ( last_report.get("failed") or last_report.get("pending_effects") or last_report.get("dead_letter_effects") or last_report.get("legacy_pending_proofs") or last_report.get("invalid_effect_bindings") or last_report.get("invalid_exported_artifacts") or last_report.get("sqlite_integrity") != "ok" ): blockers.append("OUTBOX_SUPERVISOR_NOT_QUIESCENT") 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") try: _policy_gateway_origin() except RuntimeError: blockers.append("CANONICAL_POLICY_GATEWAY_UNCONFIGURED") else: if not _canonical_policy_ready(): blockers.append("CANONICAL_POLICY_GATEWAY_UNAVAILABLE") return ( not blockers, sorted(set(blockers)), credential_count, runtime, governance_ready, ) def _require_write_ready(namespace: str) -> None: ready, blockers, _, _, _ = _write_readiness(namespace) if not ready: raise HTTPException( status_code=503, detail={ "reason": "GDW_WRITE_SURFACE_UNAVAILABLE", "write_blockers": blockers, }, ) def _require_transient_recovery_runtime( namespace: str, expected_source_revision: Optional[str], ) -> str: expected = str(expected_source_revision or "").strip().lower() observed_source = os.environ.get("SZL_GIT_SHA", "").strip().lower() if ( re.fullmatch(r"[0-9a-f]{40}", expected) is None or observed_source != expected ): raise HTTPException( status_code=409, detail="GDW recovery source revision mismatch", ) runtime = runtime_health() storage = runtime.get("storage") or {} drain = runtime.get("drain") or {} database_generation_id = str( storage.get("database_generation_id") or "" ) runtime_ready = ( runtime.get("startup_state") == "READY" and runtime.get("evidence_label") == "VERIFIED" and storage.get("persistence_required") is True and storage.get("mount_verified") is True and storage.get("journal_mode_requested") == "DELETE" and storage.get("journal_mode_observed") == "DELETE" and storage.get("synchronous_requested") == "FULL" and storage.get("synchronous_observed") == 2 and storage.get("sqlite_integrity") == "ok" and storage.get("proof_export_mode") == "outbox" and storage.get("schema_version") == GDWWorkspace.schema_version() and re.fullmatch( r"[0-9a-f]{32}", database_generation_id, ) is not None and drain.get("enabled") is True and drain.get("running") is True and _governance_ready() ) try: policy_ready = _canonical_policy_ready() except Exception: policy_ready = False if not runtime_ready or not policy_ready: raise HTTPException( status_code=503, detail="GDW recovery runtime contract is unavailable", ) return database_generation_id 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 } last_report = drain.get("last_report") if isinstance(last_report, dict): public_report = { key: last_report.get(key) for key in ( "attempted", "exported", "failed", "pending_effects", "claimed_effects", "dead_letter_effects", "legacy_pending_proofs", "sqlite_integrity", "invalid_effect_bindings", "invalid_exported_artifacts", ) if key in last_report } errors = last_report.get("errors") if isinstance(errors, list): public_report["errors"] = [ value for value in errors if isinstance(value, str) and len(value) <= 96 and re.fullmatch(r"[a-z_]+:[A-Za-z_]+", value) ] public_drain["last_report"] = public_report 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"), } def _validate_identifiers(payload: GDWStepRequest, request_id: Optional[str]) -> str: if not request_id or not _ID_PATTERN.fullmatch(request_id): raise HTTPException( status_code=422, detail="X-Request-Id must be 1-128 canonical identifier characters", ) if not _ID_PATTERN.fullmatch(payload.session_id): raise HTTPException(status_code=422, detail="invalid session_id") if len(payload.allowed_experts) > 16: raise HTTPException(status_code=422, detail="too many allowed_experts") return request_id def _decision(payload: GDWStepRequest) -> str: experts = set(payload.allowed_experts) if not experts or not experts.issubset(_EXPERTS): return "QUARANTINE" if payload.risk_budget >= 0.90: return "REJECT" if payload.risk_budget >= 0.75: return "QUARANTINE" return "ACCEPT" _LOOPBACK_POLICY_GATEWAY_ORIGIN = "http://127.0.0.1:7860" def _policy_gateway_origin() -> str: origin = os.environ.get("GDW_POLICY_ORIGIN", "").strip().rstrip("/") if ( origin != _LOOPBACK_POLICY_GATEWAY_ORIGIN and not origin.startswith("https://") ): raise RuntimeError( "GDW_POLICY_ORIGIN must be HTTPS or the exact local gateway" ) return origin def _policy_readiness_timeout() -> float: try: configured = float( os.environ.get("GDW_POLICY_READINESS_TIMEOUT_SECONDS", "3") ) except ValueError: configured = 3.0 return min(10.0, max(0.25, configured)) def _policy_readiness_ttl() -> float: try: configured = float( os.environ.get("GDW_POLICY_READINESS_TTL_SECONDS", "15") ) except ValueError: configured = 15.0 return min(300.0, max(1.0, configured)) def _policy_gateway_json(path: str) -> dict: request = UrlRequest( _policy_gateway_origin() + path, headers={"Accept": "application/json"}, method="GET", ) with urlopen(request, timeout=_policy_readiness_timeout()) as response: if response.status != 200: raise RuntimeError("canonical policy readiness returned non-200") result = json.loads(response.read().decode("utf-8")) if not isinstance(result, dict): raise RuntimeError("canonical policy readiness returned non-object JSON") return result def _canonical_policy_ready() -> bool: origin = _policy_gateway_origin() now = time.monotonic() with _POLICY_READINESS_LOCK: if ( _POLICY_READINESS_CACHE["origin"] == origin and now - float(_POLICY_READINESS_CACHE["checked_at"]) <= _policy_readiness_ttl() ): return bool(_POLICY_READINESS_CACHE["ready"]) ready = False try: health = _policy_gateway_json("/api/a11oy/v1/healthz") signing = _policy_gateway_json("/api/a11oy/v1/signing-status") backend = health.get("backend_health") backend_ready = ( str(health.get("status") or "").lower() in {"ok", "ready"} and ( not isinstance(backend, dict) or backend.get("alive") is True ) ) dsse_keyid = str(signing.get("dsse_keyid") or "") signer_ready = ( signing.get("key_persistent") is True and re.fullmatch(r"[0-9a-f]{16}", dsse_keyid) is not None ) ready = backend_ready and signer_ready except Exception: ready = False _POLICY_READINESS_CACHE.update( { "origin": origin, "checked_at": time.monotonic(), "ready": ready, } ) return ready def _canonical_policy_evaluate(action: dict) -> dict: request = UrlRequest( _policy_gateway_origin() + "/api/a11oy/v1/policy/evaluate", data=json.dumps( {"action": action}, sort_keys=True, separators=(",", ":"), ).encode("utf-8"), headers={"Content-Type": "application/json"}, method="POST", ) with urlopen(request, timeout=10) as response: if response.status != 200: raise RuntimeError("canonical policy gateway returned non-200") result = json.loads(response.read().decode("utf-8")) receipt_hash = str(result.get("receipt_hash") or "") if ( result.get("gate") != "ThresholdPolicySeverity" or len(receipt_hash) != 64 or any(ch not in "0123456789abcdef" for ch in receipt_hash) or result.get("receipt_signed") is not True or result.get("receipts_in_eq_out") is not True or result.get("receipt_error") ): raise RuntimeError("canonical policy response is not verifiable") return result def _transient_recovery_governance( *, principal: Principal, recovery_id: str, source_revision: str, database_generation_id: str, limit: int, ) -> dict: """Obtain an explicit signed canonical-policy allow for one recovery call.""" binding = { "schema": "szl.gdw.transient-effect-recovery-authorization/v1", "action_type": "gdw.transient-effect-recovery", "namespace": principal.namespace, "owner_id": principal.owner_id, "credential_key_id": principal.key_id, "recovery_id": recovery_id, "source_revision": source_revision, "database_generation_id": database_generation_id, "limit": limit, "failure_class": "hf-hard-link-enotsup/v1", } binding_sha256 = _sha(binding) action = { "actionId": f"gdw-recovery:{binding_sha256}", "severity": "high", "decisionClass": "ordinary", "confidence": 1.0, "witnesses": [ { "id": ( f"principal:{principal.namespace}:" f"{principal.owner_id}:{principal.key_id}" ), "role": "operator", "attested": True, }, { "id": f"workload:szl-holdings/a11oy@{source_revision}", "role": "workload", "attested": True, }, ], } result = _canonical_policy_evaluate(action) if result.get("decision") != "allow": raise PermissionError("canonical policy denied transient recovery") return { "schema": "szl.gdw.transient-effect-recovery-governance/v1", "decision": "ALLOW", "binding": binding, "binding_sha256": binding_sha256, "policy_gateway": { "decision": "ALLOW", "gate": result["gate"], "receipt_hash": result["receipt_hash"], "receipt_signed": True, "receipts_in_eq_out": True, "action_id": action["actionId"], "witnesses": action["witnesses"], }, } def _risk_severity(risk_budget: float) -> str: if risk_budget < 0.25: return "low" if risk_budget < 0.50: return "medium" if risk_budget < 0.75: return "high" if risk_budget < 0.90: return "critical" return "capital" async def _governance_gate( payload_data: dict, request_id: str, request_digest: str, principal: Principal, database_generation_id: str, state_before_hash: str, ) -> dict: action = { "tool": "execute", "effecting": True, "events": ["gate.evaluate"], "action_type": "gdw.step", "target": payload_data["session_id"], "request_id": request_id, "request_digest": request_digest, "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: 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, "decision": "DENY", "reason_codes": ["DOCTRINE_GATE_UNAVAILABLE"], "detail": type(exc).__name__, "writer_is_judge": False, "enforcement_mode": "LOCAL_PRECONDITIONS_PLUS_CANONICAL_POLICY_GATEWAY", "principal": { "owner_id": principal.owner_id, "namespace": principal.namespace, "key_id": principal.key_id, }, } try: import szl_codename_gate serialized = json.dumps( payload_data, sort_keys=True, separators=(",", ":"), ensure_ascii=True ) codename_hits = sorted( {str(value) for value in szl_codename_gate.scan_text(serialized)} ) except Exception as exc: return { "allowed": False, "decision": "DENY", "reason_codes": ["CODENAME_GATE_UNAVAILABLE"], "detail": type(exc).__name__, "writer_is_judge": False, "enforcement_mode": "LOCAL_PRECONDITIONS_PLUS_CANONICAL_POLICY_GATEWAY", "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", []), "bundle_sha256": colang.get("bundle_sha256"), }, } reasons = [] if not colang.get("allow"): reasons.append("DOCTRINE_POLICY_DENY") if codename_hits: reasons.append("CODENAME_POLICY_DENY") policy_gateway = None if not reasons: source_revision = os.environ.get("SZL_GIT_SHA", "").strip() if ( len(source_revision) != 40 or any(ch not in "0123456789abcdef" for ch in source_revision) ): reasons.append("RUNTIME_IDENTITY_UNAVAILABLE") else: binding = { "schema": "szl.gdw.authorization-binding/v1", "action_type": "gdw.step", "database_generation_id": database_generation_id, "namespace": principal.namespace, "owner_id": principal.owner_id, "credential_key_id": principal.key_id, "request_id": request_id, "request_digest": request_digest, "session_id": payload_data["session_id"], "state_before_hash": state_before_hash, } binding_sha256 = _sha(binding) gateway_action = { "actionId": f"gdw:{binding_sha256}", "severity": _risk_severity(float(payload_data["risk_budget"])), "decisionClass": "ordinary", "confidence": 1.0, "witnesses": [ { "id": ( f"principal:{principal.namespace}:" f"{principal.owner_id}:{principal.key_id}" ), "role": "operator", "attested": True, }, { "id": ( "workload:szl-holdings/a11oy@" f"{source_revision}" ), "role": "workload", "attested": True, }, ], } try: result = await asyncio.to_thread( _canonical_policy_evaluate, gateway_action, ) policy_gateway = { "decision": str(result.get("decision") or "").upper(), "gate": result["gate"], "receipt_hash": result["receipt_hash"], "receipt_signed": True, "receipts_in_eq_out": True, "binding_sha256": binding_sha256, "action_id": gateway_action["actionId"], "source_revision": source_revision, "witnesses": gateway_action["witnesses"], } if result.get("decision") != "allow": reasons.append("CANONICAL_POLICY_DENY") except Exception as exc: reasons.append("CANONICAL_POLICY_GATEWAY_UNAVAILABLE") policy_gateway = { "decision": "UNAVAILABLE", "detail": type(exc).__name__, } return { "allowed": not reasons, "decision": "ALLOW" if not reasons else "DENY", "reason_codes": reasons or [ "FILE_BACKED_GOVERNANCE_PASS", "CANONICAL_POLICY_GATEWAY_PASS", ], "writer_is_judge": False, "enforcement_mode": "LOCAL_PRECONDITIONS_PLUS_CANONICAL_POLICY_GATEWAY", "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", []), "bundle_sha256": colang.get("bundle_sha256"), }, "codename_gate": { "clean": not codename_hits, "hits": codename_hits, }, "policy_gateway": policy_gateway, } def _atomic_receipt( *, proposal_id: str, request_id: str, request_digest: 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 = { "schema": "szl.gdw.transaction-receipt/v1", "status": "UNSIGNED_ATOMIC", "proposal_id": proposal_id, "request_id": request_id, "request_digest": request_digest, "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, "scheduler_mode": scheduler_mode, "governance_evidence_sha256": sha256_json(governance), "governance": governance, "created_at": timestamp, } receipt["receipt_hash"] = sha256_json(receipt) return receipt def register(app, ns: str = "a11oy"): prefix = f"/api/{ns}/v1/gdw" @app.get(prefix + "/healthz") @app.get("/v1/gdw/healthz") def gdw_healthz(): ( 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"), ): _authorise( authorization, namespace=ns, required_scopes=("bench:read",), ) return { "service": "gdw-frontier", "implementation_status": "REAL", "benchmark_status": "UNMEASURED", "recommended_burst": 10000, "metrics_path": prefix + "/metrics", "notes": [ "A 10k-request run is harness evidence, not a production guarantee.", "Use p95, p99, error rate, receipt integrity, and SQLite integrity.", ], } @app.get(prefix + "/metrics", response_class=PlainTextResponse) @app.get("/v1/gdw/metrics", response_class=PlainTextResponse) def gdw_metrics( authorization: Optional[str] = Header(default=None, alias="Authorization"), ): _authorise( authorization, namespace=ns, required_scopes=("metrics:read",), ) return PlainTextResponse( _TELEMETRY.render(), media_type="text/plain; version=0.0.4; charset=utf-8", ) @app.get(prefix + "/integrity") @app.get("/v1/gdw/integrity") def gdw_integrity( authorization: Optional[str] = Header(default=None, alias="Authorization"), ): 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.post(prefix + "/recovery/transient-effects") @app.post("/v1/gdw/recovery/transient-effects") def gdw_recover_transient_effects( limit: int = Query(default=100, ge=1, le=1_000), authorization: Optional[str] = Header(default=None, alias="Authorization"), expected_source_revision: Optional[str] = Header( default=None, alias="X-Expected-Source-Revision", ), idempotency_key: Optional[str] = Header( default=None, alias="Idempotency-Key", ), ): principal = _authorise( authorization, namespace=ns, required_scopes=("effects:recover", "integrity:global"), ) runtime_generation = _require_transient_recovery_runtime( ns, expected_source_revision, ) if not idempotency_key or not _ID_PATTERN.fullmatch(idempotency_key): raise HTTPException( status_code=422, detail=( "Idempotency-Key must be 1-128 canonical identifier " "characters" ), ) workspace = _workspace(principal) if workspace.database_generation_id != runtime_generation: raise HTTPException( status_code=503, detail="GDW recovery database generation changed", ) try: governance = _transient_recovery_governance( principal=principal, recovery_id=idempotency_key, source_revision=str(expected_source_revision), database_generation_id=runtime_generation, limit=limit, ) except PermissionError as exc: raise HTTPException( status_code=403, detail="GDW recovery denied by canonical policy", ) from exc except Exception as exc: raise HTTPException( status_code=503, detail="GDW recovery canonical policy unavailable", ) from exc try: report = workspace.recover_retry_scheduled_effects( recovery_id=idempotency_key, credential_key_id=principal.key_id, expected_source_revision=str(expected_source_revision), expected_database_generation_id=runtime_generation, governance=governance, limit=limit, ) except GDWConfigurationError as exc: if str(exc) == "recovery_id was already used with different content": raise HTTPException( status_code=409, detail="Idempotency-Key was already used with different content", ) from exc raise HTTPException( status_code=503, detail="GDW recovery refused by integrity gate", ) from exc if report.get("database_generation_id") != runtime_generation: raise HTTPException( status_code=503, detail="GDW recovery database generation changed", ) return report @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}") def gdw_session( session_id: str, authorization: Optional[str] = Header(default=None, alias="Authorization"), ): 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") state = _workspace(principal).read_session(session_id) if state is None: raise HTTPException(status_code=404, detail="session not found") return state 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 = _authorise( authorization, namespace=ns, required_scopes=("step:write",), ) await asyncio.to_thread(_require_write_ready, ns) try: raw_payload = await request.json() 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) selected_mode = "unresolved" decision = "ERROR" receipt_hash = "" workspace = _workspace(principal) step_write_locks = _step_write_locks( principal.namespace, principal.owner_id, payload.session_id, request_id, ) await _acquire_step_write_locks(step_write_locks) try: authorised_generation_id = workspace.database_generation_id with workspace.transaction() as connection: cached = workspace.cached_request(connection, request_id) if cached is not None: cached_digest, cached_response = cached if cached_digest != request_digest: raise HTTPException( status_code=409, detail="X-Request-Id was already used with different content", ) 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"] receipt_hash = cached_response.get("receipt_hash") or "" _TELEMETRY.observe( (time.perf_counter() - started) * 1000.0, decision, selected_mode, False, ) return cached_response authorised_previous = workspace.session_state( connection, payload.session_id, ) if authorised_previous is None: authorised_state_hash = _sha( { "namespace": principal.namespace, "owner_id": principal.owner_id, "session_id": payload.session_id, "step": 0, "state": "GENESIS", } ) else: authorised_state_hash = authorised_previous["state_hash"] precondition_decision = _decision(payload) governance = await _governance_gate( payload_data, request_id, request_digest, principal, authorised_generation_id, authorised_state_hash, ) with workspace.transaction() as connection: cached = workspace.cached_request(connection, request_id) if cached is not None: cached_digest, cached_response = cached if cached_digest != request_digest: raise HTTPException( status_code=409, detail="X-Request-Id was already used with different content", ) 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"] receipt_hash = cached_response.get("receipt_hash") or "" _TELEMETRY.observe( (time.perf_counter() - started) * 1000.0, decision, selected_mode, False, ) return cached_response previous = workspace.session_state(connection, payload.session_id) if previous is None: before_step = 0 before_hash = _sha( { "namespace": principal.namespace, "owner_id": principal.owner_id, "session_id": payload.session_id, "step": 0, "state": "GENESIS", } ) else: before_step = previous["step"] before_hash = previous["state_hash"] if ( workspace.database_generation_id != authorised_generation_id or before_hash != authorised_state_hash ): raise HTTPException( status_code=409, detail="state changed after governance authorization", ) features = AttentionFeatures( novelty=payload.novelty if payload.novelty is not None else min(1.0, len(payload.request) / 1024.0), disagreement=payload.disagreement if payload.disagreement is not None else min(1.0, max(0, len(set(payload.allowed_experts)) - 1) / 5.0), risk=payload.risk_budget, context_tokens=payload.context_tokens or max(1, len(payload.request) // 4), active_tool_count=payload.active_tool_count, memory_pressure=payload.memory_pressure or 0.0, ) routing = choose_attention_mode(features, payload.mode_hint) selected_mode = routing["mode"] decision = 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 database_generation_id = workspace.database_generation_id proposal_id = sha256_json( { "schema": "szl.gdw.proposal-identity/v1", "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, "database_generation_id": database_generation_id, "step": step, "previous_state_hash": before_hash, "request_digest": request_digest, "scheduler_mode": selected_mode, "allowed_experts": sorted(set(payload.allowed_experts)), } after_hash = _sha(state) workspace.save_state( connection, payload.session_id, step, state, after_hash, timestamp, ) receipt = _atomic_receipt( proposal_id=proposal_id, request_id=request_id, request_digest=request_digest, 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"] else: state = previous["state"] if previous else {"state": "GENESIS"} after_hash = before_hash receipt = None proof_payload = build_proof_payload( proposal_id=proposal_id, request_id=request_id, request_digest=request_digest, namespace=principal.namespace, owner_id=principal.owner_id, database_generation_id=database_generation_id, step=step, before_hash=before_hash, after_hash=after_hash, decision=decision, scheduler_mode=selected_mode, receipt_hash=receipt_hash, dry_run=payload.dry_run, governance=governance, ) proof_mode = os.environ.get( "GDW_PROOF_EXPORT_MODE", "outbox" ).strip().lower() if proof_mode != "outbox": raise ValueError( "GDW_PROOF_EXPORT_MODE must be 'outbox'; " "synchronous external effects are not transaction-safe" ) proof_artifact = { "status": "OUTBOX_PENDING", "kind": "proof_export", "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", } plan = build_plan( tasks=("governance", "attention_route", "state_transition", "verify"), meta={"proposal_id": proposal_id}, ) response = { "service": "gdw-frontier", "implementation_status": "REAL", "benchmark_status": "UNMEASURED", "proposal_id": proposal_id, "request_id": request_id, "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, "state_before_hash": before_hash, "receipt_hash": receipt_hash or None, "receipt_status": receipt.get("status") if receipt else None, "scheduler_mode": selected_mode, "routing": routing, "kernel_execution": "NOT_EXECUTED_BY_CONTROL_API", "dry_run": payload.dry_run, "replayed": False, "audit": { "governance": governance, "precondition_decision": precondition_decision, "allowed_experts": sorted(set(payload.allowed_experts)), "plan_version": plan["plan_version"], "receipt_substrate": "gdw.sqlite.atomic/v1" if receipt is not None else None, "proof_export_mode": proof_mode, }, "proof": proof_artifact, } response_hash = _sha(response) workspace.save_request( connection, request_id, request_digest, payload.session_id, response, response_hash, timestamp, ) if receipt is not None: receipt_payload_sha256 = sha256_json(receipt) workspace.save_receipt( connection, receipt_hash, request_id, payload.session_id, step, receipt, timestamp, ) workspace.save_effect_outbox( connection, request_id, "receipt_projection", receipt, 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", proof_payload, proof_payload["payload_sha256"], workspace.scoped_effect_key( principal.namespace, principal.owner_id, request_id, "proof_export", proof_payload["payload_sha256"], ), timestamp, ) _TELEMETRY.observe( (time.perf_counter() - started) * 1000.0, decision, selected_mode, bool(receipt_hash), ) return response except HTTPException: raise 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, "ERROR", selected_mode, False, error=True, ) raise HTTPException( status_code=500, detail=f"GDW transition failed closed: {type(exc).__name__}", ) from exc finally: for lock in reversed(step_write_locks): lock.release() return { "ok": True, "state": "REAL", "routes": [ prefix + "/healthz", prefix + "/bench/meta", prefix + "/metrics", prefix + "/integrity", prefix + "/integrity/global", prefix + "/drain", prefix + "/recovery/transient-effects", prefix + "/sessions/{session_id}", prefix + "/step", ], }