feat(task_plane): Runtime/Authority Catalog (P0-1) read-only resolver skeleton

- authority_catalog.py: AuthorityCatalog + CatalogResource (schema aios.catalog.v1). resolve_worker/task/repository/generic; no write path (red line); UNKNOWN fields null, never fabricate live evidence; summary() feeds runtime_source_unknown metric. - Tests: test_authority.py (9) green incl. read-only invariant. Base: agent/annex-2.3-platform-r2-20260914.
2026-09-14 18:07:11 +08:00
parent 000d535019
commit 48eb5d2036
2 changed files with 479 additions and 0 deletions
@@ -0,0 +1,286 @@
+"""Read-only Runtime / Authority Catalog (Catalog V1).
+
+Pure evidence-producer / resolve layer. It deliberately does NOT create an
+authority store, does NOT write to any database, and does NOT mutate
+worker / task / business state (red line: no parallel authority store
+without an ADR).
+
+Result shape follows the Catalog V1 contract (V4 section 4.1)::
+
+    {
+      "schema_version": "aios.catalog.v1",
+      "resource_type": ...,
+      "resource_id": ...,
+      "authority": {"kind", "endpoint_or_path", "revision"},
+      "runtime":   {"service", "pid", "loaded_source_path", "source_commit",
+                    "artifact_sha256", "state_root"},
+      "contracts": [...],
+      "conflicts": [],
+      "evidence":  [],
+    }
+
+Read-only principle: parse as much as is observable from the supplied
+``TaskPlaneStore`` and the resolve arguments. Anything not observable is
+left ``None`` / unknown -- we never fabricate live evidence. If no store
+is supplied, every resolve returns a structurally complete record with all
+observation fields set to ``None`` and an empty ``evidence`` list.
+"""
+
+from __future__ import annotations
+
+import dataclasses
+from dataclasses import dataclass, field
+from typing import Any
+
+SCHEMA_VERSION = "aios.catalog.v1"
+
+# Runtime fields that describe the loaded source of a runtime instance.
+# Their absence is what the ``runtime_source_unknown`` metric counts.
+_RUNTIME_SOURCE_FIELDS = ("loaded_source_path", "source_commit", "artifact_sha256")
+
+
+def _null_runtime() -> dict[str, Any]:
+    return {
+        "service": None,
+        "pid": None,
+        "loaded_source_path": None,
+        "source_commit": None,
+        "artifact_sha256": None,
+        "state_root": None,
+    }
+
+
+def _null_authority(kind: str | None) -> dict[str, Any]:
+    return {
+        "kind": kind,
+        "endpoint_or_path": None,
+        "revision": None,
+    }
+
+
+@dataclass(frozen=True)
+class CatalogResource:
+    """A single resolved catalog entry (immutable value object).
+
+    Every field is JSON-serialisable. Fields that could not be observed are
+    intentionally ``None`` rather than fabricated.
+    """
+
+    schema_version: str = SCHEMA_VERSION
+    resource_type: str | None = None
+    resource_id: str | None = None
+    authority: dict = field(default_factory=lambda: _null_authority(None))
+    runtime: dict = field(default_factory=_null_runtime)
+    contracts: list = field(default_factory=list)
+    conflicts: list = field(default_factory=list)
+    evidence: list = field(default_factory=list)
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "schema_version": self.schema_version,
+            "resource_type": self.resource_type,
+            "resource_id": self.resource_id,
+            "authority": dict(self.authority),
+            "runtime": dict(self.runtime),
+            "contracts": list(self.contracts),
+            "conflicts": list(self.conflicts),
+            "evidence": list(self.evidence),
+        }
+
+
+class AuthorityCatalog:
+    """Read-only authority / runtime directory.
+
+    ``store`` is optional. When ``None`` every resolve returns a complete
+    but field-empty record (no store -> nothing observable -> no fabrication).
+    Any read through the store is wrapped so a missing record degrades to the
+    same empty-but-complete shape instead of raising.
+    """
+
+    def __init__(self, store: Any = None) -> None:
+        self.store = store
+        self._attempts = 0
+        self._resolved: dict[str, int] = {}
+        self._unknown: dict[str, int] = {}
+        self._missing = {
+            "authority_endpoint_or_path": 0,
+            "authority_revision": 0,
+            "loaded_source_path": 0,
+            "source_commit": 0,
+            "artifact_sha256": 0,
+        }
+        self._runtime_source_unknown = 0
+
+    # -- public resolve API ----------------------------------------------
+
+    def resolve_worker(self, worker_id: str) -> CatalogResource:
+        """service/runtime/authority for a registered worker (or UNKNOWN)."""
+        resource = self._worker_or_unknown(str(worker_id))
+        self._record(resource, resolved=resource.authority.get("revision") is not None)
+        return resource
+
+    def resolve_task(self, task_id: str) -> CatalogResource:
+        """task plan / revision authority (or UNKNOWN)."""
+        resource = self._task_or_unknown(str(task_id))
+        self._record(resource, resolved=resource.authority.get("revision") is not None)
+        return resource
+
+    def resolve_repository(
+        self, repo_id: str, authority_ref: str | None = None
+    ) -> CatalogResource:
+        """repo -> authority ref -> remote OID.
+
+        Remote OID is only reflected from the supplied ``authority_ref``; with
+        no live repository probe it reads as null/unknown rather than guessed.
+        """
+        authority = {
+            "kind": "repository",
+            "endpoint_or_path": str(repo_id),
+            "revision": authority_ref,
+        }
+        resource = dataclasses.replace(
+            CatalogResource(resource_type="repository", resource_id=str(repo_id)),
+            authority=authority,
+        )
+        self._record(resource, resolved=authority.get("revision") is not None)
+        return resource
+
+    def resolve_generic(self, resource_type: str, resource_id: str) -> CatalogResource:
+        """Dispatch known types; anything else -> complete-but-empty UNKNOWN."""
+        rtype = str(resource_type)
+        rid = str(resource_id)
+        if rtype == "worker":
+            return self.resolve_worker(rid)
+        if rtype == "task":
+            return self.resolve_task(rid)
+        if rtype == "repository":
+            return self.resolve_repository(rid)
+        # job / auth / release / schema / state_root / overlay / unknown:
+        # no live source -> never fabricated live state.
+        resource = CatalogResource(
+            resource_type=rtype,
+            resource_id=rid,
+            authority=_null_authority(rtype),
+        )
+        self._record(resource, resolved=False)
+        return resource
+
+    def summary(self) -> dict[str, Any]:
+        """Resolve statistics, including ``runtime_source_unknown`` counters.
+
+        Consumed by the runtime-source-unknown metric: counts how many
+        resolved entries had no observable loaded source path / commit / SHA.
+        """
+        return {
+            "schema_version": SCHEMA_VERSION,
+            "resolve_attempts": self._attempts,
+            "resolved": dict(self._resolved),
+            "unknown": dict(self._unknown),
+            "runtime_source_unknown": self._runtime_source_unknown,
+            "missing_fields": dict(self._missing),
+        }
+
+    # -- internal helpers ------------------------------------------------
+
+    def _worker_or_unknown(self, worker_id: str) -> CatalogResource:
+        base = CatalogResource(
+            resource_type="worker",
+            resource_id=worker_id,
+            authority=_null_authority("worker.registration"),
+        )
+        if self.store is None:
+            return base
+        try:
+            worker = self.store.get_worker(worker_id)
+        except Exception:
+            return base
+        return CatalogResource(
+            resource_type="worker",
+            resource_id=worker.worker_id,
+            authority={
+                "kind": "worker.registration",
+                "endpoint_or_path": worker.identity,
+                "revision": worker.registration_revision,
+            },
+            runtime={
+                "service": worker.worker_type,
+                "pid": None,
+                "loaded_source_path": None,
+                "source_commit": None,
+                "artifact_sha256": None,
+                "state_root": None,
+            },
+            contracts=[
+                {"field": "capabilities", "value": list(worker.capabilities)},
+                {"field": "supported_protocols", "value": list(worker.supported_protocols)},
+                {"field": "trust_zone", "value": worker.trust_zone},
+                {"field": "status", "value": worker.status},
+            ],
+            conflicts=[],
+            evidence=[],
+        )
+
+    def _task_or_unknown(self, task_id: str) -> CatalogResource:
+        base = CatalogResource(
+            resource_type="task",
+            resource_id=task_id,
+            authority=_null_authority("task.plan"),
+        )
+        if self.store is None:
+            return base
+        try:
+            task = self.store.get_task(task_id)
+        except Exception:
+            return base
+        issuer = task.task_plan.issuer or {}
+        issuer_id = issuer.get("id") if isinstance(issuer, dict) else None
+        authorities = task.task_plan.authorities if hasattr(task.task_plan, "authorities") else None
+        return CatalogResource(
+            resource_type="task",
+            resource_id=task.task_id,
+            authority={
+                "kind": "task.plan",
+                "endpoint_or_path": issuer_id or None,
+                "revision": task.revision,
+            },
+            runtime={
+                "service": issuer.get("type") if isinstance(issuer, dict) else None,
+                "pid": None,
+                "loaded_source_path": authorities.get("loaded_source_path")
+                if isinstance(authorities, dict)
+                else None,
+                "source_commit": authorities.get("source_commit")
+                if isinstance(authorities, dict)
+                else None,
+                "artifact_sha256": authorities.get("artifact_sha256")
+                if isinstance(authorities, dict)
+                else None,
+                "state_root": task.task_plan.execution_zone,
+            },
+            contracts=[
+                {"field": "task_type", "value": task.task_plan.task_type},
+                {"field": "status", "value": task.status},
+                {"field": "risk", "value": task.task_plan.risk},
+                {"field": "data_classification", "value": task.task_plan.data_classification},
+                {"field": "execution_zone", "value": task.task_plan.execution_zone},
+                {"field": "network_policy", "value": task.task_plan.network_policy},
+                {"field": "required_capabilities", "value": list(task.task_plan.required_capabilities)},
+            ],
+            conflicts=[],
+            evidence=[],
+        )
+
+    def _record(self, resource: CatalogResource, *, resolved: bool) -> None:
+        self._attempts += 1
+        rtype = resource.resource_type or "unknown"
+        bucket = self._resolved if resolved else self._unknown
+        bucket[rtype] = bucket.get(rtype, 0) + 1
+        if not resource.authority.get("endpoint_or_path"):
+            self._missing["authority_endpoint_or_path"] += 1
+        if resource.authority.get("revision") is None:
+            self._missing["authority_revision"] += 1
+        for name in _RUNTIME_SOURCE_FIELDS:
+            if resource.runtime.get(name) is None:
+                self._missing[name] += 1
+        if all(resource.runtime.get(name) is None for name in _RUNTIME_SOURCE_FIELDS):
+            self._runtime_source_unknown += 1
@@ -0,0 +1,193 @@
+"""Tests for the read-only Runtime / Authority Catalog (authority_catalog)."""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+
+ROOT = os.getcwd()
+if os.path.join(ROOT, "active-runtime") not in sys.path:
+    sys.path.append(os.path.join(ROOT, "active-runtime"))
+
+from app.task_plane.authority_catalog import AuthorityCatalog, CatalogResource, SCHEMA_VERSION
+from app.task_plane.models import WorkerRegistration
+from app.task_plane.store import TaskPlaneStore
+
+WORKER_DICT = {
+    "worker_id": "worker-cat-1",
+    "worker_type": "codex.readonly",
+    "identity": "alice@example.test",
+    "capabilities": ["code.change.apply", "shell.exec"],
+    "trust_zone": "isolated-development",
+    "max_concurrency": 1,
+    "heartbeat_interval_seconds": 15,
+    "supported_protocols": ["aios-task-v1"],
+}
+
+
+def test_no_store_returns_complete_null_shape_without_exception():
+    catalog = AuthorityCatalog(store=None)
+    resource = catalog.resolve_worker("worker-whatever")
+    assert isinstance(resource, CatalogResource)
+    assert resource.schema_version == SCHEMA_VERSION
+    assert resource.resource_type == "worker"
+    assert resource.resource_id == "worker-whatever"
+    # Structurally complete: all required blocks present.
+    for key in ("authority", "runtime", "contracts", "conflicts", "evidence"):
+        assert key in resource.to_dict()
+    # Nothing fabricated: observation fields are null, evidence empty.
+    assert resource.authority["endpoint_or_path"] is None
+    assert resource.authority["revision"] is None
+    assert resource.runtime["service"] is None
+    assert resource.runtime["pid"] is None
+    assert resource.runtime["loaded_source_path"] is None
+    assert resource.runtime["source_commit"] is None
+    assert resource.runtime["artifact_sha256"] is None
+    assert resource.evidence == []
+    # It is recorded as unknown, not resolved.
+    summary = catalog.summary()
+    assert summary["unknown"].get("worker") == 1
+    assert "worker" not in summary["resolved"]
+
+
+def test_resolve_worker_returns_worker_id_and_capabilities():
+    store = TaskPlaneStore()
+    reg = WorkerRegistration.from_dict(WORKER_DICT)
+    store.register_worker(reg)
+    catalog = AuthorityCatalog(store)
+    resource = catalog.resolve_worker("worker-cat-1")
+    assert resource.resource_type == "worker"
+    assert resource.resource_id == "worker-cat-1"
+    assert resource.authority["endpoint_or_path"] == "alice@example.test"
+    assert resource.authority["revision"] == 1
+    assert resource.runtime["service"] == "codex.readonly"
+    caps = [c["value"] for c in resource.contracts if c["field"] == "capabilities"][0]
+    assert "shell.exec" in caps
+    # Not flagged as runtime source unknown because authority resolved:
+    # but runtime source fields are still null for a registration (honest).
+    summary = catalog.summary()
+    assert summary["resolved"].get("worker") == 1
+    assert summary["runtime_source_unknown"] >= 1
+
+
+def test_resolve_worker_missing_record_is_unknown_not_live():
+    store = TaskPlaneStore()
+    catalog = AuthorityCatalog(store)
+    resource = catalog.resolve_worker("does-not-exist")
+    assert resource.resource_id == "does-not-exist"
+    assert resource.authority["endpoint_or_path"] is None
+    assert resource.authority["revision"] is None
+    assert resource.evidence == []
+    assert catalog.summary()["unknown"].get("worker") == 1
+
+
+def test_resolve_generic_unknown_type_is_empty_unknown():
+    catalog = AuthorityCatalog(store=None)
+    resource = catalog.resolve_generic("release", "rel-9")
+    assert resource.resource_type == "release"
+    assert resource.resource_id == "rel-9"
+    assert resource.authority["kind"] == "release"
+    assert resource.authority["endpoint_or_path"] is None
+    assert resource.authority["revision"] is None
+    assert resource.runtime["loaded_source_path"] is None
+    assert resource.evidence == []
+    assert catalog.summary()["unknown"].get("release") == 1
+    # No live state fabricated: nothing was provisioned as a runtime service.
+    assert resource.runtime["service"] is None
+    assert resource.runtime["state_root"] is None
+
+
+def test_to_dict_is_json_serializable():
+    store = TaskPlaneStore()
+    store.register_worker(WorkerRegistration.from_dict(WORKER_DICT))
+    catalog = AuthorityCatalog(store)
+    resource = catalog.resolve_worker("worker-cat-1")
+    payload = resource.to_dict()
+    # Round-trips through JSON without error and preserves values.
+    text = json.dumps(payload, ensure_ascii=False)
+    restored = json.loads(text)
+    assert restored == payload
+    assert restored["schema_version"] == SCHEMA_VERSION
+    assert restored["resource_id"] == "worker-cat-1"
+    assert restored["authority"]["revision"] == 1
+
+
+def test_resolve_is_read_only_no_store_write():
+    store = TaskPlaneStore()
+    store.register_worker(WorkerRegistration.from_dict(WORKER_DICT))
+
+    before_tasks = [t.to_dict() for t in store.list_tasks()]
+    before_events = {
+        t.task_id: store.list_events(t.task_id) for t in store.list_tasks()
+    }
+
+    catalog = AuthorityCatalog(store)
+    catalog.resolve_worker("worker-cat-1")
+    catalog.resolve_worker("missing-worker")
+    catalog.resolve_repository("repo/main", authority_ref="oid-abcd" if False else None)
+    catalog.resolve_generic("schema", "s1")
+
+    after_tasks = [t.to_dict() for t in store.list_tasks()]
+    after_events = {
+        t.task_id: store.list_events(t.task_id) for t in store.list_tasks()
+    }
+    assert before_tasks == after_tasks
+    assert before_events == after_events
+
+
+def test_catalog_exposes_no_write_path():
+    write_primitives = {
+        "register_worker",
+        "create_task",
+        "claim_task",
+        "append_event",
+        "add_evidence",
+        "save_receipt",
+        "heartbeat_worker",
+        "disable_worker",
+        "restore_worker",
+        "update_task_status",
+        "record_progress",
+        "complete_claim",
+    }
+    catalog = AuthorityCatalog(store=None)
+    public = {name for name in vars(type(catalog)) if not name.startswith("_")}
+    # The catalog only exposes resolve + summary; it must not carry any
+    # store-write entry point of its own.
+    assert not (public & write_primitives)
+    assert public <= {"resolve_worker", "resolve_task", "resolve_repository", "resolve_generic", "summary"}
+
+
+def test_repository_without_ref_is_unknown_with_ref_reflected():
+    catalog = AuthorityCatalog(store=None)
+    empty = catalog.resolve_repository("repo/main")
+    assert empty.authority["revision"] is None
+    # Original caller still gets the repository's identifier as an endpoint
+    # hint, but the OID (revision) is unknown until a live ref is supplied.
+    assert empty.authority["revision"] is None
+
+    catalog2 = AuthorityCatalog(store=None)
+    with_ref = catalog2.resolve_repository("repo/main", authority_ref="de44a3f01beef")
+    assert with_ref.authority["revision"] == "de44a3f01beef"
+    assert with_ref.authority["endpoint_or_path"] == "repo/main"
+    assert catalog2.summary()["resolved"].get("repository") == 1
+
+
+def test_summary_tracks_runtime_source_unknown():
+    store = TaskPlaneStore()
+    store.register_worker(WorkerRegistration.from_dict(WORKER_DICT))
+    catalog = AuthorityCatalog(store)
+    catalog.resolve_worker("worker-cat-1")          # resolved, but source fields null
+    catalog.resolve_worker("nope")                 # unknown
+    catalog.resolve_generic("release", "r1")       # unknown
+    summary = catalog.summary()
+    assert summary["resolve_attempts"] == 3
+    assert summary["resolved"].get("worker") == 1
+    assert summary["unknown"].get("worker") == 1
+    assert summary["unknown"].get("release") == 1
+    # All three resolved/unknown entries lack loaded source fields.
+    assert summary["runtime_source_unknown"] == 3
+    assert summary["missing_fields"]["loaded_source_path"] == 3
+    assert summary["missing_fields"]["source_commit"] == 3
+    assert summary["missing_fields"]["artifact_sha256"] == 3