fix: serve discovery census from durable snapshot

docs/INTERFACE.md

@@ -1,6 +1,6 @@
 # Agent Guild — machine interface (GENERATED)
 
-*Generated from `live/guild/contract/contract.json` v3 (service 2.6.2). Do not edit by hand — run `make contract`.*
+*Generated from `live/guild/contract/contract.json` v3 (service 2.6.3). Do not edit by hand — run `make contract`.*
 
 - Host: https://agent-guild-5d5r.onrender.com
 - MCP (streamable HTTP): https://agent-guild-5d5r.onrender.com/mcp/

live/guild/app/__init__.py

@@ -3,7 +3,12 @@
 # Single source of truth for the service version. Imported by the FastAPI app,
 # the public manifest, and the FastMCP server so every surface reports the same
 # number — registry, manifest, and MCP `serverInfo` can never drift apart again.
-__version__ = "2.6.2"   # PATCH release recovery (2026-08-31): retries a
+__version__ = "2.6.3"   # PATCH live census serving (2026-08-31): builds the
+                        # complete durable signed discovery census before
+                        # readiness, serves its immutable declared snapshot
+                        # without a request-time full-history scan, and
+                        # refreshes it on each scout cycle. History of 2.6.2:
+                        # PATCH release recovery (2026-08-31): retries a
                         # boot-time persisted scout lease without consuming
                         # the six-hour interval, and waits boundedly for
                         # GitHub branch-policy convergence before merging a

live/guild/app/main.py

@@ -63,7 +63,7 @@
 from . import ard
 from .state import store
 from . import paidcatalog
-from .store import CanonicalWriteRefused
+from .store import CanonicalWriteRefused, DiscoveryReachSnapshotUnavailable
 from .reachability import url_policy_check
 from . import abuse
 from . import coordination
@@ -154,6 +154,17 @@ async def _lifespan(app: "FastAPI"):
         swarm_ensure_built()
     except Exception as exc:
         _log.warning("swarm identity build skipped: %s", exc)
+    try:
+        # The signed discovery census reduces the COMPLETE durable event
+        # history. Build it before readiness so public reads and the release
+        # probe never rescan a six-figure SQLite table on the request path.
+        if store.discovery_reach_cache_enabled():
+            census = store.refresh_discovery_reach_cache()
+            _log.info("discovery reach snapshot: %s", census)
+    except Exception as exc:
+        # Do not make an observability derivative a boot-kill switch; the live
+        # contract gate will fail closed if the endpoint cannot serve.
+        _log.warning("discovery reach snapshot skipped: %s", exc)
     # x402 rail: FAIL CLOSED at startup. A MAINNET rail that is misconfigured
     # (unauthenticated facilitator, missing CDP credentials, wrong USDC
     # contract, invalid recipient, local resource origin, no independent
@@ -6025,7 +6036,19 @@ def discovery_reach():
     The target counter is deduplicated by privacy-safe actor and excludes
     first-party calls, tests, generic tools, registries and unlinkable traffic.
     """
-    return store.discovery_reach(target=25_000)
+    try:
+        return store.discovery_reach(target=25_000)
+    except DiscoveryReachSnapshotUnavailable as exc:
+        raise HTTPException(
+            status_code=503,
+            detail={
+                "error": "discovery_snapshot_unavailable",
+                "available_snapshot_events": (
+                    exc.available_snapshot_events),
+                "retryable": True,
+            },
+            headers={"Retry-After": "5"},
+        ) from exc
 
 
 @app.get("/discovery/reach/evidence")
@@ -6034,9 +6057,26 @@ def discovery_reach_evidence(
         limit: int = Query(500, ge=1, le=2_000),
         snapshot_events: Optional[int] = Query(None, ge=0)):
     """Replayable, double-pseudonymised actor rows committed by the proof."""
-    report = store.discovery_reach(
-        target=25_000, include_actor_evidence=True,
-        snapshot_events=snapshot_events)
+    try:
+        report = store.discovery_reach(
+            target=25_000, include_actor_evidence=True,
+            snapshot_events=snapshot_events)
+    except DiscoveryReachSnapshotUnavailable as exc:
+        # Evidence rows are committed to one exact signed snapshot. Tell the
+        # caller only which numeric snapshot is currently replayable; never
+        # fall back to a request-time historical scan.
+        raise HTTPException(
+            status_code=(409 if exc.available_snapshot_events is not None
+                         else 503),
+            detail={
+                "error": "discovery_snapshot_unavailable",
+                "available_snapshot_events": (
+                    exc.available_snapshot_events),
+                "retryable": exc.available_snapshot_events is None,
+            },
+            headers=({"Retry-After": "5"}
+                     if exc.available_snapshot_events is None else None),
+        ) from exc
     rows = report.pop("actor_evidence")
     page = rows[offset:offset + limit]
     return {

live/guild/app/store.py

@@ -10,6 +10,7 @@
 """
 from __future__ import annotations
 
+import copy
 import hashlib
 import json
 import os
@@ -62,6 +63,22 @@ def _iso_age_seconds(ts: Optional[str]) -> float:
 EVENT_RETENTION_TRIGGER = 50000
 EVENT_RETENTION_TARGET = 25000
 
+# The signed discovery census is a complete durable-history reduction, not a
+# request-time counter.  Keep its fully replayable snapshot in the existing
+# durable swarm_state KV so serving it never rescans a six-figure event table.
+DISCOVERY_REACH_CACHE_KEY = "discovery_reach_cache"
+DISCOVERY_REACH_CACHE_VERSION = 1
+
+
+class DiscoveryReachSnapshotUnavailable(ValueError):
+    """A warm census is missing, or a caller requested an older snapshot."""
+
+    def __init__(self, *, requested_snapshot_events: Optional[int],
+                 available_snapshot_events: Optional[int]):
+        self.requested_snapshot_events = requested_snapshot_events
+        self.available_snapshot_events = available_snapshot_events
+        super().__init__("discovery reach snapshot is unavailable")
+
 
 # Keepalive event dedup window (seconds). Agents that re-declare the SAME
 # endpoint on a timer (e.g. the market worker re-verifies every ~2 min so
@@ -193,6 +210,7 @@ class Store:
     def __init__(self, path: Optional[str] = None):
         self.path = path or os.environ.get("GUILD_DATA", "")
         self.lock = threading.RLock()
+        self._discovery_reach_refresh_lock = threading.Lock()
         self.agents: dict[str, dict[str, Any]] = {}
         self.tasks: dict[str, dict[str, Any]] = {}
         self.attestations: list[dict[str, Any]] = []
@@ -3045,10 +3063,120 @@ def _visit_count(events: list[dict[str, Any]]) -> int:
             "measurement_coverage": coverage,
         }
 
+    def discovery_reach_cache_enabled(self) -> bool:
+        """Production SQLite defaults to a precomputed signed census.
+
+        JSON's retained tail is small enough to compute directly.  Tests and
+        operators can disable warming explicitly without changing the census
+        rules or the on-demand calculation.
+        """
+        raw = (os.environ.get("GUILD_DISCOVERY_REACH_CACHE") or "").strip()
+        if raw:
+            return raw == "1"
+        return self.backend is not None
+
+    def _cached_discovery_reach(
+            self, target: int, snapshot_events: Optional[int]
+    ) -> Optional[dict[str, Any]]:
+        with self.lock:
+            cached = self.swarm_state.get(DISCOVERY_REACH_CACHE_KEY)
+            cached = copy.deepcopy(cached) if isinstance(cached, dict) else None
+        if (not cached
+                or cached.get("version") != DISCOVERY_REACH_CACHE_VERSION
+                or int(cached.get("target") or 0) != target
+                or not isinstance(cached.get("report"), dict)):
+            return None
+        cached_rows = int(cached.get("snapshot_events") or 0)
+        if snapshot_events is not None and snapshot_events != cached_rows:
+            return None
+        return cached
+
     def discovery_reach(self, target: int = 25_000, *,
                         include_actor_evidence: bool = False,
                         snapshot_events: Optional[int] = None
                         ) -> dict[str, Any]:
+        """Return a signed census snapshot, using the durable warm cache.
+
+        The cache is itself an exact output of the complete-history reducer:
+        its proof declares ``event_snapshot_rows`` and ``as_of``, and the
+        evidence endpoint replays that same immutable snapshot.  New events do
+        not silently alter an issued proof; startup and every scout cycle
+        publish a new complete snapshot.
+        """
+        target = max(1, int(target))
+        cached = self._cached_discovery_reach(target, snapshot_events)
+        if cached is not None:
+            report = cached["report"]
+            if not include_actor_evidence:
+                report.pop("actor_evidence", None)
+            report["snapshot_cache"] = {
+                "built_at": cached.get("built_at"),
+                "event_snapshot_rows": cached.get("snapshot_events"),
+                "refresh_policy": "service_start_and_scout_cycle",
+            }
+            return report
+        if self.discovery_reach_cache_enabled():
+            # A cache-enabled deployment must never turn a stale evidence
+            # replay or a failed warm-up into an attacker-triggerable full
+            # history scan on the request path. Refresh is an explicit
+            # startup/scout operation; public reads fail quickly and safely.
+            available = self._cached_discovery_reach(target, None)
+            raise DiscoveryReachSnapshotUnavailable(
+                requested_snapshot_events=snapshot_events,
+                available_snapshot_events=(
+                    int(available.get("snapshot_events") or 0)
+                    if available is not None else None),
+            )
+        return self._compute_discovery_reach(
+            target=target, include_actor_evidence=include_actor_evidence,
+            snapshot_events=snapshot_events)
+
+    def refresh_discovery_reach_cache(self, target: int = 25_000
+                                      ) -> dict[str, Any]:
+        """Rebuild and durably publish one complete signed census snapshot.
+
+        Only one thread computes at a time.  Existing readers keep receiving
+        the previous immutable snapshot while a refresh is in progress.
+        """
+        if not self._discovery_reach_refresh_lock.acquire(blocking=False):
+            cached = self._cached_discovery_reach(max(1, int(target)), None)
+            return {
+                "refreshed": False,
+                "reason": "refresh_already_running",
+                "snapshot_events": (cached or {}).get("snapshot_events"),
+            }
+        try:
+            target = max(1, int(target))
+            report = self._compute_discovery_reach(
+                target=target, include_actor_evidence=True)
+            snapshot_events = int(report["proof"]["payload"][
+                "event_snapshot_rows"])
+            cache = {
+                "version": DISCOVERY_REACH_CACHE_VERSION,
+                "target": target,
+                "built_at": _now(),
+                "snapshot_events": snapshot_events,
+                "report": report,
+            }
+            with self.lock, self._txn():
+                self.swarm_state[DISCOVERY_REACH_CACHE_KEY] = cache
+                if self.backend is not None:
+                    self._persist_kv("swarm_state", self.swarm_state)
+                self._save()
+            return {
+                "refreshed": True,
+                "built_at": cache["built_at"],
+                "snapshot_events": snapshot_events,
+                "qualified_distinct_autonomous_agents": report[
+                    "qualified_distinct_autonomous_agents"],
+            }
+        finally:
+            self._discovery_reach_refresh_lock.release()
+
+    def _compute_discovery_reach(self, target: int = 25_000, *,
+                                 include_actor_evidence: bool = False,
+                                 snapshot_events: Optional[int] = None
+                                 ) -> dict[str, Any]:
         """Durable proof of DISTINCT autonomous-agent discovery.
 
         A catalogue hit is not an agent and six paid catalogue rows are not

live/guild/app/swarm/runner.py

@@ -442,6 +442,16 @@ def run_once(store: Any, fetch: Callable = scout.safe_fetch_json,
             index_summary = _run_index_cycle(store)
         except Exception as exc:  # noqa: BLE001
             index_summary = {"error": type(exc).__name__}
+        if store.discovery_reach_cache_enabled():
+            try:
+                # Refresh the complete signed census on the same bounded,
+                # lease-guarded six-hour cadence. Readers continue receiving
+                # the previous immutable snapshot until this one is published.
+                index_summary["discovery_reach_cache"] = \
+                    store.refresh_discovery_reach_cache()
+            except Exception as exc:  # noqa: BLE001
+                index_summary["discovery_reach_cache"] = {
+                    "error": type(exc).__name__}
         zero_demand = not summary.get("capabilities")
         # ACK only the capabilities this cycle actually processed — demand
         # that arrived mid-run stays queued for the next cycle.

live/guild/contract/contract.json

@@ -1195,6 +1195,6 @@
   "payment_safety_mcp_card": "https://agent-guild-5d5r.onrender.com/.well-known/mcp/payment-safety-server-card.json",
   "payment_safety_mcp_url": "https://agent-guild-5d5r.onrender.com/mcp/payment-safety/",
   "repository": "https://github.com/AgentTanuki/agent-guild",
-  "version": "2.6.2"
+  "version": "2.6.3"
  }
 }

live/guild/tests/conftest.py

@@ -11,6 +11,10 @@
 
 os.environ.setdefault("GUILD_DATA", "")          # in-memory only
 os.environ.setdefault("GUILD_BOOTSTRAP_EVAL", "0")  # no auto-seed during tests
+# Production SQLite warms the complete signed discovery census at service
+# startup. Keep ordinary tests request-local; dedicated cache tests exercise
+# the production path explicitly.
+os.environ.setdefault("GUILD_DISCOVERY_REACH_CACHE", "0")
 # Abuse controls default ON in production; the suite hammers endpoints far
 # beyond real-world burst limits, so they are exercised explicitly in
 # tests/test_abuse_controls.py and disabled everywhere else.

live/guild/tests/test_discovery_reach.py

@@ -5,9 +5,10 @@
 import json
 
 from fastapi.testclient import TestClient
+import pytest
 
 from app import attribution, crypto, main
-from app.store import Store
+from app.store import DiscoveryReachSnapshotUnavailable, Store
 
 
 def test_reach_deduplicates_agents_and_excludes_vanity_traffic(tmp_path):
@@ -73,6 +74,74 @@ def test_reach_deduplicates_agents_and_excludes_vanity_traffic(tmp_path):
         "evidence"]["actor_evidence_set_sha256"]
 
 
+def test_warm_census_is_an_immutable_replayable_snapshot(tmp_path):
+    census = Store(path=str(tmp_path / "guild.json"))
+    census.record_event(
+        "http:first", "discovery_resource_fetched",
+        ua="langchain/0.2.1", discovery_surface="ard_catalog",
+        actor_distinct=True,
+    )
+    built = census.refresh_discovery_reach_cache()
+    assert built["refreshed"] is True
+
+    summary = census.discovery_reach()
+    snapshot_rows = summary["proof"]["payload"]["event_snapshot_rows"]
+    assert summary["snapshot_cache"]["event_snapshot_rows"] == snapshot_rows
+    assert summary["qualified_distinct_autonomous_agents"] == 1
+
+    # A later event cannot mutate an already signed snapshot. The next
+    # scheduled refresh publishes a new proof instead.
+    census.record_event(
+        "http:second", "query", ua="openai-agents/1.0",
+        endpoint="check", actor_distinct=True,
+    )
+    still_signed = census.discovery_reach()
+    assert still_signed["qualified_distinct_autonomous_agents"] == 1
+    assert still_signed["proof"] == summary["proof"]
+
+    replay = census.discovery_reach(
+        include_actor_evidence=True, snapshot_events=snapshot_rows)
+    assert len(replay["actor_evidence"]) == 1
+    replay["actor_evidence"].clear()  # returned values never mutate the cache
+    assert len(census.discovery_reach(
+        include_actor_evidence=True,
+        snapshot_events=snapshot_rows)["actor_evidence"]) == 1
+
+    rebuilt = census.refresh_discovery_reach_cache()
+    assert rebuilt["snapshot_events"] > snapshot_rows
+    assert census.discovery_reach()[
+        "qualified_distinct_autonomous_agents"] == 2
+
+    restarted = Store(path=str(tmp_path / "guild.json"))
+    restored = restarted.discovery_reach()
+    assert restored["snapshot_cache"]["event_snapshot_rows"] == rebuilt[
+        "snapshot_events"]
+    assert restored["qualified_distinct_autonomous_agents"] == 2
+
+
+def test_warm_census_never_rescans_for_an_unavailable_snapshot(
+        tmp_path, monkeypatch):
+    monkeypatch.setenv("GUILD_DISCOVERY_REACH_CACHE", "1")
+    census = Store(path=str(tmp_path / "guild.json"))
+    census.record_event(
+        "http:first", "discovery_resource_fetched",
+        ua="langchain/0.2.1", discovery_surface="ard_catalog",
+        actor_distinct=True,
+    )
+    built = census.refresh_discovery_reach_cache()
+
+    def forbidden_scan(*args, **kwargs):
+        raise AssertionError("request path attempted a durable-history scan")
+
+    monkeypatch.setattr(census, "measurement_event_snapshot", forbidden_scan)
+    with pytest.raises(DiscoveryReachSnapshotUnavailable) as caught:
+        census.discovery_reach(
+            include_actor_evidence=True,
+            snapshot_events=built["snapshot_events"] - 1,
+        )
+    assert caught.value.available_snapshot_events == built["snapshot_events"]
+
+
 def test_machine_resource_fetch_records_one_noncommercial_observation():
     client = TestClient(main.app)
     before = len(main.store.events)