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): routers/__init__.py, routers/frontier_reads.py, routers/series_a_control_plane.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.

routers/__init__.py

@@ -18,9 +18,11 @@ SAME lexical position the routes used to occupy — i.e. BEFORE the SPA
   * a missing/broken group can NEVER take down the SPA (guarded), and
   * the register() functions are IMPORTED + CALLED (register-invocation-guard clean).
 
-This is REFACTOR-ONLY. No endpoint behavior changes. Parity is proven by
-tools_serve_split/capture_routes.py (before/after route-table fingerprint) plus the
-Wave-J frontier contract suite (tests/test_frontier_endpoints.py) and TestClient.
+The additive `series_a_control_plane` module is not a refactor-only route group. It
+is the single governed integration seam for current estate truth, Counterfactual
+Action Passports, signed receipts, and bounded one-attempt effectors. It is exported
+here so invocation and package-integrity checks can prove the production module is
+intentional rather than an orphaned source file.
 
 The package top-level name is `routers` (not szl_*/a11oy_*), so it is intentionally
 OUTSIDE the guarded-import-liveness first-party scan — and the files exist anyway.
@@ -28,4 +30,9 @@ OUTSIDE the guarded-import-liveness first-party scan — and the files exist any
 Signed-off-by: Stephen P. Lutar Jr. <stephenlutar2@gmail.com>
 """
 
-__all__ = ["lambda_bounty", "research_3d", "frontier_reads"]
+__all__ = [
+    "lambda_bounty",
+    "research_3d",
+    "frontier_reads",
+    "series_a_control_plane",
+]

routers/frontier_reads.py

@@ -14,7 +14,9 @@ Shared serve.py module-scope state referenced (unchanged, via `import serve`):
 this group, so it moves here with the routes. Registered BEFORE the /api/a11oy/
 {path:path} Node proxy + SPA catch-all, identical to the pre-refactor inline block.
 
-REFACTOR-ONLY: paths, methods, and payloads are byte-identical to before.
+The additive Series-A controller is registered at this same pre-catch-all seam. It
+keeps GET/HEAD read-only, uses explicit POSTs for refresh/evaluate/execute, and
+fails one surface closed without taking down the existing frontier reads.
 
 Signed-off-by: Stephen P. Lutar Jr. <stephenlutar2@gmail.com>
 """
@@ -42,8 +44,7 @@ _A11OY_VERTICALS = [
 
 
 def register(app) -> dict:
-    """Attach the frontier read route group to `app`, identically to the prior
-    inline serve.py block. Called BEFORE the Node proxy + SPA catch-all."""
+    """Attach frontier reads and the additive Series-A control plane."""
     import serve  # shared module-scope state lives at serve module scope
 
     @app.get("/api/a11oy/v1/forecast-baseline")
@@ -60,7 +61,6 @@ def register(app) -> dict:
                              "verticals": _A11OY_VERTICALS,
                              "honesty": "Live = shipping pack; stub = scaffolded, roadmap."})
 
-    # ---- Business Observability (5 domains) on REAL in-image data (no fabricated KPIs) ----
     @app.get("/api/a11oy/v1/observability/business")
     @app.get("/v1/observability/business")
     async def a11oy_business_observability_v2() -> JSONResponse:
@@ -89,8 +89,27 @@ def register(app) -> dict:
             "lambda_status": "Conjecture 1 (advisory)",
         })
 
-    return {"ok": True, "ns": "a11oy", "group": "frontier-reads", "routes": [
-        "/api/a11oy/v1/forecast-baseline", "/v1/forecast-baseline",
-        "/api/a11oy/v1/vertical-packs", "/v1/vertical-packs",
-        "/api/a11oy/v1/observability/business", "/v1/observability/business",
-    ]}
+    try:
+        from routers import series_a_control_plane as _series_a_control_plane
+
+        series_a = _series_a_control_plane.register(app, ns="a11oy")
+    except Exception as exc:  # one additive surface must never take down A11oy
+        series_a = {
+            "ok": False,
+            "state": "UNAVAILABLE",
+            "reason": type(exc).__name__,
+            "effectors": [],
+        }
+
+    return {
+        "ok": True,
+        "ns": "a11oy",
+        "group": "frontier-reads",
+        "series_a": series_a,
+        "routes": [
+            "/api/a11oy/v1/forecast-baseline", "/v1/forecast-baseline",
+            "/api/a11oy/v1/vertical-packs", "/v1/vertical-packs",
+            "/api/a11oy/v1/observability/business", "/v1/observability/business",
+            "/series-a", "/api/a11oy/v1/series-a/status",
+        ],
+    }

routers/series_a_control_plane.py

@@ -0,0 +1,991 @@
+# SPDX-License-Identifier: Apache-2.0
+# © 2026 Lutar, Stephen P. — SZL Holdings
+"""Series-A Live Control Plane for A11oy.
+
+One additive controller combines three previously separate payload families:
+
+* a signed, current estate truth plane;
+* a bounded Counterfactual Action Passport; and
+* a zero-bandaid, one-attempt local action executor.
+
+It uses real GitHub, Hugging Face, HTTP, SQLite, and ECDSA-P256 boundaries.
+GET/HEAD requests never sign or mutate state. Refresh/evaluate/execute operations
+are explicit POSTs, append hash-linked receipts, and fail closed.
+"""
+from __future__ import annotations
+
+import asyncio
+import base64
+import hashlib
+import json
+import os
+import sqlite3
+import threading
+import time
+import uuid
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from typing import Any, AsyncIterator, Callable, Mapping
+from urllib.parse import urlsplit
+
+import httpx
+from fastapi import FastAPI, HTTPException, Request
+from fastapi.responses import HTMLResponse, JSONResponse, Response, StreamingResponse
+
+SCHEMA_MANIFEST = "szl.estate-manifest/v2"
+SCHEMA_PASSPORT = "szl.counterfactual-action-passport/v3"
+SCHEMA_RECEIPT = "szl.series-a-receipt/v1"
+SCHEMA_STATUS = "szl.series-a-status/v1"
+SCHEMA_TRUST = "szl.agent-trust-factor/v1"
+PAYLOAD_TYPE = "application/vnd.szl.series-a-receipt.v1+json"
+ORG = "szl-holdings"
+HF_ORG = "SZLHOLDINGS"
+CANONICAL_SPACE = f"{HF_ORG}/a11oy"
+FORBIDDEN_CLONES = tuple(f"{HF_ORG}/a11oy-clone-{index}" for index in range(1, 5))
+TTL_SECONDS = 300
+MAX_BODY = 64 * 1024
+MAX_RESPONSE_BYTES = 2 * 1024 * 1024
+MAX_PAGES = 20
+ALLOWED_ACTIONS = {"estate.refresh", "probe.public_surface"}
+ALLOWED_PROBE_HOSTS = {
+    "a-11-oy.com",
+    "a11oy.net",
+    "szlholdings-a11oy.hf.space",
+    "szlholdings-killinchu.hf.space",
+}
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")
+
+
+def _future(seconds: int) -> str:
+    return (datetime.now(timezone.utc) + timedelta(seconds=seconds)).isoformat(
+        timespec="milliseconds"
+    ).replace("+00:00", "Z")
+
+
+def _canonical(value: Any) -> bytes:
+    """Narrow deterministic JSON for signed control-plane records."""
+
+    def walk(item: Any, path: str = "$") -> None:
+        if isinstance(item, float):
+            raise ValueError(f"{path}: floats are forbidden in signed records")
+        if isinstance(item, dict):
+            for key, child in item.items():
+                if not isinstance(key, str):
+                    raise ValueError(f"{path}: keys must be strings")
+                lowered = key.lower()
+                if any(token in lowered for token in ("password", "secret_value", "private_key", "authorization")):
+                    raise ValueError(f"{path}.{key}: secret-shaped field is forbidden")
+                walk(child, f"{path}.{key}")
+            return
+        if isinstance(item, list):
+            for index, child in enumerate(item):
+                walk(child, f"{path}[{index}]")
+            return
+        if item is None or isinstance(item, (str, int, bool)):
+            return
+        raise ValueError(f"{path}: unsupported type {type(item).__name__}")
+
+    walk(value)
+    return json.dumps(
+        value,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+        allow_nan=False,
+    ).encode("utf-8")
+
+
+def _sha(value: Any) -> str:
+    payload = value if isinstance(value, (bytes, bytearray)) else _canonical(value)
+    return hashlib.sha256(payload).hexdigest()
+
+
+def _pae(payload_type: str, payload: bytes) -> bytes:
+    ptype = payload_type.encode("utf-8")
+    return b"DSSEv1 " + str(len(ptype)).encode() + b" " + ptype + b" " + str(len(payload)).encode() + b" " + payload
+
+
+def _safe_error(exc: Exception) -> dict[str, str]:
+    return {"error_class": type(exc).__name__, "error": str(exc)[:240]}
+
+
+def _git_revision() -> str:
+    for key in ("SZL_GIT_SHA", "A11OY_GIT_SHA", "GITHUB_SHA"):
+        value = (os.environ.get(key) or "").strip().lower()
+        if len(value) == 40 and all(ch in "0123456789abcdef" for ch in value):
+            return value
+    return "UNKNOWN"
+
+
+class ReceiptSigner:
+    def __init__(self) -> None:
+        self.private_key = None
+        self.public_pem = ""
+        self.source = "unavailable"
+        self.error = ""
+        try:
+            from a11oy_signing_key import load_signing_key
+
+            private_key, public_pem, source, error = load_signing_key()
+            self.private_key = private_key
+            self.public_pem = public_pem or ""
+            self.source = source or "unavailable"
+            self.error = error or ""
+        except Exception as exc:
+            self.error = f"{type(exc).__name__}: {str(exc)[:180]}"
+
+    @property
+    def keyid(self) -> str | None:
+        return _sha(self.public_pem.encode("utf-8")) if self.public_pem else None
+
+    def sign(self, payload: Mapping[str, Any]) -> dict[str, Any]:
+        body = _canonical(dict(payload))
+        envelope: dict[str, Any] = {
+            "payloadType": PAYLOAD_TYPE,
+            "payload": base64.b64encode(body).decode("ascii"),
+            "signatures": [],
+            "pae_sha256": hashlib.sha256(_pae(PAYLOAD_TYPE, body)).hexdigest(),
+            "key_source": self.source,
+        }
+        if self.private_key is None:
+            envelope["signature_status"] = "UNSIGNED_UNAVAILABLE"
+            envelope["signature_error"] = self.error or "signing key unavailable"
+            return envelope
+        try:
+            from cryptography.hazmat.primitives import hashes
+            from cryptography.hazmat.primitives.asymmetric import ec
+
+            signature = self.private_key.sign(
+                _pae(PAYLOAD_TYPE, body), ec.ECDSA(hashes.SHA256())
+            )
+            envelope["signatures"] = [
+                {
+                    "keyid": self.keyid,
+                    "sig": base64.b64encode(signature).decode("ascii"),
+                }
+            ]
+            envelope["signature_status"] = "SIGNED"
+            return envelope
+        except Exception as exc:
+            envelope["signature_status"] = "UNSIGNED_ERROR"
+            envelope["signature_error"] = f"{type(exc).__name__}: {str(exc)[:180]}"
+            return envelope
+
+
+class Store:
+    def __init__(self, requested_path: str | None = None) -> None:
+        self.path = self._resolve_path(requested_path)
+        self.lock = threading.RLock()
+        self._init()
+
+    @staticmethod
+    def _resolve_path(requested: str | None) -> str:
+        candidates = [
+            requested or os.environ.get("A11OY_SERIES_A_DB") or "/data/series-a/control-plane.sqlite3",
+            "/tmp/a11oy_series_a_control_plane.sqlite3",
+        ]
+        for candidate in candidates:
+            try:
+                path = Path(candidate)
+                path.parent.mkdir(parents=True, exist_ok=True)
+                with path.parent.joinpath(".write-probe").open("w", encoding="utf-8") as probe:
+                    probe.write("ok")
+                path.parent.joinpath(".write-probe").unlink(missing_ok=True)
+                return str(path)
+            except Exception:
+                continue
+        raise RuntimeError("no writable SQLite location")
+
+    def connect(self) -> sqlite3.Connection:
+        connection = sqlite3.connect(self.path, timeout=30)
+        connection.row_factory = sqlite3.Row
+        connection.execute("PRAGMA foreign_keys=ON")
+        connection.execute("PRAGMA journal_mode=WAL")
+        connection.execute("PRAGMA synchronous=FULL")
+        return connection
+
+    def _init(self) -> None:
+        with self.lock, self.connect() as db:
+            db.executescript(
+                """
+                CREATE TABLE IF NOT EXISTS snapshots(
+                  digest TEXT PRIMARY KEY,
+                  payload TEXT NOT NULL,
+                  envelope TEXT NOT NULL,
+                  observed_at TEXT NOT NULL,
+                  valid_until TEXT NOT NULL
+                );
+                CREATE TABLE IF NOT EXISTS passports(
+                  digest TEXT PRIMARY KEY,
+                  payload TEXT NOT NULL,
+                  decision TEXT NOT NULL,
+                  attempts INTEGER NOT NULL DEFAULT 0 CHECK(attempts BETWEEN 0 AND 1),
+                  created_at TEXT NOT NULL
+                );
+                CREATE TABLE IF NOT EXISTS receipts(
+                  sequence INTEGER PRIMARY KEY AUTOINCREMENT,
+                  receipt_id TEXT NOT NULL UNIQUE,
+                  kind TEXT NOT NULL,
+                  payload TEXT NOT NULL,
+                  envelope TEXT NOT NULL,
+                  previous_hash TEXT NOT NULL,
+                  receipt_hash TEXT NOT NULL UNIQUE,
+                  created_at TEXT NOT NULL
+                );
+                CREATE TABLE IF NOT EXISTS events(
+                  sequence INTEGER PRIMARY KEY AUTOINCREMENT,
+                  event_id TEXT NOT NULL UNIQUE,
+                  kind TEXT NOT NULL,
+                  payload TEXT NOT NULL,
+                  created_at TEXT NOT NULL
+                );
+                """
+            )
+
+    def append_event(self, kind: str, payload: Mapping[str, Any]) -> None:
+        with self.lock, self.connect() as db:
+            db.execute(
+                "INSERT INTO events(event_id,kind,payload,created_at) VALUES(?,?,?,?)",
+                (f"evt_{uuid.uuid4().hex}", kind, json.dumps(dict(payload), sort_keys=True), _now()),
+            )
+
+    def events_since(self, sequence: int, limit: int = 100) -> list[dict[str, Any]]:
+        with self.lock, self.connect() as db:
+            rows = db.execute(
+                "SELECT sequence,event_id,kind,payload,created_at FROM events WHERE sequence>? ORDER BY sequence LIMIT ?",
+                (max(0, sequence), max(1, min(limit, 500))),
+            ).fetchall()
+        return [
+            {
+                "sequence": row["sequence"],
+                "event_id": row["event_id"],
+                "kind": row["kind"],
+                "payload": json.loads(row["payload"]),
+                "created_at": row["created_at"],
+            }
+            for row in rows
+        ]
+
+    def append_receipt(
+        self, kind: str, payload: Mapping[str, Any], signer: ReceiptSigner
+    ) -> dict[str, Any]:
+        with self.lock, self.connect() as db:
+            row = db.execute(
+                "SELECT receipt_hash FROM receipts ORDER BY sequence DESC LIMIT 1"
+            ).fetchone()
+            previous = row["receipt_hash"] if row else "0" * 64
+            receipt = {
+                "schema": SCHEMA_RECEIPT,
+                "receipt_id": f"rcpt_{uuid.uuid4().hex}",
+                "kind": kind,
+                "created_at": _now(),
+                "source_revision": _git_revision(),
+                "previous_receipt_hash": previous,
+                "payload": dict(payload),
+            }
+            envelope = signer.sign(receipt)
+            receipt_hash = _sha(envelope)
+            db.execute(
+                """INSERT INTO receipts(receipt_id,kind,payload,envelope,previous_hash,receipt_hash,created_at)
+                   VALUES(?,?,?,?,?,?,?)""",
+                (
+                    receipt["receipt_id"],
+                    kind,
+                    json.dumps(receipt, sort_keys=True),
+                    json.dumps(envelope, sort_keys=True),
+                    previous,
+                    receipt_hash,
+                    receipt["created_at"],
+                ),
+            )
+        self.append_event(kind, {"receipt_hash": receipt_hash, "receipt_id": receipt["receipt_id"]})
+        return {"receipt": receipt, "envelope": envelope, "receipt_hash": receipt_hash}
+
+    def list_receipts(self, limit: int = 50) -> list[dict[str, Any]]:
+        with self.lock, self.connect() as db:
+            rows = db.execute(
+                "SELECT sequence,kind,payload,envelope,receipt_hash,created_at FROM receipts ORDER BY sequence DESC LIMIT ?",
+                (max(1, min(limit, 200)),),
+            ).fetchall()
+        return [
+            {
+                "sequence": row["sequence"],
+                "kind": row["kind"],
+                "receipt": json.loads(row["payload"]),
+                "envelope": json.loads(row["envelope"]),
+                "receipt_hash": row["receipt_hash"],
+                "created_at": row["created_at"],
+            }
+            for row in rows
+        ]
+
+    def save_snapshot(self, manifest: Mapping[str, Any], envelope: Mapping[str, Any]) -> str:
+        digest = _sha(manifest)
+        with self.lock, self.connect() as db:
+            db.execute(
+                "INSERT OR REPLACE INTO snapshots(digest,payload,envelope,observed_at,valid_until) VALUES(?,?,?,?,?)",
+                (
+                    digest,
+                    json.dumps(dict(manifest), sort_keys=True),
+                    json.dumps(dict(envelope), sort_keys=True),
+                    manifest["observed_at"],
+                    manifest["valid_until"],
+                ),
+            )
+        return digest
+
+    def latest_snapshot(self) -> dict[str, Any] | None:
+        with self.lock, self.connect() as db:
+            row = db.execute(
+                "SELECT digest,payload,envelope,observed_at,valid_until FROM snapshots ORDER BY observed_at DESC LIMIT 1"
+            ).fetchone()
+        if row is None:
+            return None
+        return {
+            "digest": row["digest"],
+            "manifest": json.loads(row["payload"]),
+            "envelope": json.loads(row["envelope"]),
+            "observed_at": row["observed_at"],
+            "valid_until": row["valid_until"],
+        }
+
+    def save_passport(self, passport: Mapping[str, Any]) -> str:
+        digest = _sha(passport)
+        with self.lock, self.connect() as db:
+            db.execute(
+                "INSERT INTO passports(digest,payload,decision,attempts,created_at) VALUES(?,?,?,?,?)",
+                (digest, json.dumps(dict(passport), sort_keys=True), passport["decision"], 0, passport["created_at"]),
+            )
+        return digest
+
+    def load_passport(self, digest: str) -> dict[str, Any] | None:
+        with self.lock, self.connect() as db:
+            row = db.execute(
+                "SELECT payload,decision,attempts FROM passports WHERE digest=?", (digest,)
+            ).fetchone()
+        if row is None:
+            return None
+        value = json.loads(row["payload"])
+        value["attempts"] = row["attempts"]
+        return value
+
+    def consume_attempt(self, digest: str) -> None:
+        with self.lock, self.connect() as db:
+            result = db.execute(
+                "UPDATE passports SET attempts=1 WHERE digest=? AND attempts=0", (digest,)
+            )
+            if result.rowcount != 1:
+                raise RuntimeError("passport attempt is absent or already consumed")
+
+
+@dataclass
+class Observation:
+    state: str
+    value: Any = None
+    detail: Mapping[str, Any] | None = None
+
+    def as_dict(self) -> dict[str, Any]:
+        value = {"state": self.state}
+        if self.value is not None:
+            value["value"] = self.value
+        if self.detail:
+            value["detail"] = dict(self.detail)
+        return value
+
+
+class Collector:
+    def __init__(self) -> None:
+        self.github_token = (os.environ.get("GITHUB_TOKEN") or "").strip()
+        self.hf_token = (os.environ.get("HF_TOKEN") or "").strip()
+
+    async def _json(
+        self,
+        client: httpx.AsyncClient,
+        url: str,
+        *,
+        params: Mapping[str, Any] | None = None,
+        allowed_host: str,
+    ) -> tuple[Any, httpx.Response]:
+        parsed = urlsplit(url)
+        if parsed.scheme != "https" or parsed.hostname != allowed_host or parsed.username or parsed.password:
+            raise RuntimeError("outbound URL left the fixed HTTPS origin")
+        response = await client.get(url, params=params)
+        if response.status_code != 200:
+            raise RuntimeError(f"HTTP {response.status_code}")
+        if len(response.content) > MAX_RESPONSE_BYTES:
+            raise RuntimeError("response exceeded byte limit")
+        final = urlsplit(str(response.url))
+        if final.scheme != "https" or final.hostname != allowed_host:
+            raise RuntimeError("redirect left the fixed HTTPS origin")
+        return response.json(), response
+
+    async def github(self) -> Observation:
+        headers = {"accept": "application/vnd.github+json", "user-agent": "szl-series-a/1"}
+        if self.github_token:
+            headers["authorization"] = f"Bearer {self.github_token}"
+        try:
+            repos: list[dict[str, Any]] = []
+            async with httpx.AsyncClient(headers=headers, timeout=30, follow_redirects=False) as client:
+                complete = False
+                for page in range(1, MAX_PAGES + 1):
+                    values, _ = await self._json(
+                        client,
+                        f"https://api.github.com/orgs/{ORG}/repos",
+                        params={"type": "all", "per_page": 100, "page": page},
+                        allowed_host="api.github.com",
+                    )
+                    if not isinstance(values, list):
+                        raise RuntimeError("repository listing was not an array")
+                    repos.extend(item for item in values if isinstance(item, dict))
+                    if len(values) < 100:
+                        complete = True
+                        break
+                if not complete:
+                    raise RuntimeError("repository pagination exceeded bounded window")
+                pr_data, _ = await self._json(
+                    client,
+                    "https://api.github.com/search/issues",
+                    params={"q": f"org:{ORG} is:pr is:open", "per_page": 1},
+                    allowed_host="api.github.com",
+                )
+            rows = [
+                {
+                    "name": str(item.get("name") or ""),
+                    "archived": bool(item.get("archived")),
+                    "visibility": str(item.get("visibility") or "unknown"),
+                    "default_branch": str(item.get("default_branch") or ""),
+                    "updated_at": str(item.get("updated_at") or ""),
+                }
+                for item in repos
+            ]
+            return Observation(
+                "OBSERVED",
+                {
+                    "repository_count": len(rows),
+                    "open_pull_request_count": int((pr_data or {}).get("total_count", 0)),
+                    "pagination_complete": True,
+                    "repositories": rows,
+                },
+                {"authenticated": bool(self.github_token)},
+            )
+        except Exception as exc:
+            return Observation("UNAVAILABLE", detail=_safe_error(exc))
+
+    def _hf_list(self, method_name: str, kwargs: Mapping[str, Any]) -> list[Any]:
+        from huggingface_hub import HfApi
+
+        api = HfApi(token=self.hf_token or None)
+        method = getattr(api, method_name, None)
+        if method is None:
+            raise AttributeError(f"HfApi.{method_name} unavailable")
+        return list(method(**dict(kwargs)))
+
+    async def _hf_kernels(self) -> list[dict[str, Any]]:
+        headers = {"accept": "application/json", "user-agent": "szl-series-a/1"}
+        if self.hf_token:
+            headers["authorization"] = f"Bearer {self.hf_token}"
+        output: list[dict[str, Any]] = []
+        url: str | None = "https://huggingface.co/api/kernels"
+        params: Mapping[str, Any] | None = {"author": HF_ORG, "limit": 1000, "full": "true"}
+        async with httpx.AsyncClient(headers=headers, timeout=30, follow_redirects=False) as client:
+            for _ in range(MAX_PAGES):
+                if not url:
+                    return output
+                values, response = await self._json(
+                    client, url, params=params, allowed_host="huggingface.co"
+                )
+                if not isinstance(values, list):
+                    raise RuntimeError("kernel listing was not an array")
+                output.extend(item for item in values if isinstance(item, dict))
+                link = response.links.get("next") or {}
+                url = link.get("url") if isinstance(link, dict) else None
+                params = None
+                if not url:
+                    return output
+        raise RuntimeError("kernel pagination exceeded bounded window")
+
+    async def huggingface(self) -> Observation:
+        categories: dict[str, Any] = {}
+        errors: dict[str, Any] = {}
+        methods = {
+            "models": ("list_models", {"author": HF_ORG}),
+            "datasets": ("list_datasets", {"author": HF_ORG}),
+            "spaces": ("list_spaces", {"author": HF_ORG}),
+            "collections": ("list_collections", {"owner": HF_ORG}),
+            "buckets": ("list_buckets", {"namespace": HF_ORG}),
+        }
+        for name, (method, kwargs) in methods.items():
+            try:
+                items = await asyncio.to_thread(self._hf_list, method, kwargs)
+                rows = []
+                for item in items:
+                    item_id = None
+                    for field in ("id", "repo_id", "name", "slug"):
+                        candidate = item.get(field) if isinstance(item, dict) else getattr(item, field, None)
+                        if isinstance(candidate, str) and candidate:
+                            item_id = candidate
+                            break
+                    rows.append({"id": item_id})
+                categories[name] = {"state": "OBSERVED", "count": len(rows), "items": rows}
+            except Exception as exc:
+                categories[name] = {"state": "UNAVAILABLE"}
+                errors[name] = _safe_error(exc)
+        try:
+            kernels = await self._hf_kernels()
+            categories["kernels"] = {
+                "state": "OBSERVED",
+                "count": len(kernels),
+                "items": [{"id": str(item.get("id") or item.get("repo_id") or "")} for item in kernels],
+            }
+        except Exception as exc:
+            categories["kernels"] = {"state": "UNAVAILABLE"}
+            errors["kernels"] = _safe_error(exc)
+
+        space_ids = {
+            row.get("id")
+            for row in categories.get("spaces", {}).get("items", [])
+            if isinstance(row, dict)
+        }
+        clones_present = sorted(value for value in FORBIDDEN_CLONES if value in space_ids)
+        canonical_present = CANONICAL_SPACE in space_ids
+        state = "OBSERVED" if categories.get("spaces", {}).get("state") == "OBSERVED" else "PARTIAL"
+        return Observation(
+            state,
+            {
+                "categories": categories,
+                "canonical_space": CANONICAL_SPACE,
+                "canonical_present": canonical_present,
+                "forbidden_clones_present": clones_present,
+                "singleton_ok": canonical_present and not clones_present,
+            },
+            {"authenticated": bool(self.hf_token), "errors": errors},
+        )
+
+    async def collect(self) -> dict[str, Any]:
+        github, hf = await asyncio.gather(self.github(), self.huggingface())
+        critical_failures: list[str] = []
+        if github.state != "OBSERVED":
+            critical_failures.append("github_inventory_unavailable")
+        if hf.state not in {"OBSERVED", "PARTIAL"}:
+            critical_failures.append("huggingface_inventory_unavailable")
+        hf_value = hf.value if isinstance(hf.value, dict) else {}
+        if hf_value and not hf_value.get("singleton_ok"):
+            critical_failures.append("canonical_a11oy_singleton_failed")
+        categories = hf_value.get("categories", {}) if isinstance(hf_value, dict) else {}
+        counts = {
+            name: value.get("count") if isinstance(value, dict) and value.get("state") == "OBSERVED" else None
+            for name, value in categories.items()
+        }
+        manifest = {
+            "schema": SCHEMA_MANIFEST,
+            "observed_at": _now(),
+            "valid_until": _future(TTL_SECONDS),
+            "source_revision": _git_revision(),
+            "organization": ORG,
+            "huggingface_organization": HF_ORG,
+            "status": "BLOCKED" if critical_failures else "OBSERVED",
+            "critical_failures": critical_failures,
+            "github": github.as_dict(),
+            "huggingface": hf.as_dict(),
+            "counts": {
+                "github_repositories": (
+                    github.value.get("repository_count")
+                    if isinstance(github.value, dict) and github.state == "OBSERVED"
+                    else None
+                ),
+                "github_open_pull_requests": (
+                    github.value.get("open_pull_request_count")
+                    if isinstance(github.value, dict) and github.state == "OBSERVED"
+                    else None
+                ),
+                **counts,
+            },
+            "claim": "CURRENT_OBSERVATION_NOT_ETERNAL_TRUTH",
+            "counterfactual_label": "MODELED",
+            "private_reasoning_collected": False,
+        }
+        manifest["manifest_digest"] = _sha(manifest)
+        return manifest
+
+
+class Service:
+    def __init__(self, db_path: str | None = None) -> None:
+        self.store = Store(db_path)
+        self.signer = ReceiptSigner()
+        self.collector = Collector()
+        self.refresh_lock = asyncio.Lock()
+        self.started = False
+        self.background_task: asyncio.Task[Any] | None = None
+
+    async def start(self) -> None:
+        if self.started:
+            return
+        self.started = True
+        if (os.environ.get("A11OY_SERIES_A_STARTUP_REFRESH") or "1").strip() == "0":
+            self.store.append_event("estate.refresh.skipped", {"reason": "explicit test/runtime configuration"})
+            return
+
+        async def run() -> None:
+            try:
+                await self.refresh("startup")
+            except Exception as exc:
+                self.store.append_event("estate.refresh.failed", _safe_error(exc))
+
+        self.background_task = asyncio.create_task(run(), name="a11oy-series-a-startup-refresh")
+
+    async def refresh(self, actor: str) -> dict[str, Any]:
+        if self.refresh_lock.locked():
+            raise HTTPException(status_code=409, detail="estate refresh already running")
+        async with self.refresh_lock:
+            manifest = await self.collector.collect()
+            envelope = self.signer.sign(manifest)
+            digest = self.store.save_snapshot(manifest, envelope)
+            receipt = self.store.append_receipt(
+                "estate.refresh",
+                {
+                    "actor": actor,
+                    "manifest_digest": digest,
+                    "status": manifest["status"],
+                    "counts": manifest["counts"],
+                },
+                self.signer,
+            )
+            return {"manifest": manifest, "envelope": envelope, "refresh_receipt": receipt}
+
+    def latest_status(self) -> dict[str, Any]:
+        latest = self.store.latest_snapshot()
+        if latest is None:
+            return {
+                "schema": SCHEMA_STATUS,
+                "state": "PENDING",
+                "terminal": True,
+                "source_revision": _git_revision(),
+                "signing_key_source": self.signer.source,
+                "database": self.store.path,
+                "detail": "no completed refresh is persisted yet",
+            }
+        valid_until = datetime.fromisoformat(latest["valid_until"].replace("Z", "+00:00"))
+        stale = datetime.now(timezone.utc) >= valid_until
+        manifest = latest["manifest"]
+        return {
+            "schema": SCHEMA_STATUS,
+            "state": "STALE" if stale else manifest["status"],
+            "terminal": True,
+            "source_revision": _git_revision(),
+            "manifest_digest": latest["digest"],
+            "observed_at": latest["observed_at"],
+            "valid_until": latest["valid_until"],
+            "counts": manifest.get("counts", {}),
+            "critical_failures": manifest.get("critical_failures", []),
+            "signature_status": latest["envelope"].get("signature_status"),
+            "signing_key_source": self.signer.source,
+            "database": self.store.path,
+        }
+
+    def evaluate_passport(self, body: Mapping[str, Any]) -> dict[str, Any]:
+        action = body.get("action")
+        if not isinstance(action, dict):
+            raise HTTPException(status_code=422, detail="action must be an object")
+        action_type = str(action.get("type") or "")
+        target = str(action.get("target") or "")
+        impact = str(action.get("impact") or "MODERATE").upper()
+        irreversible = bool(action.get("irreversible", False))
+        if action_type not in ALLOWED_ACTIONS:
+            decision = "BLOCK"
+            reasons = ["ACTION_TYPE_NOT_ALLOWLISTED"]
+        elif not target:
+            decision = "BLOCK"
+            reasons = ["TARGET_REQUIRED"]
+        elif impact in {"HIGH", "CRITICAL"} or irreversible:
+            decision = "REQUIRE_APPROVAL"
+            reasons = ["INDEPENDENT_APPROVAL_REQUIRED"]
+        else:
+            decision = "ALLOW"
+            reasons = ["BOUNDED_REVERSIBLE_ACTION"]
+
+        evidence = body.get("evidence")
+        if not isinstance(evidence, list) or not evidence:
+            decision = "BLOCK"
+            reasons = sorted(set(reasons + ["EVIDENCE_REQUIRED"]))
+        else:
+            for item in evidence:
+                if not isinstance(item, dict) or item.get("label") in {"UNKNOWN", "UNAVAILABLE"}:
+                    decision = "BLOCK"
+                    reasons = sorted(set(reasons + ["NON_ACTIONABLE_EVIDENCE"]))
+                    break
+
+        no_action = {
+            "scenario_id": "no-action",
+            "kind": "NO_ACTION",
+            "label": "MODELED",
+            "outcome": str(body.get("expected_if_withheld") or "current state persists"),
+        }
+        proposed = {
+            "scenario_id": "proposed-action",
+            "kind": "PROPOSED_ACTION",
+            "label": "MODELED",
+            "outcome": str(body.get("expected_if_acted") or "bounded action completes or fails closed"),
+        }
+        passport = {
+            "schema": SCHEMA_PASSPORT,
+            "passport_id": f"cap_{uuid.uuid4().hex}",
+            "created_at": _now(),
+            "source_revision": _git_revision(),
+            "subject": {
+                "principal_id": str(body.get("principal_id") or "anonymous-proposer"),
+                "workload_id": str(body.get("workload_id") or "a11oy-series-a"),
+            },
+            "action": action,
+            "action_digest": _sha(action),
+            "evidence": evidence if isinstance(evidence, list) else [],
+            "counterfactuals": [no_action, proposed],
+            "decision": decision,
+            "reason_codes": reasons,
+            "max_attempts": 1,
+            "private_reasoning_collected": False,
+        }
+        digest = self.store.save_passport(passport)
+        receipt = self.store.append_receipt(
+            "passport.evaluate",
+            {"passport_digest": digest, "decision": decision, "reason_codes": reasons},
+            self.signer,
+        )
+        return {"passport": passport, "passport_digest": digest, "decision_receipt": receipt}
+
+    async def execute(self, body: Mapping[str, Any]) -> dict[str, Any]:
+        digest = str(body.get("passport_digest") or "")
+        passport = self.store.load_passport(digest)
+        if passport is None:
+            raise HTTPException(status_code=404, detail="passport not found")
+        if passport["attempts"] != 0:
+            raise HTTPException(status_code=409, detail="passport attempt already consumed")
+        if passport["decision"] != "ALLOW":
+            raise HTTPException(status_code=403, detail=f"passport decision is {passport['decision']}")
+        self.store.consume_attempt(digest)
+        action = passport["action"]
+        started = _now()
+        try:
+            if action["type"] == "estate.refresh":
+                result = await self.refresh(passport["passport_id"])
+                outcome = {
+                    "status": "SUCCEEDED",
+                    "manifest_digest": result["manifest"]["manifest_digest"],
+                    "estate_status": result["manifest"]["status"],
+                }
+            elif action["type"] == "probe.public_surface":
+                outcome = await self._probe(str(action["target"]))
+            else:
+                raise RuntimeError("action left allowlist after authorization")
+        except Exception as exc:
+            outcome = {"status": "FAILED", **_safe_error(exc)}
+        outcome.update(
+            {
+                "started_at": started,
+                "completed_at": _now(),
+                "attempt": 1,
+                "max_attempts": 1,
+                "passport_digest": digest,
+            }
+        )
+        receipt = self.store.append_receipt("passport.outcome", outcome, self.signer)
+        return {"outcome": outcome, "outcome_receipt": receipt}
+
+    async def _probe(self, target: str) -> dict[str, Any]:
+        parsed = urlsplit(target)
+        if parsed.scheme != "https" or parsed.hostname not in ALLOWED_PROBE_HOSTS or parsed.username or parsed.password:
+            raise RuntimeError("probe target is not in the fixed HTTPS allowlist")
+        start = time.monotonic()
+        async with httpx.AsyncClient(timeout=15, follow_redirects=False) as client:
+            response = await client.get(target, headers={"accept": "application/json,text/html;q=0.9"})
+        final = urlsplit(str(response.url))
+        if final.hostname not in ALLOWED_PROBE_HOSTS:
+            raise RuntimeError("probe redirect left the allowlist")
+        return {
+            "status": "SUCCEEDED" if 200 <= response.status_code < 400 else "FAILED",
+            "target": target,
+            "http_status": response.status_code,
+            "latency_ms": int((time.monotonic() - start) * 1000),
+            "bytes": len(response.content),
+            "content_type": response.headers.get("content-type", ""),
+        }
+
+    def trust_factor(self) -> dict[str, Any]:
+        receipts = self.store.list_receipts(200)
+        decisions = [
+            item["receipt"]["payload"].get("decision")
+            for item in receipts
+            if item["kind"] == "passport.evaluate"
+        ]
+        counts = {name: decisions.count(name) for name in ("ALLOW", "BLOCK", "REQUIRE_APPROVAL")}
+        total = sum(counts.values())
+        penalty = counts["BLOCK"] * 10 + counts["REQUIRE_APPROVAL"] * 3
+        score = 100 if total == 0 else max(0, 100 - (penalty * 100 // max(1, total * 10)))
+        return {
+            "schema": SCHEMA_TRUST,
+            "state": "OBSERVED",
+            "total_evaluations": total,
+            "counts": counts,
+            "score_0_to_100": score,
+            "basis": "local signed passport decision receipts",
+            "not_a_security_certification": True,
+        }
+
+
+async def _bounded_json(request: Request) -> dict[str, Any]:
+    content_type = request.headers.get("content-type", "").split(";", 1)[0].lower()
+    if content_type != "application/json":
+        raise HTTPException(status_code=415, detail="content-type must be application/json")
+    declared = request.headers.get("content-length")
+    if declared:
+        try:
+            if int(declared) > MAX_BODY:
+                raise HTTPException(status_code=413, detail="request exceeds 64 KiB")
+        except ValueError as exc:
+            raise HTTPException(status_code=400, detail="invalid content-length") from exc
+    body = await request.body()
+    if len(body) > MAX_BODY:
+        raise HTTPException(status_code=413, detail="request exceeds 64 KiB")
+    try:
+        value = json.loads(body.decode("utf-8"))
+    except Exception as exc:
+        raise HTTPException(status_code=400, detail="request must be UTF-8 JSON") from exc
+    if not isinstance(value, dict):
+        raise HTTPException(status_code=422, detail="request must be one JSON object")
+    return value
+
+
+def _asset(name: str) -> str:
+    path = Path(__file__).resolve().parent / "series_a_web" / name
+    if not path.is_file():
+        raise HTTPException(status_code=404, detail=f"asset missing: {name}")
+    return path.read_text(encoding="utf-8")
+
+
+def register(app: FastAPI, ns: str = "a11oy", *, db_path: str | None = None) -> dict[str, Any]:
+    if any(getattr(route, "path", None) == f"/api/{ns}/v1/series-a/status" for route in app.router.routes):
+        return {"ok": True, "state": "ALREADY_REGISTERED", "routes": []}
+
+    service = Service(db_path)
+    prefix = f"/api/{ns}/v1/series-a"
+
+    async def page(request: Request) -> Response:
+        if request.method == "HEAD":
+            return Response(status_code=200, media_type="text/html")
+        return HTMLResponse(_asset("index.html"), headers={"cache-control": "no-store"})
+
+    async def js(request: Request) -> Response:
+        if request.method == "HEAD":
+            return Response(status_code=200, media_type="application/javascript")
+        return Response(_asset("app.js"), media_type="application/javascript", headers={"cache-control": "public,max-age=300"})
+
+    async def css(request: Request) -> Response:
+        if request.method == "HEAD":
+            return Response(status_code=200, media_type="text/css")
+        return Response(_asset("styles.css"), media_type="text/css", headers={"cache-control": "public,max-age=300"})
+
+    async def status(request: Request) -> Response:
+        payload = service.latest_status()
+        if request.method == "HEAD":
+            return Response(status_code=200, media_type="application/json")
+        return JSONResponse(payload, headers={"cache-control": "no-store"})
+
+    async def manifest(request: Request) -> Response:
+        latest = service.store.latest_snapshot()
+        if latest is None:
+            payload = {"schema": SCHEMA_MANIFEST, "status": "PENDING", "terminal": True}
+        else:
+            payload = latest
+        if request.method == "HEAD":
+            return Response(status_code=200, media_type="application/json")
+        return JSONResponse(payload, headers={"cache-control": "no-store"})
+
+    async def refresh(request: Request) -> Response:
+        body = await _bounded_json(request)
+        actor = str(body.get("actor") or "operator")[:120]
+        return JSONResponse(await service.refresh(actor))
+
+    async def evaluate(request: Request) -> Response:
+        return JSONResponse(service.evaluate_passport(await _bounded_json(request)))
+
+    async def execute(request: Request) -> Response:
+        return JSONResponse(await service.execute(await _bounded_json(request)))
+
+    async def receipts(request: Request) -> Response:
+        if request.method == "HEAD":
+            return Response(status_code=200, media_type="application/json")
+        return JSONResponse({"schema": "szl.series-a-receipts/v1", "items": service.store.list_receipts(50)})
+
+    async def trust(request: Request) -> Response:
+        if request.method == "HEAD":
+            return Response(status_code=200, media_type="application/json")
+        return JSONResponse(service.trust_factor())
+
+    async def public_key(request: Request) -> Response:
+        if request.method == "HEAD":
+            return Response(status_code=200, media_type="text/plain")
+        if not service.signer.public_pem:
+            return JSONResponse({"state": "UNAVAILABLE", "reason": service.signer.error}, status_code=503)
+        return Response(service.signer.public_pem, media_type="text/plain", headers={"cache-control": "public,max-age=300"})
+
+    async def events(request: Request) -> StreamingResponse:
+        last = int(request.query_params.get("after", "0") or 0)
+
+        async def generate() -> AsyncIterator[bytes]:
+            cursor = max(0, last)
+            for _ in range(120):
+                values = service.store.events_since(cursor)
+                for event in values:
+                    cursor = event["sequence"]
+                    yield f"id: {cursor}\nevent: {event['kind']}\ndata: {json.dumps(event, separators=(',', ':'))}\n\n".encode()
+                if await request.is_disconnected():
+                    break
+                yield b": heartbeat\n\n"
+                await asyncio.sleep(1)
+
+        return StreamingResponse(generate(), media_type="text/event-stream", headers={"cache-control": "no-store"})
+
+    routes: list[tuple[str, Callable[..., Any], list[str]]] = [
+        ("/series-a", page, ["GET", "HEAD"]),
+        ("/series-a/app.js", js, ["GET", "HEAD"]),
+        ("/series-a/styles.css", css, ["GET", "HEAD"]),
+        (f"{prefix}/status", status, ["GET", "HEAD"]),
+        (f"{prefix}/manifest", manifest, ["GET", "HEAD"]),
+        (f"{prefix}/refresh", refresh, ["POST"]),
+        (f"{prefix}/passports/evaluate", evaluate, ["POST"]),
+        (f"{prefix}/passports/execute", execute, ["POST"]),
+        (f"{prefix}/receipts", receipts, ["GET", "HEAD"]),
+        (f"{prefix}/trust", trust, ["GET", "HEAD"]),
+        (f"{prefix}/public-key", public_key, ["GET", "HEAD"]),
+        (f"{prefix}/events", events, ["GET"]),
+    ]
+    added: list[str] = []
+    for path, endpoint, methods in routes:
+        app.add_api_route(path, endpoint, methods=methods, include_in_schema=False)
+        added.append(path)
+
+    route_set = set(added)
+    selected = [route for route in app.router.routes if getattr(route, "path", None) in route_set]
+    selected_ids = {id(route) for route in selected}
+    app.router.routes[:] = selected + [route for route in app.router.routes if id(route) not in selected_ids]
+
+    app.state.szl_series_a_service = service
+    add_handler = getattr(app, "add_event_handler", None)
+    if callable(add_handler):
+        add_handler("startup", service.start)
+
+    return {
+        "ok": True,
+        "state": "REGISTERED",
+        "namespace": ns,
+        "routes": sorted(added),
+        "database": service.store.path,
+        "signing_key_source": service.signer.source,
+        "sign_on_read": False,
+        "effectors": sorted(ALLOWED_ACTIONS),
+        "max_attempts": 1,
+        "private_reasoning_collected": False,
+    }