GRAPHOS-IDENTITY-R012.2.3.1: SAML replay check

.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:7a6286e3952d0525803507b1a6ddbcd6c8023a3b65fd2305c93369b41e3ae149"
 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/api/ops/registry_factory.py

@@ -32,6 +32,7 @@ def get_registry() -> Registry:
         memory,
         ops,
         policy,
+        retrieval,
         security,
         swarm,
         telemetry,
@@ -51,6 +52,7 @@ def get_registry() -> Registry:
         memory,
         ops,
         policy,
+        retrieval,
         security,
         swarm,
         telemetry,

graph_os/api/ops/retrieval.py

@@ -0,0 +1,150 @@
+"""Retrieval read operations (GRAPHOS-OPS-R021.2.1, retrieval slice).
+
+``retrieval.search`` runs a context-budgeted retrieval through one typed port
+under the caller's own tenant and authority; there is no local cache
+substitute. An uncomposed retriever fails closed with a typed ``UNAVAILABLE``
+refusal, matching :mod:`graph_os.api.ops.ingest` and
+:mod:`graph_os.api.ops.memory`.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any, Protocol, runtime_checkable
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from graph_os.api.ops._common import Params, bound_service, build_read_op
+from graph_os.api.registry import OpSpec, Verb
+
+
+class RetrievalSearchParams(Params):
+    query: str = Field(min_length=1, max_length=4096)
+    context_budget: int = Field(default=4096, ge=1, le=1_000_000)
+
+
+class RetrievalHit(BaseModel):
+    """One retrieved passage with its provenance reference."""
+
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    ref: str
+    text: str
+    tokens: int = Field(ge=0)
+
+
+class RetrievalSearchOutcome(BaseModel):
+    """Budget-bounded hits for one tenant-scoped query."""
+
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    tenant: str
+    hits: tuple[RetrievalHit, ...]
+
+
+class RetrievalSearchResult(Params):
+    value: dict[str, Any]
+
+
+class RetrievalFreshnessParams(Params):
+    """``retrieval.freshness`` takes no parameters; the tenant is the caller's."""
+
+
+class RetrievalFreshnessOutcome(BaseModel):
+    """Freshness of one tenant's retrieval sources."""
+
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    tenant: str
+    sources: dict[str, str]
+
+
+class RetrievalFreshnessResult(Params):
+    value: dict[str, Any]
+
+
+@runtime_checkable
+class Retriever(Protocol):
+    """The one typed port GraphOS calls the engine retrieval through."""
+
+    async def search(
+        self, *, tenant: str, query: str, context_budget: int
+    ) -> RetrievalSearchOutcome: ...
+
+    async def freshness(self, *, tenant: str) -> RetrievalFreshnessOutcome: ...
+
+
+async def handle_retrieval_search(
+    context: Any, params: Mapping[str, Any], op: OpSpec
+) -> dict[str, Any]:
+    """Return budget-bounded hits for the caller's own tenant."""
+    retriever: Retriever = bound_service(
+        context, "retriever", reason="retriever is not composed"
+    )
+    budget = params["context_budget"]
+    outcome = await retriever.search(
+        tenant=context.caller.tenant, query=params["query"], context_budget=budget
+    )
+    kept: list[RetrievalHit] = []
+    used = 0
+    for hit in outcome.hits:
+        if used + hit.tokens > budget:
+            break
+        used += hit.tokens
+        kept.append(hit)
+    bounded = outcome.model_copy(update={"hits": tuple(kept)})
+    return {"value": bounded.model_dump(mode="json")}
+
+
+async def handle_retrieval_freshness(
+    context: Any, params: Mapping[str, Any], op: OpSpec
+) -> dict[str, Any]:
+    """Return source freshness for the caller's own tenant."""
+    retriever: Retriever = bound_service(
+        context, "retriever", reason="retriever is not composed"
+    )
+    outcome = await retriever.freshness(tenant=context.caller.tenant)
+    return {"value": outcome.model_dump(mode="json")}
+
+
+def operations() -> tuple[OpSpec, ...]:
+    return (
+        build_read_op(
+            verb=Verb.FIND,
+            op_id="retrieval.search",
+            summary="Retrieve passages for a query within a context budget",
+            examples=("find passages about this topic",),
+            params=RetrievalSearchParams,
+            result=RetrievalSearchResult,
+            handler="graph_os.api.ops.retrieval.handle_retrieval_search",
+            scope="memory:read",
+        ),
+        build_read_op(
+            verb=Verb.FIND,
+            op_id="retrieval.freshness",
+            summary="Report freshness of the caller's retrieval sources",
+            examples=("how fresh are my retrieval sources",),
+            params=RetrievalFreshnessParams,
+            result=RetrievalFreshnessResult,
+            handler="graph_os.api.ops.retrieval.handle_retrieval_freshness",
+            scope="memory:read",
+        ),
+    )
+
+
+specs = operations
+
+__all__ = [
+    "RetrievalFreshnessOutcome",
+    "RetrievalFreshnessParams",
+    "RetrievalFreshnessResult",
+    "RetrievalHit",
+    "RetrievalSearchOutcome",
+    "RetrievalSearchParams",
+    "RetrievalSearchResult",
+    "Retriever",
+    "handle_retrieval_freshness",
+    "handle_retrieval_search",
+    "operations",
+    "specs",
+]

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