Exclude internal traffic from public demand summaries

live/guild/app/a2a.py

@@ -968,9 +968,11 @@ async def a2a_endpoint(request: Request):
     elif caller_kind == "objective_no_match":
         payload = _objective.unresolved_capsule(objective)
     elif caller_kind == "capabilities_map":
+        report = store.demand_summary_report()
         payload: dict[str, Any] = {
             "supplied": store.capability_index(),
-            "demand": store.demand_summary(),
+            "demand": report["summary"],
+            "demand_measurement": {k: v for k, v in report.items() if k != "summary"},
         }
     elif caller_kind == "coordination_policy":
         from . import coordination as _coord

live/guild/app/main.py

@@ -3710,6 +3710,12 @@ def demand_feed(request: Request,
                                   "endpoint (routable)",
             "transports": "where the demand arrived (http/mcp/a2a)",
             "first_seen/last_seen": "UTC timestamps",
+            "qualified_first_seen/qualified_last_seen": (
+                "UTC timestamps of qualifying asks only; owned or "
+                "unattributable activity cannot refresh this recency"),
+            "qualified_supplied_lookups": (
+                "qualifying asks recorded as finding supply; legacy asks "
+                "may lack that observation"),
         },
         "supplier_path": {
             "claim_passport": {
@@ -3788,12 +3794,13 @@ def demand_feed(request: Request,
 @app.get("/capabilities")
 def capabilities():
     """The supply/demand map, free. `supplied` lists every capability with
-    registered agents (and how many). `unmet_demand` lists capabilities agents
-    have actually asked /check about that currently have NO supply — real,
-    dated demand a new supplier can register against. Free because it recruits
-    supply."""
+    registered agents (and how many). `unmet_demand` lists qualified historical
+    asks for capabilities with no registered supplier. These are potential
+    leads, not funded jobs. /demand/feed additionally retains asks where
+    registered suppliers have no verified reachable endpoint."""
     supplied = store.capability_index()
-    demand = store.demand_summary()
+    report = store.demand_summary_report()
+    demand = report["summary"]
     unmet = {
         cap: row for cap, row in sorted(
             demand.items(), key=lambda kv: -kv[1]["lookups"])
@@ -3805,11 +3812,13 @@ def capabilities():
         "claim_passport": _passport_offer_block("capabilities"),
         "supplied": supplied,
         "unmet_demand": unmet,
+        "demand_measurement": {k: v for k, v in report.items() if k != "summary"},
         "demand_feed": "/demand/feed",
         "how_to_supply": (
             "POST /agents/register {\"name\": \"<you>\", \"capabilities\": "
-            "[\"<capability>\"]} — free. The first competent supplier of an "
-            "in-demand capability starts at rank 1. Signed, cacheable, "
+            "[\"<capability>\"]} — free. Evaluate the historical asks and "
+            "their qualification before choosing work to supply; registration "
+            "does not establish a buyer or a budget. Signed, cacheable, "
             "paginated unmet-demand feed for machines: GET /demand/feed."
         ),
     }

live/guild/app/store.py

@@ -4464,30 +4464,47 @@ def agent_by_did(self, did: str) -> Optional[dict[str, Any]]:
         return None
 
     def demand_summary(self) -> dict[str, dict[str, Any]]:
-        """Aggregate recorded capability demand: capability → lookup count,
-        how many found supply, and the latest lookup time. The supply-side
-        mirror of /check — lets an agent pick a capability where demand is
-        demonstrated but supply is missing."""
-        summary: dict[str, dict[str, Any]] = {}
-        for e in self.events:
-            if e.get("type") != "capability_demand":
-                continue
-            cap = e.get("capability", "")
-            if not cap:
-                continue
-            # Demand honesty (machine-economics audit R3): before 2026-07-06 the
-            # a2a first-token fallback recorded greetings ("hello", "ping") as
-            # demand. Only explicit asks (marked at record time) or asks that
-            # found supply count — advertised demand data must be priceable.
-            if not (e.get("explicit") or e.get("supplied")):
-                continue
-            row = summary.setdefault(
-                cap, {"lookups": 0, "supplied_lookups": 0, "last_lookup": None})
-            row["lookups"] += 1
-            if e.get("supplied"):
-                row["supplied_lookups"] += 1
-            row["last_lookup"] = e.get("at")
-        return summary
+        """Qualified historical asks, using the same rules as /demand/feed."""
+        return self.demand_summary_report()["summary"]
+
+    def demand_summary_report(self) -> dict[str, Any]:
+        """Supplier-facing counts and recency exclude owned/unqualified asks.
+
+        Read one bounded durable snapshot; never infer demand from just the
+        retained serving cache. The qualification is not proof of ownership,
+        budget or useful work. Raw events remain available for measurement.
+        """
+        events, coverage = self.measurement_event_view(
+            types=("capability_demand", "query"))
+        rows = self._derive_demand_rows(events)
+        summary = {
+            cap: {
+                "lookups": row["genuine_lookups"],
+                "supplied_lookups": row["qualified_supplied_lookups"],
+                "first_lookup": row["qualified_first_seen"],
+                "last_lookup": row["qualified_last_seen"],
+                "verified_lookups": row["verified_lookups"],
+                "heuristic_lookups": row["heuristic_lookups"],
+                "provenance": row["provenance"],
+            }
+            for cap, row in rows.items() if row["genuine_lookups"] > 0
+        }
+        return {
+            "summary": summary,
+            "measurement_version": "capability-demand-summary-v2",
+            "measurement_coverage": {k: coverage.get(k) for k in (
+                "source", "read_mode", "history_complete", "history_floor")},
+            "interpretation": (
+                "Qualified historical capability asks, not funded jobs or "
+                "demonstrated useful outcomes. Counts and recency exclude "
+                "Guild-operated, crawler and unattributable asks, and "
+                "deduplicate each actor/capability/hour. Caller proof verifies "
+                "identity, not external ownership; other qualification is "
+                "heuristic. Restored history is not new demand. "
+                "supplied_lookups counts qualified asks recorded as finding "
+                "supply; legacy asks may lack that observation. Compacted "
+                "JSON history remains incomplete."),
+        }
 
     def _derive_demand_rows(self, events=None) -> dict[str, dict[str, Any]]:
         """Read-time aggregation of demand, keyed by capability, from TWO
@@ -4531,11 +4548,13 @@ def _row(cap: str, at: Any) -> dict[str, Any]:
                 "demand_id": demand_mod.demand_id_for(cap),
                 "lookups": 0, "genuine_lookups": 0,
                 "verified_lookups": 0, "heuristic_lookups": 0,
+                "qualified_supplied_lookups": 0,
+                "qualified_first_seen": None, "qualified_last_seen": None,
                 "provenance": [],
                 "first_seen": at, "last_seen": at, "transports": []})
 
         def _count(cap, actor, at, transport, *, genuine, verified, heuristic,
-                   provenance):
+                   provenance, supplied=False):
             key = (actor or "anon", cap, _bucket(at))
             if key in seen:
                 return
@@ -4544,6 +4563,15 @@ def _count(cap, actor, at, transport, *, genuine, verified, heuristic,
             row["lookups"] += 1                # total, incl. non-genuine
             if genuine:
                 row["genuine_lookups"] += 1
+                if supplied:
+                    row["qualified_supplied_lookups"] += 1
+                if at:
+                    first = row["qualified_first_seen"]
+                    last = row["qualified_last_seen"]
+                    if first is None or str(at) < str(first):
+                        row["qualified_first_seen"] = at
+                    if last is None or str(at) > str(last):
+                        row["qualified_last_seen"] = at
                 if verified:
                     row["verified_lookups"] += 1
                 if heuristic:
@@ -4587,6 +4615,7 @@ def _count(cap, actor, at, transport, *, genuine, verified, heuristic,
             _count(cap, actor, e.get("at"), e.get("transport"),
                    genuine=genuine, verified=genuine and verified,
                    heuristic=genuine and not verified,
+                   supplied=bool(e.get("supplied")),
                    provenance=("verified_machine_demand" if verified
                                else "recorder_heuristic"))
 
@@ -4606,6 +4635,7 @@ def _count(cap, actor, at, transport, *, genuine, verified, heuristic,
                 continue
             _count(cap, e.get("actor") or e.get("key"), e.get("at"), "a2a",
                    genuine=True, verified=False, heuristic=True,
+                   supplied=bool(e.get("supplied")),
                    provenance="legacy_derived_heuristic")
         return rows
 

live/guild/app/swarm/agents.py

@@ -166,12 +166,9 @@ def _tick_gap_scout(store, asgi_client, http_client) -> dict:
     """Mandate: read unmet demand from our own surfaces and propose (only
     propose) seed-capability candidates."""
     demand = store.demand_summary() if hasattr(store, "demand_summary") else {}
-    asks = [e.get("capability") for e in store.events
-            if e.get("type") == "query" and e.get("caller_kind") == "capability_ask"
-            and e.get("capability")]
     from .capabilities import CAPABILITIES
     supplied = set(CAPABILITIES)
-    proposals = sorted({a for a in asks if a and a not in supplied})[:20]
+    proposals = sorted(cap for cap in demand if cap not in supplied)[:20]
     _log_action(store, agent="gap-scout", reason_code="demand_scan",
                 target="events+demand_watches", protocol="internal",
                 outcome=f"proposals:{len(proposals)}", policy="allowed",

live/guild/tests/test_capability_demand_summary.py

@@ -0,0 +1,131 @@
+"""A supplier must not mistake owned checks or lost history for real demand."""
+import json
+
+import pytest
+from fastapi.testclient import TestClient
+
+from app import a2a, main, store as store_module
+from app.store import Store
+from app.swarm import agents
+
+
+@pytest.fixture(params=["json", "sqlite"])
+def isolated_store(request, tmp_path, monkeypatch):
+    monkeypatch.setenv("GUILD_STORE", request.param)
+    monkeypatch.setenv("GUILD_STORE_PATH", str(tmp_path / "history.sqlite3"))
+    s = Store(path="")
+    monkeypatch.setattr(main, "store", s)
+    monkeypatch.setattr(a2a, "store", s)
+    return s
+
+
+def ask(s, actor, cap, *, at="2026-09-01T10:00:00+00:00", **fields):
+    s.record_event(actor, "capability_demand", ua=fields.pop("ua", "langchain/0.3"),
+                   capability=cap, actor=actor, at=at, explicit=True,
+                   transport="http", **fields)
+
+
+def maps():
+    client = TestClient(main.app)
+    http = client.get("/capabilities").json()
+    response = client.post("/a2a", json={
+        "jsonrpc": "2.0", "id": 1, "method": "message/send",
+        "params": {"message": {"parts": [{"kind": "text", "text": "capabilities"}]}},
+    }).json()
+    a2a = json.loads(response["result"]["parts"][0]["text"])
+    return http, a2a
+
+
+def test_owned_and_unattributable_asks_are_not_supplier_opportunities(isolated_store):
+    s = isolated_store
+    ask(s, "owned", "owned-only", demand_first_party=True,
+        caller_proof_verified=True)
+    ask(s, "fixture", "test-only", fp=True, fp_role="test")
+    ask(s, "crawler", "crawl-only", ua="Glama-Bot/2.0 (+crawler)")
+    ask(s, "tool", "tool-only", ua="curl/8.5")
+    ask(s, "buyer", "wanted")
+    retained = [dict(e) for e in s.events if e["type"] == "capability_demand"]
+    http, a2a = maps()
+    assert set(http["unmet_demand"]) == {"wanted"}
+    assert set(a2a["demand"]) == {"wanted"}
+    assert [e for e in s.events if e["type"] == "capability_demand"] == retained
+
+
+def test_owned_traffic_does_not_inflate_counts_or_refresh_buyer_recency(isolated_store):
+    s = isolated_store
+    ask(s, "buyer", "mixed", supplied=True)
+    ask(s, "buyer", "mixed", supplied=True)  # same actor and hour: one ask
+    ask(s, "owned", "mixed", at="2026-09-15T18:00:00+00:00",
+        demand_first_party=True, supplied=True)
+    http, a2a = maps()
+    for row in [http["unmet_demand"]["mixed"], a2a["demand"]["mixed"]]:
+        assert row["lookups"] == row["supplied_lookups"] == 1
+        assert row["last_lookup"] == "2026-09-01T10:00:00+00:00"
+        assert row["verified_lookups"] == 0
+        assert row["heuristic_lookups"] == 1
+    for result in [http, a2a]:
+        measurement = result["demand_measurement"]
+        assert measurement["measurement_version"] == "capability-demand-summary-v2"
+        assert "not funded jobs" in measurement["interpretation"]
+
+
+def test_verified_identity_and_legacy_asks_remain_distinct(isolated_store):
+    s = isolated_store
+    ask(s, "signed", "signed-work", ua="curl/8.5", caller_proof_verified=True)
+    s.record_event("old-buyer", "query", ua="a2a:langchain/0.3",
+                   actor="old-buyer", endpoint="a2a_message",
+                   caller_kind="capability_ask", capability="legacy-work",
+                   at="2026-08-01T10:00:00+00:00")
+    summary = s.demand_summary()
+    assert summary["signed-work"]["verified_lookups"] == 1
+    assert summary["signed-work"]["heuristic_lookups"] == 0
+    assert summary["legacy-work"]["heuristic_lookups"] == 1
+    assert summary["legacy-work"]["provenance"] == ["legacy_derived_heuristic"]
+
+
+def test_durable_demand_survives_retention_with_honest_json_coverage(
+        isolated_store, monkeypatch):
+    s = isolated_store
+    monkeypatch.setattr(store_module, "EVENT_RETENTION_TRIGGER", 5)
+    monkeypatch.setattr(store_module, "EVENT_RETENTION_TARGET", 3)
+    ask(s, "buyer", "retained-work")
+    for i in range(8):
+        s.record_event(None, "unrelated_activity", n=i)
+    assert s.events_omitted_by_retention > 0
+    if s.backend is not None:
+        monkeypatch.setattr(s.backend, "fetch_events",
+                            lambda **kw: pytest.fail("unbounded history fetch"))
+    result = TestClient(main.app).get("/capabilities").json()
+    coverage = result["demand_measurement"]["measurement_coverage"]
+    if s.backend is not None:
+        assert result["unmet_demand"]["retained-work"]["lookups"] == 1
+        assert coverage["history_complete"] is True
+        assert coverage["read_mode"] == "bounded_memory"
+    else:
+        assert coverage["history_complete"] is False
+        assert "retained-work" not in result["unmet_demand"]
+
+
+def test_gap_scout_uses_qualified_requests_not_raw_query_labels(isolated_store):
+    s = isolated_store
+    s.record_event("owned", "query", ua="guild-ops-check/1", fp=True,
+                   endpoint="a2a_message", caller_kind="capability_ask",
+                   capability="owned-proposal")
+    ask(s, "buyer", "actual-proposal")
+    result = agents._tick_gap_scout(s, None, None)
+    assert result["capability_proposals"] == ["actual-proposal"]
+
+
+def test_authenticated_empty_search_check_is_not_advertised_as_demand(
+        isolated_store, monkeypatch):
+    monkeypatch.setenv("GUILD_FIRST_PARTY_TOKEN", "fixture-owned-token")
+    client = TestClient(main.app)
+    response = client.get("/search?capability=owned-empty-search", headers={
+        "User-Agent": "langchain/0.3",
+        "X-Agent-Guild-First-Party": "fixture-owned-token",
+        "X-Agent-Guild-Role": "test",
+    })
+    assert response.status_code == 200 and response.json()["count"] == 0
+    assert any(e.get("capability") == "owned-empty-search"
+               for e in isolated_store.events)
+    assert "owned-empty-search" not in client.get("/capabilities").json()["unmet_demand"]

live/guild/tests/test_check_onboarding.py

@@ -126,18 +126,20 @@ def test_check_proven_supplier_has_no_guild_next():
     assert "guild_next" not in r
 
 
-def test_check_records_capability_demand_and_summary():
-    """Every /check is recorded as dated demand; the summary separates
-    supplied from unsupplied lookups so /capabilities can be honest."""
+def test_direct_checks_are_recorded_without_claiming_external_demand():
+    """Internal store calls remain in history but have no external caller."""
     s = _seeded_store()
     s.check("fact-check")
     s.check("web-research")
     s.check("web-research")
-    d = s.demand_summary()
-    assert d["fact-check"]["supplied_lookups"] == 1
-    assert d["web-research"]["lookups"] == 2
-    assert d["web-research"]["supplied_lookups"] == 0
-    assert d["web-research"]["last_lookup"] is not None
+    events = [e for e in s.events if e["type"] == "capability_demand"]
+    supplied = [e for e in events if e["capability"] == "fact-check"]
+    missing = [e for e in events if e["capability"] == "web-research"]
+    assert len(supplied) == 1 and supplied[0]["supplied"]
+    assert len(missing) == 2 and not any(e["supplied"] for e in missing)
+    assert all(e.get("at") for e in events)
+    assert "fact-check" not in s.demand_summary()
+    assert "web-research" not in s.demand_summary()
 
 
 def test_capability_index_counts_suppliers():

live/guild/tests/test_machine_economics.py

@@ -130,6 +130,7 @@ def test_probe_messages_get_probe_ack_and_pollute_no_demand():
         assert len(json.dumps(payload).encode()) < 1024
     demand = store.demand_summary()
     assert "hello" not in demand and "ping" not in demand and "你好" not in demand
-    # Explicit asks still count as demand, supplied or not.
-    client.get("/check?capability=underwater-basket-weaving")
+    # A qualified external ask still counts, supplied or not.
+    client.get("/check?capability=underwater-basket-weaving",
+               headers={"User-Agent": "langchain/0.3"})
     assert "underwater-basket-weaving" in store.demand_summary()