Fail closed on exact-treatment scoping and seed the baseline in the arm's own frame

live/guild/app/a2a.py

@@ -940,6 +940,12 @@ async def a2a_endpoint(request: Request):
             store.record_event(actor, "x402_payment_required", ua=ua_tag,
                                endpoint="best_agent", transport="a2a",
                                capability=caller_cap)
+            store.record_event(actor, "paid_offer_shown", ua=ua_tag,
+                               endpoint="x402_challenge", transport="a2a",
+                               challenged_operation=preq.operation,
+                               impression="challenge_402",
+                               actor_distinct=True,
+                               price_credits=preq.cost)
             resp = {"jsonrpc": "2.0", "id": id_, "result": task}
             return _with_extension_header(resp, request)
         payload = store.check(caller_cap, demand_recorded=True)
@@ -964,6 +970,12 @@ async def a2a_endpoint(request: Request):
             store.record_event(actor, "x402_payment_required", ua=ua_tag,
                                endpoint="preflight_deep", transport="a2a",
                                target=_target[:300])
+            store.record_event(actor, "paid_offer_shown", ua=ua_tag,
+                               endpoint="x402_challenge", transport="a2a",
+                               challenged_operation=preq.operation,
+                               impression="challenge_402",
+                               actor_distinct=True,
+                               price_credits=preq.cost)
             resp = {"jsonrpc": "2.0", "id": id_, "result": task}
             return _with_extension_header(resp, request)
         payload = deepcheck.deep_preflight(store, _target)

live/guild/app/a2a_x402.py

@@ -373,14 +373,18 @@ def _produce_for(preq: PaidRequest, settled: Any,
         "settlement_tx": (settled.record or {}).get("transaction"),
     }
     params = dict(preq.query)
+    # The credits QUOTED when this task was created — not today's price. A
+    # payment settled late must count as evidence about the price the payer
+    # was actually shown.
+    quoted = task.get("credits_cost")
     actor = task.get("actor") or "a2a"
     ua = task.get("ua") or "a2a/x402"
     if preq.operation == "deep_preflight":
         url = params.get("url") or ""
         out = deepcheck.deep_preflight(store, url)
         store.record_event(actor, "deep_preflight_run", ua=ua,
                            endpoint="preflight_deep", transport="a2a",
-                           target=url, paid=True,
+                           target=url, paid=True, price_credits=quoted,
                            verdict=(out.get("policy") or {}).get("decision"),
                            **facts)
         return out
@@ -390,7 +394,8 @@ def _produce_for(preq: PaidRequest, settled: Any,
             store, url, ttl_s=int(params.get("ttl_seconds") or 3600))
         store.record_event(actor, "evidence_bundle_issued", ua=ua,
                            endpoint="evidence_bundle", transport="a2a",
-                           target=url, paid=True, **facts)
+                           target=url, paid=True, price_credits=quoted,
+                           **facts)
         return out
     return store.check(params.get("capability") or "", demand_recorded=True)
 

live/guild/app/indexops.py

@@ -4,7 +4,13 @@
 service that reach out to third-party infrastructure on a schedule, so they are
 in one file where their bounds can be read in a single sitting:
 
-  * ingest is default-OFF for remote sources and capped per run;
+  * remote ingest is bounded default-ON, and ONLY for the cleared sources in
+    ``indexsources.CLEARED_SOURCES`` (currently the documented, public,
+    read-only MCP Registry API). ``GUILD_INDEX_INGEST=0`` remains the
+    one-config-change kill switch, with no deploy — the property that matters
+    when the traffic lands on someone else's servers. Sources we will not
+    ingest are named with their exact gate in
+    ``indexsources.UNAVAILABLE_SOURCES``, not silently omitted;
   * recheck probes at most ``recheck_batch()`` endpoints per cycle, oldest
     observation first, so the loop degrades to slow rather than to abusive;
   * a watch is charged per cycle ACTUALLY performed, so a dormant endpoint bills
@@ -44,7 +50,7 @@ def ingest(store: Any, records: Optional[list[dict[str, Any]]] = None
     so an ingest run that adds zero new endpoints is a perfectly good run."""
     if records is None:
         records = indexsources.collect(store)
-    added = updated = skipped = 0
+    added = updated = skipped = aliased = 0
     with store.lock, store._txn():
         for rec in records:
             url = (rec.get("endpoint") or "").strip()
@@ -54,6 +60,21 @@ def ingest(store: Any, records: Optional[list[dict[str, Any]]] = None
                 continue
             fp = trustindex.fingerprint(norm)
             entry = store.trust_index.get(fp)
+            did = (rec.get("did") or "").strip()
+            # DECLARED-DID COALESCING. Two endpoints declaring the same did:key
+            # are one subject with two addresses. Fold the newcomer in as an
+            # alias rather than creating a second entry, which would inflate
+            # inventory — the one number this index is judged on.
+            if entry is None and did:
+                canonical = _by_did(store, did)
+                if canonical is not None and canonical.get("endpoint") != norm:
+                    provisional = trustindex.new_entry(
+                        norm, rec.get("source", "unknown"),
+                        declared=rec.get("declared"), did=did)
+                    trustindex.merge_alias(canonical, provisional)
+                    store.trust_index[canonical["id"]] = canonical
+                    aliased += 1
+                    continue
             if entry is None:
                 entry = trustindex.new_entry(
                     norm, rec.get("source", "unknown"),
@@ -76,12 +97,64 @@ def ingest(store: Any, records: Optional[list[dict[str, Any]]] = None
             store._persist_kv("trust_index", store.trust_index)
         store._save()
     return {"added": added, "provenance_updated": updated,
+            "aliased_to_existing_did": aliased,
             "skipped_unusable": skipped,
             "total_entries": len(store.trust_index),
-            "note": ("added = endpoints not previously known. "
+            "note": ("added = SUBJECTS not previously known. "
                      "provenance_updated = already known, another source saw "
-                     "it — NOT a new endpoint. Inventory is a supporting "
-                     "metric and is never reported as adoption.")}
+                     "it — NOT a new endpoint. aliased_to_existing_did = a new "
+                     "endpoint folded into an existing subject because it "
+                     "declares the same DID; it is an address, not a subject. "
+                     "Inventory is a supporting metric and is never reported "
+                     "as adoption.")}
+
+
+def _by_did(store: Any, did: str) -> Optional[dict[str, Any]]:
+    """The canonical entry declaring `did`, if any. Deterministic lookup only —
+    no fuzzy matching, ever."""
+    if not did:
+        return None
+    for entry in store.trust_index.values():
+        if (entry.get("did") or "") == did:
+            return entry
+    return None
+
+
+def reconcile_identities(store: Any) -> dict[str, Any]:
+    """MIGRATION: fold pre-existing duplicates that share a declared DID.
+
+    The index shipped keyed on endpoint alone, so two endpoints of one subject
+    are already stored as two entries. This coalesces them once, deterministic-
+    ally and idempotently: the OLDEST entry (by first_indexed_at, then by id
+    for stability) wins as canonical, the rest become aliases with their
+    provenance and last observation intact. Safe to run on every cycle — with
+    nothing to merge it is a no-op."""
+    by_did: dict[str, list] = {}
+    for entry in store.trust_index.values():
+        did = (entry.get("did") or "").strip()
+        if did:
+            by_did.setdefault(did, []).append(entry)
+    merged = 0
+    with store.lock, store._txn():
+        for did, entries in by_did.items():
+            if len(entries) < 2:
+                continue
+            entries.sort(key=lambda e: (e.get("first_indexed_at") or "",
+                                        e.get("id") or ""))
+            canonical, rest = entries[0], entries[1:]
+            for other in rest:
+                trustindex.merge_alias(canonical, other)
+                store.trust_index.pop(other["id"], None)
+                merged += 1
+            store.trust_index[canonical["id"]] = canonical
+        if merged and store.backend is not None:
+            store._persist_kv("trust_index", store.trust_index)
+        if merged:
+            store._save()
+    return {"merged_into_canonical": merged,
+            "subjects": len(store.trust_index),
+            "rule": ("same DECLARED did:key only. Operator equivalence is "
+                     "never inferred from names, domains or contact strings.")}
 
 
 def _owner_class_for(store: Any, entry: dict[str, Any]) -> str:

live/guild/app/main.py

@@ -168,6 +168,10 @@ async def _lifespan(app: "FastAPI"):
 
 # Capture the caller's User-Agent per request so the activity feed can show who
 # is calling (a framework UA like "python-httpx" / "langchain" vs a browser).
+#: Stable, privacy-safe actor for THIS request (see the middleware). Used for
+#: paid-offer impressions so a challenge is attributable to a distinct caller.
+_req_actor: contextvars.ContextVar = contextvars.ContextVar(
+    "req_actor", default=None)
 _ua: contextvars.ContextVar[str] = contextvars.ContextVar("ua", default="")
 # x402: the payment headers travel via contextvars so meter() can settle paid
 # reads on the real rail without threading a parameter through every endpoint.
@@ -219,6 +223,18 @@ async def _capture_ua(request: Request, call_next):
         request.scope["path"] = "/mcp/"
         request.scope["raw_path"] = b"/mcp/"
     _ua.set(request.headers.get("user-agent", ""))
+    # Request-scoped ACTOR for paid-offer impressions. _challenge_http has no
+    # request handle, so it previously recorded actor=None — every genuine
+    # external HTTP challenge incremented the event count and never the ACTOR
+    # count, which is the number the experiment engine actually gates on. The
+    # same stable, privacy-safe identity used for demand dedupe is bound here
+    # (a purpose-scoped hash of the key, or of client+UA — never a raw IP,
+    # never a raw key).
+    try:
+        _req_actor.set(_http_demand_actor(request,
+                                          request.headers.get("x-api-key")))
+    except Exception:  # noqa: BLE001 — attribution must never break a request
+        _req_actor.set(None)
     _xpay_sig.set(request.headers.get("payment-signature", ""))
     _xpay_v1.set(request.headers.get("x-payment", ""))
     _fp_flag.set(_is_first_party(request.headers.get("x-guild-source"),
@@ -449,13 +465,61 @@ def _rate_limit_key_op(agent_id: str) -> None:
 # BaseHTTPMiddleware's downstream task; mutations of the shared holder do).
 
 
+def _record_paid_offer(preq, ua: str, transport: str,
+                       actor: Optional[str] = None) -> None:
+    """Record that a specific caller was quoted a specific paid operation.
+
+    The canonical impression for the experiment engine. Kept in one function so
+    HTTP, MCP and A2A cannot drift into recording different things — an
+    exposure metric assembled differently per transport is not a metric."""
+    try:
+        operation = getattr(preq, "operation", None)
+        if not operation:
+            return
+        store.record_event(actor, "paid_offer_shown", ua=ua,
+                           endpoint="x402_challenge", transport=transport,
+                           challenged_operation=operation,
+                           impression="challenge_402",
+                           price_credits=getattr(preq, "cost", None))
+    except Exception:  # noqa: BLE001 — telemetry must never break a 402
+        pass
+
+
+def _record_price_impression(actor: Optional[str], operation: str,
+                             price_credits: int, ua: str,
+                             transport: str) -> None:
+    """The caller was SHOWN a price without a 402 — e.g. the per-cycle price in
+    a successful watch-provisioning response.
+
+    Same canonical event and same boundary as a 402 quote, distinguished by
+    `impression` so the two are never conflated in analysis. Provisioning
+    remains free, and this is emphatically not a payment: it records that a
+    price was displayed to a specific caller, which is the thing an experiment
+    on that price needs to know."""
+    try:
+        store.record_event(actor, "paid_offer_shown", ua=ua,
+                           endpoint="price_shown", transport=transport,
+                           challenged_operation=operation,
+                           impression="price_displayed",
+                           price_credits=price_credits)
+    except Exception:  # noqa: BLE001
+        pass
+
+
 def _challenge_http(exc: PaymentChallenge,
                     status: int = 402) -> HTTPException:
     """One PaymentChallenge → one HTTP 402 with the PAYMENT-REQUIRED header."""
     # The 402 body leads with the free passport path (x402.payment_required_
     # body `claim_passport`); count the offer where it is actually served.
     store.record_event(None, "offer_served", ua=_ua.get(), offer="passport",
                        endpoint="x402_challenge")
+    # PAID-OFFER IMPRESSION. This is the only moment a caller is actually shown
+    # a price for a specific operation. An experiment on that operation's price
+    # may count THIS and nothing else — a free preflight caller has never been
+    # quoted the deep-preflight price, so counting them let the engine halve a
+    # price nobody was ever offered.
+    _record_paid_offer(getattr(exc, "preq", None), _ua.get(), "http",
+                       actor=_req_actor.get())
     try:
         hdrs = {x402.PAYMENT_REQUIRED_HEADER: exc.header_value()}
     except Exception:                                    # never mask the 402
@@ -3029,11 +3093,13 @@ def deep_preflight_route(request: Request, response: Response,
 
     The free tier (`GET /preflight`) is not degraded to make this attractive:
     it returns the full live check set and verdict, and always will."""
-    facts = meter(payments.deep_preflight_request(url), x_api_key, response)
+    _preq = payments.deep_preflight_request(url)
+    _quoted = _preq.cost
+    facts = meter(_preq, x_api_key, response)
     out = deepcheck.deep_preflight(store, url)
     store.record_event(creds.sanitize_actor_key(x_api_key) if x_api_key else None,
                        "deep_preflight_run", ua=_ua.get(), endpoint="preflight_deep",
-                       target=url,
+                       target=url, price_credits=_quoted,
                        paid=(facts["settlement_mode"] == "x402"),
                        verdict=(out.get("policy") or {}).get("decision"),
                        **facts)
@@ -3064,10 +3130,12 @@ def evidence_bundle_route(body: dict[str, Any], response: Response,
             "error": "evidence_issuance_refused", "code": e.code,
             "detail": str(e),
             "billing": "NOT CHARGED — issuance failed, so no meter ran"})
+    _quoted = preq.cost
     facts = meter(preq, x_api_key, response)
     store.record_event(creds.sanitize_actor_key(x_api_key) if x_api_key else None,
                        "evidence_bundle_issued", ua=_ua.get(),
                        endpoint="evidence_bundle", target=url,
+                       price_credits=_quoted,
                        paid=(facts["settlement_mode"] == "x402"), **facts)
     return bundle
 
@@ -3113,6 +3181,13 @@ def watch_provision_route(body: dict[str, Any], response: Response,
         raise HTTPException(422, str(e))
     store.record_event(creds.sanitize_actor_key(x_api_key), "watch_provisioned",
                        ua=_ua.get(), endpoint="watch", target=url)
+    # PRICE IMPRESSION. A watch has no 402 — provisioning is free — so the
+    # only moment the caller is shown the per-cycle price is this response.
+    # Recorded as an impression, explicitly NOT as a payment, so a watch
+    # pricing experiment can become decidable at all. Without it the boundary
+    # could never observe a watch price being offered to anyone.
+    _record_price_impression(creds.sanitize_actor_key(x_api_key), "watch_cycle",
+                             pricing.price("watch_cycle"), _ua.get(), "http")
     return {**rec, "price_per_cycle_credits": pricing.price("watch_cycle"),
             "feed": f"GET /watch/{rec['id']}",
             "billing": ("charged per recheck ACTUALLY performed; provisioning "