GRAPHOS-OPS-R022.2.1: evolution.loops.status read op (R022.2 split)

.config/dupehound-distinct.toml

@@ -36,9 +36,9 @@ left_name = "operations"
 left_digest = "sha256:1dd92a3cf607864448698507dd859bedcd0b2837cddb97a0aeef56d1adc0a706"
 right_file = "graph_os/api/ops/ingest.py"
 right_name = "operations"
-right_digest = "sha256:ee097ee89d61f7ddcec595ea9f1a06767ad045baf374b12e808b69399a111d71"
+right_digest = "sha256:b34d30a8a6914622efd841d8584e31cac462f51acf90ab8f9a1e51992dbe2193"
 reason = "Both declare an unrelated domain's own OpSpec registry (operational-admin vs ingestion) using the one mandated declarative registration shape every ops module shares by design; the registry factory, not this duplication, is the single point of consolidation."
-reviewed_on = "2026-10-09"
+reviewed_on = "2026-10-10"
 
 [[pair]]
 left_file = "graph_os/api/ops/memory.py"

graph_os/api/ops/ingest.py

@@ -25,8 +25,10 @@
     Verb,
 )
 from graph_os.ingest.service import (
+    get_job_status,
     get_source_status,
     index_repository,
+    list_packs,
     list_sources,
     sync_source,
 )
@@ -38,8 +40,10 @@
 #: (``graph_os.ingest.service``).
 __all__ = [
     "operations",
+    "get_job_status",
     "get_source_status",
     "index_repository",
+    "list_packs",
     "list_sources",
     "sync_source",
 ]
@@ -82,6 +86,22 @@ class IngestSourceStatusResult(_Params):
     value: dict[str, str]
 
 
+class IngestPacksListParams(_Params):
+    pass
+
+
+class IngestPackListResult(_Params):
+    value: dict[str, object]
+
+
+class IngestJobStatusParams(_Params):
+    job_id: str = Field(min_length=1, max_length=256)
+
+
+class IngestJobStatusResult(_Params):
+    value: dict[str, str]
+
+
 def operations() -> tuple[OpSpec, ...]:
     return (
         OpSpec(
@@ -132,4 +152,26 @@ def operations() -> tuple[OpSpec, ...]:
             scopes=frozenset({"ingest:read"}),
             effect=Effect.READ,
         ),
+        OpSpec(
+            id="ingest.packs.list",
+            verb=Verb.FIND,
+            summary="List this tenant's ingestion packs",
+            examples=("list my ingestion packs",),
+            params=IngestPacksListParams,
+            result=IngestPackListResult,
+            binding=Composite(handler="graph_os.api.ops.ingest.list_packs"),
+            scopes=frozenset({"ingest:read"}),
+            effect=Effect.READ,
+        ),
+        OpSpec(
+            id="ingest.jobs.status",
+            verb=Verb.ASK,
+            summary="Show one durable ingestion job's current status",
+            examples=("show the status of this ingestion job",),
+            params=IngestJobStatusParams,
+            result=IngestJobStatusResult,
+            binding=Composite(handler="graph_os.api.ops.ingest.get_job_status"),
+            scopes=frozenset({"ingest:read"}),
+            effect=Effect.READ,
+        ),
     )

graph_os/fleet/mcp_resource_reconciliation.py

@@ -106,3 +106,42 @@ def gate_mcp_resource_publication(
         return _unreconciled("stale", receipt)
 
     return _reconciled(receipt)
+
+
+class PublicationRefusedError(Exception):
+    """Raised when a candidate generation may not be swapped in as published.
+
+    Carries the typed ``PublicationGateResult`` so the caller can surface the
+    stable ``reason`` instead of swallowing the refusal.
+    """
+
+    def __init__(self, result: PublicationGateResult) -> None:
+        self.result = result
+        super().__init__(f"{result.code}: publication refused (reason={result.reason})")
+
+
+def require_reconciled_for_swap(
+    receipt: ReconciliationReceipt | None,
+    *,
+    tenant_id: str,
+    expected_generation: int,
+    expected_digest: str,
+    now_ms: int,
+    max_age_ms: int,
+) -> ReconciliationReceipt:
+    """Return the matching receipt, or raise ``PublicationRefusedError``.
+
+    Wraps ``gate_mcp_resource_publication`` so the publish step cannot bypass
+    the gate by ignoring a returned status.
+    """
+    result = gate_mcp_resource_publication(
+        receipt,
+        tenant_id=tenant_id,
+        expected_generation=expected_generation,
+        expected_digest=expected_digest,
+        now_ms=now_ms,
+        max_age_ms=max_age_ms,
+    )
+    if result.status != "reconciled" or result.receipt is None:
+        raise PublicationRefusedError(result)
+    return result.receipt

graph_os/fleet/multiplexer.py

@@ -4917,15 +4917,15 @@ async def _probe_protocol_families(
                 errors["prompts"] = type(exc).__name__
             native_prompts = []
 
-        await self._harvest_resource_bodies(
+        await self._resolve_via_local_catalog_or_harvest(
             server_name,
             session,
             skills,
             _SKILL_HARVEST_SPEC,
             self._read_skill_body,
             probe_deadline=probe_deadline,
         )
-        await self._harvest_resource_bodies(
+        await self._resolve_via_local_catalog_or_harvest(
             server_name,
             session,
             prompt_resources,

graph_os/identity/api_keys.py

@@ -6,7 +6,9 @@
 
 from __future__ import annotations
 
+from collections.abc import Collection
 from dataclasses import dataclass
+from datetime import datetime
 
 from .engine import IdentityUnavailable
 
@@ -20,6 +22,7 @@ class ApiKeyGrant:
     key_id: str
     owner_principal_id: str
     scopes: frozenset[str]
+    expires_at: datetime | None = None
 
     def __post_init__(self) -> None:
         if not isinstance(self.key_id, str) or not self.key_id:
@@ -31,3 +34,21 @@ def __post_init__(self) -> None:
             raise IdentityUnavailable(
                 f"API keys may not carry approver-class scope(s): {sorted(rejected)}"
             )
+
+
+def effective_scopes(
+    grant: ApiKeyGrant,
+    owner_scopes: frozenset[str],
+    *,
+    now: datetime,
+    revoked_key_ids: Collection[str] = (),
+) -> frozenset[str]:
+    """Scopes usable right now: key scopes intersected with the owner's current scopes.
+
+    Fails closed: a revoked or expired key is refused immediately.
+    """
+    if grant.key_id in revoked_key_ids:
+        raise IdentityUnavailable("API key has been revoked")
+    if grant.expires_at is not None and now >= grant.expires_at:
+        raise IdentityUnavailable("API key has expired")
+    return (grant.scopes & owner_scopes) - _APPROVER_SCOPES

graph_os/identity/composition.py

@@ -0,0 +1,58 @@
+"""Identity composition root: build the concrete runtime once, expose only ports.
+
+Dependents receive ``CredentialAuthority`` and ``SessionAuthority`` objects and
+never import a concrete implementation. Absent authoritative facts or a
+malformed authority answer fail closed with ``IdentityUnavailable``.
+"""
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from typing import Any
+
+from .engine import IdentityUnavailable
+from .ports import (
+    CredentialAuthority,
+    CredentialState,
+    SessionAuthority,
+    SessionState,
+)
+
+
+@dataclass(frozen=True, slots=True)
+class IdentityPorts:
+    """The only identity surface handed to dependents."""
+
+    credentials: CredentialAuthority
+    sessions: SessionAuthority
+
+
+class _CheckedRuntime:
+    """Port-typed view over the single concrete runtime; validates answers."""
+
+    def __init__(self, runtime: Any) -> None:
+        self._runtime = runtime
+
+    async def resolve_credential(self, credential: str) -> CredentialState:
+        state = await self._runtime.resolve_credential(credential)
+        if not isinstance(state, CredentialState):
+            raise IdentityUnavailable("credential authority returned no state")
+        return state
+
+    async def resolve_session(self, credential: str) -> SessionState:
+        state = await self._runtime.resolve_session(credential)
+        if not isinstance(state, SessionState):
+            raise IdentityUnavailable("session authority returned no state")
+        return state
+
+
+def build_identity_ports(
+    credentials: Any,
+    sessions: Any,
+    *,
+    runtime_factory: Callable[[Any, Any], Any],
+) -> IdentityPorts:
+    """Construct the concrete runtime exactly once and return port-typed views."""
+    if credentials is None or sessions is None:
+        raise IdentityUnavailable("authoritative identity sources required")
+    runtime = _CheckedRuntime(runtime_factory(credentials, sessions))
+    return IdentityPorts(credentials=runtime, sessions=runtime)

graph_os/ingest/service.py

@@ -66,6 +66,34 @@ class IngestSourceInventory(BaseModel):
     sources: tuple[IngestSourceRecord, ...]
 
 
+class IngestPackRecord(BaseModel):
+    """One ingestion pack's identity and current state."""
+
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    pack_id: str
+    tenant: str
+    state: IngestSourceState
+
+
+class IngestPackInventory(BaseModel):
+    """A tenant's full ingestion-pack inventory."""
+
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    packs: tuple[IngestPackRecord, ...]
+
+
+class IngestJobRecord(BaseModel):
+    """One durable ingestion job's identity and current state."""
+
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    job_id: str
+    tenant: str
+    status: IngestJobStatus
+
+
 @runtime_checkable
 class IngestRunner(Protocol):
     """The one typed port GraphOS calls the ingestion SDK through.
@@ -90,6 +118,10 @@ async def get_source_status(
         self, *, tenant: str, source_id: str
     ) -> IngestSourceRecord: ...
 
+    async def list_packs(self, *, tenant: str) -> IngestPackInventory: ...
+
+    async def get_job_status(self, *, tenant: str, job_id: str) -> IngestJobRecord: ...
+
 
 def _bound_runner(context: Any) -> IngestRunner:
     runner = context.services.get("ingest_runner")
@@ -142,3 +174,19 @@ async def get_source_status(context: Any, params: Mapping[str, Any], op: Any) ->
         tenant=context.caller.tenant, source_id=params["source_id"]
     )
     return {"value": record.model_dump(mode="json")}
+
+
+async def list_packs(context: Any, params: Mapping[str, Any], op: Any) -> Any:
+    """List this tenant's ingestion packs through the composed runner."""
+    runner = _bound_runner(context)
+    inventory = await runner.list_packs(tenant=context.caller.tenant)
+    return {"value": inventory.model_dump(mode="json")}
+
+
+async def get_job_status(context: Any, params: Mapping[str, Any], op: Any) -> Any:
+    """Report one durable ingestion job's state through the composed runner."""
+    runner = _bound_runner(context)
+    record = await runner.get_job_status(
+        tenant=context.caller.tenant, job_id=params["job_id"]
+    )
+    return {"value": record.model_dump(mode="json")}

graph_os/mcp_server/legacy_action_mapping.py

@@ -82,5 +82,6 @@ def validate_served_actions(
         ActionOperationMapping("vault_sync", "secret_vault.vault_sync"),
         ActionOperationMapping("install_hooks", "graph_loops.install_hooks"),
         ActionOperationMapping("uninstall_hooks", "graph_loops.uninstall_hooks"),
+        ActionOperationMapping("sync", "ingest.sources.sync"),
     )
 )

specs/fleet-catalog-and-tools/status.json

@@ -324,8 +324,8 @@
         "a94c87aa090e"
       ],
       "verified_by": [
-        "tests/fleet/test_harvest_inventory.py:106",
-        "tests/fleet/test_harvest_inventory.py:117"
+        "tests/fleet/test_harvest_inventory.py:105",
+        "tests/fleet/test_harvest_inventory.py:116"
       ]
     },
     {
@@ -360,9 +360,14 @@
     {
       "id": "GRAPHOS-FLEET-R006.3.1",
       "title": "`_probe_protocol_families` harvest calls routed through the cutover helper",
-      "delivery_state": "SPECIFIED",
-      "landed_in": [],
-      "verified_by": []
+      "delivery_state": "VERIFIED",
+      "landed_in": [
+        "ddb10c07995e"
+      ],
+      "verified_by": [
+        "tests/fleet/test_harvest_inventory.py:128",
+        "tests/fleet/test_harvest_inventory.py:135"
+      ]
     },
     {
       "id": "GRAPHOS-FLEET-R006.3.2",

specs/graphos-a2a-002/status.json

@@ -64,9 +64,13 @@
     {
       "id": "GRAPHOS-A2A-002-R005",
       "title": "No advisory, cached, or partial substitute",
-      "delivery_state": "SPECIFIED",
-      "landed_in": [],
-      "verified_by": []
+      "delivery_state": "VERIFIED",
+      "landed_in": [
+        "63762147f0bc"
+      ],
+      "verified_by": [
+        "tests/a2a/test_admission.py:52"
+      ]
     },
     {
       "id": "GRAPHOS-A2A-002-R006",