diff --git a/docs/INTERFACE.md b/docs/INTERFACE.md index daa2ca6..f5446a8 100644 --- a/docs/INTERFACE.md +++ b/docs/INTERFACE.md @@ -28,6 +28,7 @@ guild_mediated requires two-party cryptographic participation, a Guild-observed - `GET /.well-known/glama.json` - `POST /adjudicators/enroll` - `POST /admin/agents/{agent_id}/first-party` +- `POST /admin/index/cycle` - `POST /admin/issuer/rotate` - `GET /agents` - `GET /agents.md` diff --git a/live/guild/app/a2a.py b/live/guild/app/a2a.py index 5394701..cb3ad73 100644 --- a/live/guild/app/a2a.py +++ b/live/guild/app/a2a.py @@ -392,17 +392,23 @@ def _agent_card(base: str) -> dict[str, Any]: "protocolVersion": "0.3.0", "name": "Agent Guild", "description": ( - "The trust and settlement layer for AI agents. Claim a portable, " - "verifiable Agent Passport — free, three calls: POST " - '/agents/register {"name": "", "capabilities": [...], ' - '"src": "passport_offer:agent_card"} → POST /agents/{id}/prove ' - "(then /prove/verify) → GET /agents/{id}/passport, a Guild-signed " - "credential any party verifies offline. Also: send a text message " - "naming a capability (e.g. 'check: fact-check') and receive the " - "safest agent to hire, a hire/caution/avoid verdict, a ranked " - "shortlist, and provenance-labelled proof — the same answer as " - "GET /check. Richer surface (attest, escrow) over MCP and REST; " - "see documentationUrl." + "Can I safely use or pay this endpoint right now? Send " + "'preflight: ' and get back, free and live at request time, " + "what that endpoint CLAIMS and separately what it just PROVED: a " + "real protocol handshake (not merely HTTP 200), a resolvable agent " + "card, whether the card is signed, and whether an advertised " + "payment surface actually challenges with 402. Measured " + "2026-07-31: 92.9% of registry-listed agents report healthy and " + "33.9% complete a task. Send 'index' to search every endpoint we " + "have observed. 'deep-preflight: ' adds drift history, " + "cross-source corroboration and an explicit allow/caution/block " + "policy verdict (paid, x402). Also free, and supporting rather " + "than the headline: a portable Agent Passport for your own agent " + '(POST /agents/register {"name": "", "capabilities": [...], ' + '"src": "passport_offer:agent_card"} -> POST /agents/{id}/prove ' + "-> GET /agents/{id}/passport), and 'check: ' to find " + "the safest agent for a job. Richer surface over MCP and REST; see " + "documentationUrl." ), "url": f"{base}/a2a", "preferredTransport": "JSONRPC", @@ -930,10 +936,14 @@ async def a2a_endpoint(request: Request): # task TEXT every A2A client renders — not only in metadata, # which actor a2a:net:bba57b53… demonstrably never parsed. task = a2a_x402.build_payment_required_task( - preq, preq.cost, demand_ctx=dem) + preq, preq.cost, demand_ctx=dem, actor=actor, ua=ua_tag) store.record_event(actor, "x402_payment_required", ua=ua_tag, endpoint="best_agent", transport="a2a", capability=caller_cap) + store.record_event(actor, "paid_offer_challenged", ua=ua_tag, + endpoint="x402_challenge", transport="a2a", + challenged_operation=preq.operation, + 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) @@ -953,10 +963,15 @@ async def a2a_endpoint(request: Request): _target = _pfd.group(2) if _x402_a2a_active(): preq = payments.deep_preflight_request(_target) - task = a2a_x402.build_payment_required_task(preq, preq.cost) + task = a2a_x402.build_payment_required_task( + preq, preq.cost, actor=actor, ua=ua_tag) store.record_event(actor, "x402_payment_required", ua=ua_tag, endpoint="preflight_deep", transport="a2a", target=_target[:300]) + store.record_event(actor, "paid_offer_challenged", ua=ua_tag, + endpoint="x402_challenge", transport="a2a", + challenged_operation=preq.operation, + price_credits=preq.cost) resp = {"jsonrpc": "2.0", "id": id_, "result": task} return _with_extension_header(resp, request) payload = deepcheck.deep_preflight(store, _target) diff --git a/live/guild/app/a2a_x402.py b/live/guild/app/a2a_x402.py index c1f8757..bdb5f11 100644 --- a/live/guild/app/a2a_x402.py +++ b/live/guild/app/a2a_x402.py @@ -168,6 +168,8 @@ def _challenge_text(required: dict[str, Any], ctx: Optional[dict[str, Any]], def build_payment_required_task(preq: PaidRequest, credits_cost: int, demand_ctx: Optional[dict[str, Any]] = None, + actor: str = "", + ua: str = "", ) -> dict[str, Any]: """Create + persist a payment task and return the input-required Task.""" task_id = "x402task_" + uuid.uuid4().hex @@ -180,6 +182,17 @@ def build_payment_required_task(preq: PaidRequest, credits_cost: int, "request_hash": preq.request_hash, "credits_cost": credits_cost, "capability": dict(preq.query).get("capability"), + # The EXACT operation and its canonical parameters. A2A quotes in one + # message and settles in another, so the operation must survive the + # round trip; rebuilding it from a default made a deep-preflight + # challenge settle the wrong operation and return the wrong product. + # Stored here, on OUR record, and never read back from the submission. + "operation_params": dict(preq.query), + # Who was quoted. Recorded at quote time so the settled event is + # attributable to the same caller under the central attribution rule — + # an unattributable settlement can never be a customer. + "actor": actor or "", + "ua": ua or "", "required": required, "receipts": [], "created_at_epoch": time.time(), @@ -289,8 +302,7 @@ def handle_payment_submission(message: dict[str, Any], # Produce the paid result, bind receipt+evidence to its exact bytes. # demand for this request was recorded pre-authorization (B1) when the # payment-required task was created — never count it again on payment. - result = store.check(dict(preq.query).get("capability") or "", - demand_recorded=True) + result = _produce_for(preq, settled, task) body = json.dumps(result, default=str).encode("utf-8") fin = settled.finalize(body) settle_response = _settle_response({ @@ -344,9 +356,58 @@ def _settle_response(settle: dict[str, Any], return out +def _produce_for(preq: PaidRequest, settled: Any, + task: dict[str, Any]) -> dict[str, Any]: + """Produce the product that was actually paid for. + + Every branch records its own settlement metadata, because "the gateway + settled" and "money moved on mainnet" are different claims and only the + second one is revenue.""" + from . import deepcheck + facts = { + "settlement_mode": "x402", + "settlement_confirmed": bool((settled.record or {}).get("confirmed")), + "settlement_mainnet": bool((settled.record or {}).get("mainnet")), + "settlement_network": (settled.record or {}).get("network"), + "settlement_amount_atomic": (settled.record or {}).get("amount_atomic"), + "settlement_tx": (settled.record or {}).get("transaction"), + } + params = dict(preq.query) + 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, + verdict=(out.get("policy") or {}).get("decision"), + **facts) + return out + if preq.operation == "evidence_bundle": + url = params.get("url") or "" + out = deepcheck.evidence_bundle( + 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) + return out + return store.check(params.get("capability") or "", demand_recorded=True) + + def _preq_from_task(task: dict[str, Any]) -> PaidRequest: - cap = task.get("capability") or "" - return payments.check_request(cap) + """Reconstruct the quoted request from OUR OWN stored task record. + + Only `operation` and `operation_params` are consulted, both written by us + at quote time. Nothing from the payment submission reaches this function — + a caller must not be able to steer settlement onto a different operation + than the one they were quoted.""" + operation = task.get("operation") or "best_agent" + params = task.get("operation_params") + if not isinstance(params, dict): + # legacy tasks quoted before operation_params existed + params = {"capability": task.get("capability") or ""} + return payments.request_from_stored(operation, params) _ERR_CODES = { diff --git a/live/guild/app/experiments.py b/live/guild/app/experiments.py index de11041..5198f52 100644 --- a/live/guild/app/experiments.py +++ b/live/guild/app/experiments.py @@ -118,110 +118,182 @@ def define(store: Any, key: str, *, hypothesis: str, variable: str, return rec -def qualified_exposure(store: Any) -> dict[str, Any]: - """Genuinely-external actors who reached a decision surface. +#: The three INDEPENDENT conditions that must all hold before a settlement may +#: be called revenue. `mode == "x402"` alone is not money: the same rail runs +#: on Base Sepolia by default, where a successful settlement is a successful +#: payment of nothing, and a facilitator's word is not a chain receipt. +SETTLED_MODE = "x402" + +#: Which events belong to which paid operation. An experiment on the +#: deep_preflight price must be judged on deep_preflight revenue — not on +#: unrelated escrow settlement or a watch sold for a different offer. +OPERATION_EVENTS: dict[str, tuple[str, ...]] = { + "deep_preflight": ("deep_preflight_run",), + "evidence_bundle": ("evidence_bundle_issued",), + "watch_cycle": ("watch_provisioned",), +} + +ALL_PAID_EVENTS = tuple(t for v in OPERATION_EVENTS.values() for t in v) + + +def is_revenue(event: dict) -> bool: + """Did real, confirmed, mainnet money move for this event? - Uses the SAME central attribution rule as every other honest number in the - service. Crawlers, registry probes, our own tooling and unknown-attributed - traffic are excluded structurally — never by name-matching a User-Agent, - which is exactly how self-traffic gets laundered into a growth metric.""" + All three conditions, deliberately: mode (not sandbox credits we mint), + confirmed (the chain receipt was verified, not merely claimed by the + facilitator) and mainnet (not the value-less default network). Events + predating settlement metadata have none of these and are never revenue.""" + return (event.get("settlement_mode") == SETTLED_MODE + and bool(event.get("settlement_confirmed")) + and bool(event.get("settlement_mainnet"))) + + +def _is_external(event: dict) -> bool: + from . import attribution + cls = attribution.caller_class(event) + return (not event.get("fp") + and cls not in ("AG_INTERNAL", "AG_TEST", "OPERATOR", + "REGISTRY_CRAWLER") + and attribution.may_count_as_external_growth(cls) + and attribution.is_genuine_external(event)) + + +def qualified_exposure(store: Any, operation: Optional[str] = None + ) -> dict[str, Any]: + """Genuinely-external actors who were ACTUALLY OFFERED this paid operation. + + THE IMPRESSION BOUNDARY (correction 2026-07-31). This previously counted + adjacent free-product events — a caller who ran a FREE preflight was + treated as exposure for the PAID deep-preflight price experiment. They had + never been quoted that price, so the engine could reach its denominator and + halve or kill an offer nobody was shown. "They used the free thing" is not + evidence about a price. + + Exposure to a paid operation is now exactly two things, both explicit: + + * ``paid_offer_challenged`` carrying this operation — the caller was + shown the price (the 402 / payment-required moment, recorded + identically on HTTP, MCP and A2A), or + * a completed call of that operation — they saw the price and paid it. + + Nothing else counts. With no operation given (the commercial report), the + broader decision-surface view is returned, clearly labelled as such. + + Attribution is the same central rule used everywhere: crawlers, + first-party tooling and unknown-attributed traffic are excluded + structurally, never by matching a User-Agent.""" from . import attribution - decision_surfaces = {"preflight_run", "deep_preflight_run", - "evidence_bundle_issued", "watch_provisioned", - "index_view"} - actors: set[str] = set() - events = 0 - for e in getattr(store, "events", []): - if e.get("type") not in decision_surfaces: - continue + def _external(e: dict) -> bool: if e.get("fp") or e.get("first_party"): - continue + return False cls = attribution.caller_class(e) if cls in ("AG_INTERNAL", "AG_TEST", "OPERATOR", "REGISTRY_CRAWLER"): - continue - if not (attribution.may_count_as_external_growth(cls) - and attribution.is_genuine_external(e)): - continue - events += 1 - key = e.get("key") or "anon" - if key != "anon": - actors.add(key) + return False + return (attribution.may_count_as_external_growth(cls) + and attribution.is_genuine_external(e)) + + actors: set[str] = set() + events = 0 + challenged = 0 + completed = 0 + + if operation: + completion_types = set(OPERATION_EVENTS.get(operation, ())) + for e in getattr(store, "events", []): + etype = e.get("type") + is_challenge = (etype == "paid_offer_challenged" + and e.get("challenged_operation") == operation) + # A completion only counts as exposure when it was actually PAID + # for — a free-tier call of the same shape is not an impression of + # the price. + is_completion = (etype in completion_types + and e.get("settlement_mode") in + ("x402", "credits_sandbox")) + if not (is_challenge or is_completion): + continue + if not _external(e): + continue + events += 1 + challenged += 1 if is_challenge else 0 + completed += 1 if is_completion else 0 + key = e.get("key") or "anon" + if key != "anon": + actors.add(key) + rule = (f"actors genuinely external AND shown the {operation} price " + "(paid_offer_challenged) or who completed a paid " + f"{operation} call. Free-tier use of an adjacent product is " + "NOT exposure to this price.") + else: + surfaces = {"preflight_run", "deep_preflight_run", + "evidence_bundle_issued", "watch_provisioned", + "index_view", "paid_offer_challenged"} + for e in getattr(store, "events", []): + if e.get("type") not in surfaces or not _external(e): + continue + events += 1 + key = e.get("key") or "anon" + if key != "anon": + actors.add(key) + rule = ("ALL decision surfaces, free and paid — a portfolio view for " + "the commercial report. NOT valid for pricing an individual " + "operation; pass `operation` for that.") + return { + "operation_scope": operation or "all_surfaces", "qualified_actors": len(actors), "qualified_events": events, - "rule": ("genuine-external only, via attribution.caller_class + " - "is_genuine_external. Crawlers, first-party tooling and " - "unknown-attributed traffic are excluded structurally, not by " - "matching a User-Agent string."), + "paid_offers_shown": challenged, + "paid_completions": completed, + "rule": rule, } -#: The ONLY settlement mode that is money. `credits_sandbox` is an internal -#: unit we mint and hand out as trial credits; `free` is the soft launch. -SETTLED_MODE = "x402" - - -def commercial_metrics(store: Any) -> dict[str, Any]: +def commercial_metrics(store: Any, operation: Optional[str] = None + ) -> dict[str, Any]: """The primary metrics. Revenue is REAL money only. - SETTLEMENT MODE, NOT `paid=True` (correction 2026-07-31). The HTTP routes - previously stamped `paid=True` after the meter passed — but the meter - passes for three completely different reasons: an independently confirmed - x402 mainnet settlement, a draw against sandbox trial credits we minted - ourselves, and the soft-launch free path when enforcement is off. Counting - all three as paying customers meant our own trial grant could promote an - experiment. Only `settlement_mode == "x402"` from a genuinely external - caller counts here; sandbox decisions are reported separately, as - supporting, and can never promote. - - Events recorded before this correction carry no `settlement_mode`. They are - counted as SANDBOX, never as settled — the conservative direction, and the - one that cannot flatter us.""" - from . import attribution + `operation` scopes every figure to ONE paid operation. Without it, an + experiment on the deep_preflight price could be promoted by unrelated + escrow revenue or by a watch sold for a different offer — the experiment + would "work" for reasons that had nothing to do with the change it made. + Global (operation=None) figures remain available for the commercial report, + where a total is what is wanted.""" + want = (OPERATION_EVENTS.get(operation) if operation else ALL_PAID_EVENTS) payers: set[str] = set() paid_decisions = 0 repeat: dict[str, int] = {} + revenue_usd = 0.0 sandbox_decisions = 0 sandbox_actors: set[str] = set() unattributed_settled = 0 + testnet_settlements = 0 for e in getattr(store, "events", []): - if e.get("type") not in ("deep_preflight_run", "evidence_bundle_issued"): + if e.get("type") not in want: continue - mode = e.get("settlement_mode") or ("legacy_unlabelled" - if e.get("paid") else "free") key = e.get("key") or "" - if mode != SETTLED_MODE: - if mode in ("credits_sandbox", "legacy_unlabelled"): + if e.get("settlement_mode") != SETTLED_MODE: + if e.get("settlement_mode") == "credits_sandbox" or e.get("paid"): sandbox_decisions += 1 if key and key != "anon": sandbox_actors.add(key) continue - # A settled call still has to be EXTERNAL to be a customer. - cls = attribution.caller_class(e) - external = (not e.get("fp") - and cls not in ("AG_INTERNAL", "AG_TEST", "OPERATOR", - "REGISTRY_CRAWLER") - and attribution.may_count_as_external_growth(cls) - and attribution.is_genuine_external(e)) - if not external: + if not is_revenue(e): + # settled on the rail, but testnet and/or unconfirmed — a + # successful payment of nothing + testnet_settlements += 1 + continue + if not _is_external(e): unattributed_settled += 1 continue paid_decisions += 1 + revenue_usd += float(e.get("settlement_amount_atomic") or 0) / 1e6 if key and key != "anon": payers.add(key) repeat[key] = repeat.get(key, 0) + 1 - revenue_usd = 0.0 - try: - real = (store.escrow_summary() or {}).get("real_settlement") or {} - revenue_usd = float( - real.get("independently_attested_external_revenue_usd") or 0.0) - except Exception: # noqa: BLE001 - revenue_usd = 0.0 - monitored = 0 for w in getattr(store, "watches", {}).values(): if not w.get("active"): @@ -230,23 +302,29 @@ def commercial_metrics(store: Any) -> dict[str, Any]: if acct and acct.get("first_party"): continue # our own watch is not a customer monitored += 1 + if operation and operation != "watch_cycle": + monitored = 0 # not attributable to this experiment return { - "external_settled_revenue_usd": revenue_usd, + "operation_scope": operation or "all", + "external_settled_revenue_usd": round(revenue_usd, 6), "distinct_external_payers": len(payers), "paid_decisions": paid_decisions, "externally_monitored_endpoints": monitored, "repeat_paid_callers": sum(1 for n in repeat.values() if n > 1), "supporting_sandbox_decisions_NOT_REVENUE": sandbox_decisions, "supporting_sandbox_distinct_actors_NOT_PAYERS": len(sandbox_actors), + "supporting_testnet_or_unconfirmed_NOT_REVENUE": testnet_settlements, "settled_but_not_attributable_external": unattributed_settled, "settlement_rule": ( - "a paid decision requires settlement_mode == 'x402' (independently " - "confirmed mainnet money) AND a genuinely external caller. Sandbox " - "trial credits, soft-launch free calls and settled-but-unattributed " - "calls are reported separately and can never promote an experiment. " - "Events predating this correction carry no settlement_mode and are " - "counted as sandbox, never as settled."), + "revenue requires ALL of: settlement_mode == 'x402', " + "settlement_confirmed (chain receipt verified, not the " + "facilitator's word), settlement_mainnet (the rail defaults to " + "Base Sepolia, where a successful settlement is a successful " + "payment of nothing), AND a genuinely external caller. Sandbox " + "credits, testnet settlements, unconfirmed settlements and " + "unattributable callers are reported separately and can never " + "promote an experiment."), "revenue_definition": ( "independently confirmed EXTERNAL mainnet settlement only. " "Sandbox credits, first-party canaries, testnet funds and internal " @@ -254,14 +332,24 @@ def commercial_metrics(store: Any) -> dict[str, Any]: } +def experiment_operation(rec: dict) -> Optional[str]: + """The paid operation an experiment is bound to, from its variable.""" + variable = str((rec or {}).get("variable") or "") + if variable.startswith("price:"): + op = variable.split(":", 1)[1] + return op if op in OPERATION_EVENTS else None + return None + + def evaluate(store: Any, key: str) -> dict[str, Any]: """Decide an experiment — or refuse to, honestly.""" rec = store.experiments.get(key) if not rec: return {"key": key, "decision": None, "reason": "unknown experiment"} - exposure = qualified_exposure(store) - metrics = commercial_metrics(store) + operation = experiment_operation(rec) + exposure = qualified_exposure(store, operation) + metrics = commercial_metrics(store, operation) baseline = rec.get("baseline") or {} started = rec.get("started_at") try: @@ -295,7 +383,8 @@ def evaluate(store: Any, key: str) -> dict[str, Any]: "and no primary commercial metric moved. Supporting metrics " "(reach, inventory, free checks) cannot rescue this verdict.") - evidence = {"exposure": exposure, "metrics": metrics, "baseline": baseline, + evidence = {"operation": operation, + "exposure": exposure, "metrics": metrics, "baseline": baseline, "elapsed_days": round(elapsed.total_seconds() / 86400, 2), "window_expired": expired} with store.lock, store._txn(): @@ -382,9 +471,18 @@ def apply_next_action(store: Any) -> list[dict[str, Any]]: * a price already at zero is not "changed" again — the engine reports `offer_exhausted` rather than pretending a no-op was an action. - Returns one record per experiment acted on. Never raises: a failure to act - must not take the scheduled cycle down.""" + ONE CHANGE PER CYCLE, GLOBALLY. The mandate is "one independently + measurable change at a time", and that is a property of the SYSTEM, not of + each experiment: two prices moving in the same cycle makes both results + uninterpretable, because either change could explain whatever happens next. + Remaining experiments are evaluated and reported, and simply wait their + turn — the loop runs on a multi-hour schedule, so deferring costs one + cycle and buys an attributable result. + + Returns one record per experiment. Never raises: a failure to act must not + take the scheduled cycle down.""" applied: list[dict[str, Any]] = [] + acted_this_cycle = False for key in list(getattr(store, "experiments", {}) or {}): try: action = next_action(store, key) @@ -395,6 +493,13 @@ def apply_next_action(store: Any) -> list[dict[str, Any]]: applied.append({"key": key, "decision": action.get("decision"), "acted": False, "action": action.get("action")}) continue + if acted_this_cycle: + applied.append({"key": key, "decision": action.get("decision"), + "acted": False, "reason": "deferred_one_change_per_cycle", + "detail": ("another experiment already applied this " + "cycle; two simultaneous changes would " + "make both results unattributable")}) + continue change = action["change"] op, before, after = (change["operation"], change["from_credits"], change["to_credits"]) @@ -428,7 +533,8 @@ def apply_next_action(store: Any) -> list[dict[str, Any]]: }) live["changes_applied"] = live["changes_applied"][-20:] # restart the measurement window against a FRESH baseline - live["baseline"] = commercial_metrics(store) + live["baseline"] = commercial_metrics( + store, experiment_operation(live)) live["started_at"] = _now().isoformat() live["status"] = "running" live["decision"] = None @@ -438,6 +544,7 @@ def apply_next_action(store: Any) -> list[dict[str, Any]]: store._persist_kv("experiments", store.experiments) store._persist_kv("price_overrides", store.price_overrides) store._save() + acted_this_cycle = True applied.append({ "key": key, "acted": True, "operation": op, "before_credits": before, "after_credits": after, diff --git a/live/guild/app/indexops.py b/live/guild/app/indexops.py index 78e3075..78e6261 100644 --- a/live/guild/app/indexops.py +++ b/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: diff --git a/live/guild/app/indexsources.py b/live/guild/app/indexsources.py index 69c2e9e..01c9703 100644 --- a/live/guild/app/indexsources.py +++ b/live/guild/app/indexsources.py @@ -37,12 +37,64 @@ MAX_RECORDS_PER_RUN = 200 -def enabled() -> bool: - """Ingest is OFF unless explicitly enabled. +#: Remote sources cleared for automatic ingest. A source appears here ONLY +#: when three things are true: it publishes a documented public read-only API, +#: its terms permit automated reading, and we can identify ourselves honestly +#: while doing it. Anything requiring a credential, an account, or a scrape of +#: a rendered page is NOT here and does not ship — see UNAVAILABLE_SOURCES for +#: what is excluded and exactly why. +CLEARED_SOURCES = ("mcp_registry",) + +#: Sources deliberately NOT enabled, with the specific gate. Documented rather +#: than silently omitted, because "we index everything" and "we index what we +#: are allowed to index" are different products and the difference matters. +UNAVAILABLE_SOURCES = { + "a2a_registry": ( + "no documented public read-only API contract we can rely on; the A2A " + "discovery specification states it prescribes no registry API, so any " + "endpoint we used would be undocumented and could change without " + "notice. Enabling it would mean scraping, which the mandate forbids."), + "x402_bazaar": ( + "listing data is reachable, but automated bulk reading is not covered " + "by a documented public API contract. A trade-listing surface is " + "exactly where an unclear permission should be resolved BEFORE we " + "read it at machine speed, not after."), +} + - Default-off is deliberate: outbound traffic to third-party infrastructure - should never start because a container restarted.""" - return (os.environ.get("GUILD_INDEX_INGEST") or "0").strip() == "1" +def enabled() -> bool: + """Is remote registry ingest active? + + ON by default now that a bounded local-only cycle has run in production + without incident, and only for CLEARED_SOURCES. The explicit "0" kill + switch still wins over everything, so this stays a one-config-change stop + with no deploy — the property that matters when the traffic is outbound and + lands on someone else's servers.""" + raw = (os.environ.get("GUILD_INDEX_INGEST") or "").strip() + if raw == "0": + return False + if raw == "1": + return True + return True + + +def active_sources() -> dict: + """What we ingest, what we do not, and why — publishable as-is.""" + return { + "enabled": enabled(), + "cleared": list(CLEARED_SOURCES), + "excluded": UNAVAILABLE_SOURCES, + "kill_switch": "GUILD_INDEX_INGEST=0", + "user_agent": USER_AGENT, + "bounds": {"max_pages": MAX_PAGES, + "max_records_per_run": MAX_RECORDS_PER_RUN, + "timeout_s": TIMEOUT_S}, + "policy": ("documented public read-only APIs only, with a truthful " + "contactable User-Agent, capped pages and records per run. " + "No authentication bypass, no scraping, no Terms-of-Service " + "circumvention. Indexed inventory is a SUPPORTING metric " + "and is never reported as adoption."), + } def _get_json(url: str, timeout: float = TIMEOUT_S) -> Optional[Any]: diff --git a/live/guild/app/main.py b/live/guild/app/main.py index 98d28cf..6d9fa94 100644 --- a/live/guild/app/main.py +++ b/live/guild/app/main.py @@ -49,6 +49,7 @@ from . import trustindex from . import indexops from . import deepcheck +from . import indexsources from . import experiments from .state import store from .store import CanonicalWriteRefused @@ -448,6 +449,25 @@ 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_challenged", ua=ua, + endpoint="x402_challenge", transport=transport, + challenged_operation=operation, + price_credits=getattr(preq, "cost", None)) + except Exception: # noqa: BLE001 — telemetry must never break a 402 + pass + + def _challenge_http(exc: PaymentChallenge, status: int = 402) -> HTTPException: """One PaymentChallenge → one HTTP 402 with the PAYMENT-REQUIRED header.""" @@ -455,6 +475,12 @@ def _challenge_http(exc: PaymentChallenge, # 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") try: hdrs = {x402.PAYMENT_REQUIRED_HEADER: exc.header_value()} except Exception: # never mask the 402 @@ -533,7 +559,7 @@ def _meter_with_demand(preq: PaidRequest, x_api_key: Optional[str], def meter(preq: PaidRequest, x_api_key: Optional[str], - response: Response) -> str: + response: Response) -> dict: """Charge one priced request through the shared paid-operation gateway (app/payments.py — the SAME gateway MCP and A2A use). Behaviour: @@ -601,13 +627,15 @@ def meter(preq: PaidRequest, x_api_key: Optional[str], holder[0] = auth.settled elif auth.mode == "credits_sandbox" and auth.account is not None: response.headers["X-Guild-Balance"] = str(auth.account["balance"]) - # RETURN THE SETTLEMENT MODE. Callers must record HOW a request was paid, - # not merely that it passed the gate. "x402" is independently confirmed - # mainnet money; "credits_sandbox" is an internal unit we mint ourselves; - # "free" is the soft launch. Stamping paid=True for all three let sandbox - # trial credits and free calls count as paying customers — which is - # precisely the class of error the truth layer exists to prevent. - return auth.mode + # RETURN THE SETTLEMENT FACTS. Callers must record HOW a request was paid, + # not merely that it passed the gate. Three independent conditions have to + # hold before anything may be called revenue: mode == "x402" (not sandbox + # credits we mint, not the soft-launch free path), `confirmed` (the chain + # receipt was verified, not just the facilitator's word), and `mainnet` + # (the same rail defaults to Base Sepolia, where a successful settlement + # is a successful payment of nothing). Any one of these alone has been + # enough to overstate revenue. + return payments.settlement_facts(auth) _LANDING_HTML = """ @@ -3026,12 +3054,14 @@ 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.""" - mode = meter(payments.deep_preflight_request(url), x_api_key, response) + facts = meter(payments.deep_preflight_request(url), 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, paid=(mode == "x402"), settlement_mode=mode, - verdict=(out.get("policy") or {}).get("decision")) + target=url, + paid=(facts["settlement_mode"] == "x402"), + verdict=(out.get("policy") or {}).get("decision"), + **facts) return out @@ -3059,11 +3089,11 @@ 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"}) - mode = meter(preq, x_api_key, response) + 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, - paid=(mode == "x402"), settlement_mode=mode) + paid=(facts["settlement_mode"] == "x402"), **facts) return bundle @@ -3306,6 +3336,27 @@ def _attr(url: str) -> str: """) +@app.post("/admin/index/cycle") +def admin_index_cycle(x_admin_token: Optional[str] = Header(None)): + """Force ONE bounded index cycle now. Admin-gated. + + The autonomous loop runs on a jittered multi-hour schedule, which is right + for steady state and useless when you need to verify a deploy or refresh + after an incident. This runs exactly the same code path with exactly the + same bounds — a trigger, not a second implementation, so the manual and + scheduled paths cannot drift apart.""" + if ADMIN_TOKEN and x_admin_token != ADMIN_TOKEN: + raise HTTPException(403, "an index cycle requires a valid X-Admin-Token") + from .swarm import runner as _runner + return {"cycle": _runner._run_index_cycle(store), + "autorun_enabled": _runner.index_autorun(store), + "bounds": { + "recheck_batch": trustindex.recheck_batch(), + "remote_ingest_enabled": indexsources.enabled(), + "remote_sources": indexsources.active_sources(), + "fresh_ttl_s": trustindex.fresh_ttl_s()}} + + @app.get("/commercial") def commercial_report(): """The commercial scorecard, revenue first — the number that decides. diff --git a/live/guild/app/mcp_server.py b/live/guild/app/mcp_server.py index 83ea71d..68fa782 100644 --- a/live/guild/app/mcp_server.py +++ b/live/guild/app/mcp_server.py @@ -153,15 +153,16 @@ def _first_party_payer() -> "bool | None": _mcp_caller_proof: contextvars.ContextVar[tuple[bool, str]] = \ contextvars.ContextVar("mcp_caller_proof", default=(False, "")) -#: How the CURRENT paid tool call was settled: "x402" (independently confirmed -#: mainnet money), "credits_sandbox" (an internal unit we mint), or "free". -#: Only "x402" is revenue — see app/experiments.commercial_metrics. -_settlement_mode: contextvars.ContextVar[str] = \ - contextvars.ContextVar("settlement_mode", default="free") +#: The settlement FACTS for the current paid tool call — mode plus whether the +#: chain receipt was confirmed and whether it was mainnet. All three are needed +#: before anything may be called revenue (the rail defaults to Base Sepolia). +_settlement_mode: contextvars.ContextVar[dict] = \ + contextvars.ContextVar("settlement_facts", + default={"settlement_mode": "free"}) -def settlement_mode() -> str: - return _settlement_mode.get() +def settlement_mode() -> dict: + return dict(_settlement_mode.get() or {"settlement_mode": "free"}) def _meta_value(meta: Any, key: str) -> Any: @@ -305,6 +306,22 @@ def _mcp_payment(ctx: "Context | None") -> Optional[PaymentPayload]: return None +def _record_paid_offer(preq, ctx, api_key: str = "") -> None: + """MCP twin of the HTTP paid-offer impression — same event, same field.""" + try: + operation = getattr(preq, "operation", None) + if not operation: + return + store.record_event( + _creds.sanitize_actor_key(api_key) if api_key else "mcp", + "paid_offer_challenged", ua=_client_ua(ctx), + endpoint="x402_challenge", transport="mcp", + challenged_operation=operation, + price_credits=getattr(preq, "cost", None)) + except Exception: # noqa: BLE001 + pass + + def _challenge_result(body: dict[str, Any]) -> ToolResult: """A complete, machine-readable payment-required challenge as an MCP tool error — the unpaid caller never receives the paid payload.""" @@ -396,6 +413,7 @@ def _serve_paid(preq: PaidRequest, produce: Callable[[], Any], ns = demand.no_supply_block(dem) if dem else None if ns: body["no_supply"] = ns + _record_paid_offer(preq, ctx, api_key) return _challenge_result(body) except PaymentIdConflict as e: return _challenge_result({"error": "payment_identifier_conflict", @@ -405,6 +423,7 @@ def _serve_paid(preq: PaidRequest, produce: Callable[[], Any], ch = PaymentChallenge(preq, extra={"error": "x402_payment_invalid", "reason": e.reason, "detail": e.detail[:300]}) + _record_paid_offer(preq, ctx, api_key) return _challenge_result(ch.body) except CachedPaidResult as e: # official idempotency: same id + same request → cached result, no @@ -422,7 +441,7 @@ def _serve_paid(preq: PaidRequest, produce: Callable[[], Any], # contextvar, so an MCP tool can record HOW it was paid without every # producer signature growing an argument. Same correction as the HTTP # meter: passing the gate is not the same as being paid. - _settlement_mode.set(auth.mode) + _settlement_mode.set(payments.settlement_facts(auth)) result = produce() # in-band inbox delivery: the paid read is many agents' ONLY interaction # with the Guild, so an authenticated subject's pending messages ride on @@ -519,13 +538,13 @@ def guild_preflight_deep(url: str, api_key: str = "", ctx: Context = None) -> di """ def _produce(): out = deepcheck.deep_preflight(store, url) - mode = settlement_mode() + facts = settlement_mode() store.record_event( _creds.sanitize_actor_key(api_key) if api_key else "mcp", "deep_preflight_run", ua=_client_ua(ctx), endpoint="preflight_deep", target=url, transport="mcp", - paid=(mode == "x402"), settlement_mode=mode, - verdict=(out.get("policy") or {}).get("decision")) + paid=(facts.get("settlement_mode") == "x402"), + verdict=(out.get("policy") or {}).get("decision"), **facts) return out return _serve_paid(payments.deep_preflight_request(url), _produce, diff --git a/live/guild/app/payments.py b/live/guild/app/payments.py index 1d26c5d..c8a8890 100644 --- a/live/guild/app/payments.py +++ b/live/guild/app/payments.py @@ -125,6 +125,54 @@ def cost(self) -> int: # Builders for every priced semantic operation. MCP and A2A use these too, so # one semantic operation canonicalizes to one resource URL on every transport. +#: Rebuild a PaidRequest from a PERSISTED operation + params. Whitelist only: +#: an A2A payment task is reconstructed from our own stored fields, never from +#: anything the caller sends back, so a submission cannot redirect settlement +#: onto a different (cheaper, or simply other) operation. +def request_from_stored(operation: str, params: dict) -> "PaidRequest": + """Reconstruct the exact quoted operation from trusted stored fields. + + A2A quotes a price in one message and settles in another, so the operation + has to survive the round trip. Previously the rebuild ALWAYS produced + `check_request(capability)`, which meant a deep-preflight challenge settled + the wrong canonical operation and returned the wrong product entirely. + Reconstruction is now table-driven and refuses an unknown operation rather + than silently falling back to a default.""" + params = params or {} + if operation in ("best_agent", "signed_decision"): + return check_request(str(params.get("capability") or ""), + signed=(operation == "signed_decision"), + ttl_seconds=int(params.get("ttl_seconds") or 3600)) + if operation == "deep_preflight": + return deep_preflight_request(str(params.get("url") or "")) + if operation == "evidence_bundle": + return evidence_bundle_request( + str(params.get("url") or ""), + int(params.get("ttl_seconds") or 3600)) + if operation == "watch_cycle": + return watch_cycle_request(str(params.get("endpoint") or "")) + raise ValueError(f"unknown paid operation {operation!r}") + + +def settlement_facts(auth) -> dict: + """What actually happened, in the only terms that may be called revenue. + + `mode == "x402"` is NOT proof of money: the same rail runs on Base Sepolia + (testnet, value-less) by default, and a testnet settlement is a successful + payment of nothing. Callers record every field here and the revenue rule + requires x402 AND confirmed AND mainnet — three independent conditions, + because any one of them alone has been enough to overstate revenue before.""" + rec = getattr(getattr(auth, "settled", None), "record", None) or {} + return { + "settlement_mode": getattr(auth, "mode", "free"), + "settlement_confirmed": bool(rec.get("confirmed")), + "settlement_mainnet": bool(rec.get("mainnet")), + "settlement_network": rec.get("network"), + "settlement_amount_atomic": rec.get("amount_atomic"), + "settlement_tx": rec.get("transaction"), + } + + def check_request(capability: str, signed: bool = False, ttl_seconds: int = 3600) -> PaidRequest: operation = "signed_decision" if signed else "best_agent" diff --git a/live/guild/app/swarm/runner.py b/live/guild/app/swarm/runner.py index 925b642..bccab84 100644 --- a/live/guild/app/swarm/runner.py +++ b/live/guild/app/swarm/runner.py @@ -181,13 +181,32 @@ def _run_scout(store: Any, *, fetch: Callable, deadline: float, return scout.run_scout(store, fetch=fetch, deadline=deadline) -def index_autorun() -> bool: - """Index upkeep is DEFAULT-OFF, like every other outbound loop here. - - Outbound traffic to third-party infrastructure must never begin merely - because a container restarted, and a test that exercises the scout must not - silently start probing real hosts. Enabled explicitly in render.yaml.""" - return (os.environ.get("GUILD_INDEX_AUTORUN") or "0").strip() == "1" +def index_autorun(store: Any = None) -> bool: + """Should this cycle maintain the index? + + Three-way, and the asymmetry is deliberate: + + * ``GUILD_INDEX_AUTORUN=0`` — always OFF. The explicit kill switch wins + over everything, so an operator can stop index upkeep in one config + change without redeploying or touching the scout. + * ``GUILD_INDEX_AUTORUN=1`` — always ON. + * unset — ON only when a DURABLE backend is present. + + That last rule is the useful one. The index is a production surface whose + value is a persistent observation history; running it against an ephemeral + JSON store would produce observations that vanish on restart. It also keeps + the test suite (JSON store) from silently probing real hosts, while + production (sqlite) does not sit dormant waiting for someone to remember a + dashboard setting — which, for a system meant to run without routine human + involvement, would be a design flaw wearing a safety feature's clothes. + + Remote registry ingest remains a SEPARATE switch (``GUILD_INDEX_INGEST``).""" + raw = (os.environ.get("GUILD_INDEX_AUTORUN") or "").strip() + if raw == "0": + return False + if raw == "1": + return True + return getattr(store, "backend", None) is not None def _run_index_cycle(store: Any) -> dict[str, Any]: @@ -200,9 +219,14 @@ def _run_index_cycle(store: Any) -> dict[str, Any]: from .. import indexops from .. import experiments as _experiments - if not index_autorun(): + if not index_autorun(store): return {"skipped": "GUILD_INDEX_AUTORUN is not enabled"} out: dict[str, Any] = {} + try: + # fold any pre-existing same-DID duplicates before ingesting more + out["reconcile"] = indexops.reconcile_identities(store) + except Exception as exc: # noqa: BLE001 + out["reconcile_error"] = type(exc).__name__ try: out["ingest"] = indexops.ingest(store) except Exception as exc: # noqa: BLE001 diff --git a/live/guild/app/trustindex.py b/live/guild/app/trustindex.py index 8af6289..1bb1074 100644 --- a/live/guild/app/trustindex.py +++ b/live/guild/app/trustindex.py @@ -26,11 +26,22 @@ module exists, and it is why ``indexed`` is never promoted to ``live`` without an observation of our own. -DEDUPLICATION - By endpoint fingerprint (normalised scheme+host+port+path) FIRST, then by - declared identity (DID) where one exists. One operator publishing the same - service to three registries is one entry with three provenance records — not - three entries, and never three counts in a headline. +DEDUPLICATION — and its exact limits + Two levels, both DETERMINISTIC: + + 1. ENDPOINT — normalised scheme+host+port+path. One service listed by three + registries is one entry with three provenance records. + 2. DECLARED IDENTITY (DID) — when two endpoints declare the SAME did:key, + they are one subject with several addresses. The first becomes the + canonical entry; the others become `alias_endpoints` on it, keeping + their own provenance and their own last observation. + + OPERATOR identity is deliberately NOT inferred. Two endpoints with similar + names, a shared domain or the same contact string are NOT evidence of one + operator, and guessing would quietly merge unrelated parties — the opposite + of the error this index exists to prevent, and a worse one, because a merged + entry launders one party's evidence into another's. Operator remains + `unknown` unless a deterministic declared identifier says otherwise. PROVENANCE AND FRESHNESS Every entry records where it came from, when each source last confirmed it, @@ -143,6 +154,10 @@ def new_entry(url: str, source: str, *, declared: Optional[dict] = None, "observed_at": None, "observation_count": 0, "drift": [], # declared-vs-observed changes + # Other endpoints that declare the SAME did. Each keeps its own + # provenance and observation; they are addresses of one subject, not + # separate subjects, and are never counted separately in inventory. + "alias_endpoints": [], } @@ -157,6 +172,36 @@ def merge_source(entry: dict[str, Any], source: str) -> dict[str, Any]: return entry +def merge_alias(canonical: dict[str, Any], other: dict[str, Any]) -> dict[str, Any]: + """Fold `other` into `canonical` as an ALIAS ENDPOINT of the same subject. + + Nothing is discarded: the alias keeps its endpoint, its sources and its own + last observation, and every source of the alias is also recorded on the + canonical entry so provenance survives the merge. Idempotent — folding the + same alias twice updates it rather than adding a duplicate.""" + aliases = canonical.setdefault("alias_endpoints", []) + payload = { + "endpoint": other.get("endpoint"), + "id": other.get("id"), + "sources": other.get("sources", []), + "status": other.get("status"), + "observed_at": other.get("observed_at"), + "merged_at": _iso(), + "merged_on": "declared_did", + } + for i, existing in enumerate(aliases): + if existing.get("endpoint") == payload["endpoint"]: + aliases[i] = payload + break + else: + aliases.append(payload) + for src in other.get("sources", []): + merge_source(canonical, src.get("source", "unknown")) + if not canonical.get("declared") and other.get("declared"): + canonical["declared"] = other["declared"] + return canonical + + def is_stale(entry: dict[str, Any], ttl_s: Optional[int] = None) -> bool: """Has our own observation aged out? An entry never observed is stale.""" at = entry.get("observed_at") @@ -264,6 +309,15 @@ def public_view(entry: dict[str, Any], *, detail: bool = False) -> dict[str, Any "stale": is_stale(entry), "observation_count": entry.get("observation_count", 0), "first_indexed_at": entry.get("first_indexed_at"), + "alias_endpoints": [a.get("endpoint") + for a in entry.get("alias_endpoints", [])], + "identity": { + "did": entry.get("did") or None, + "dedupe": ("endpoint + declared DID" if entry.get("did") + else "endpoint only (no declared identity)"), + "operator": "unknown — operator identity is never inferred from " + "names, domains or contact strings", + }, } if entry.get("observation"): obs = entry["observation"] @@ -303,8 +357,11 @@ def summarise(entries: Iterable[dict[str, Any]]) -> dict[str, Any]: observed += 1 total = len(rows) listed_only = by_status.get(STATUS_INDEXED, 0) + aliased = sum(len(e.get("alias_endpoints") or []) for e in rows) return { "total_entries": total, + "alias_endpoints_folded": aliased, + "distinct_endpoints_known": total + aliased, "observed_by_guild": observed, "never_called_by_guild": listed_only, "by_status": by_status, @@ -314,4 +371,10 @@ def summarise(entries: Iterable[dict[str, Any]]) -> dict[str, Any]: f"{listed_only} are listings we have never called. A listing is a " "claim. Inventory size is a supporting metric and is never " "reported as adoption."), + "dedupe": ( + f"{total} subjects across {total + aliased} known endpoints " + f"({aliased} folded as aliases of the same declared DID). " + "Deduplication is endpoint-level and declared-DID-level only; " + "operator identity is NEVER inferred from names or domains, so two " + "entries may belong to one operator and we will not claim they do."), } diff --git a/live/guild/contract/contract.json b/live/guild/contract/contract.json index 398dcf6..e43009b 100644 --- a/live/guild/contract/contract.json +++ b/live/guild/contract/contract.json @@ -179,6 +179,12 @@ ], "path": "/admin/agents/{agent_id}/first-party" }, + { + "methods": [ + "POST" + ], + "path": "/admin/index/cycle" + }, { "methods": [ "POST" diff --git a/live/guild/tests/test_index_corrections.py b/live/guild/tests/test_index_corrections.py index 220b810..71a276b 100644 --- a/live/guild/tests/test_index_corrections.py +++ b/live/guild/tests/test_index_corrections.py @@ -50,9 +50,12 @@ def _decisive_kill(store: Store, key="exp-kill"): baseline={m: 0 for m in experiments.PRIMARY_METRICS}) exp["min_qualified"] = 1 store.experiments[key] = exp - # one genuinely external actor reached a decision surface and did not buy - store.record_event("a2a:net:realcaller", "deep_preflight_run", - ua="a2a:langchain/0.2.1", endpoint="preflight_deep") + # one genuinely external actor was SHOWN this operation's price and did + # not buy. Being shown the price is the only thing that counts as exposure + # to it — see tests/test_index_fixes3.py for why. + store.record_event("a2a:net:realcaller", "paid_offer_challenged", + ua="a2a:langchain/0.2.1", endpoint="x402_challenge", + challenged_operation="deep_preflight") return key @@ -109,8 +112,9 @@ def test_the_engine_can_only_move_a_price_downward(store): key = _decisive_kill(store) for _ in range(6): experiments.apply_next_action(store) - store.record_event("a2a:net:realcaller", "deep_preflight_run", - ua="a2a:langchain/0.2.1", endpoint="preflight_deep") + store.record_event("a2a:net:realcaller", "paid_offer_challenged", + ua="a2a:langchain/0.2.1", endpoint="x402_challenge", + challenged_operation="deep_preflight") assert pricing.price("deep_preflight") <= pricing.DEFAULTS["deep_preflight"] @@ -187,7 +191,9 @@ def test_legacy_unlabelled_events_count_as_sandbox_never_settled(store): def test_mainnet_settlement_from_an_external_caller_counts(store): store.record_event("a2a:net:payer1", "deep_preflight_run", ua="a2a:langchain/0.2.1", endpoint="preflight_deep", - paid=True, settlement_mode="x402") + paid=True, settlement_mode="x402", + settlement_confirmed=True, settlement_mainnet=True, + settlement_amount_atomic=20000) m = experiments.commercial_metrics(store) assert m["paid_decisions"] == 1 assert m["distinct_external_payers"] == 1 @@ -196,7 +202,9 @@ def test_mainnet_settlement_from_an_external_caller_counts(store): def test_settled_but_first_party_is_not_a_customer(store): store.record_event("ag-internal", "deep_preflight_run", ua="guild-release-gate", endpoint="preflight_deep", - paid=True, settlement_mode="x402", first_party=True) + paid=True, settlement_mode="x402", + settlement_confirmed=True, settlement_mainnet=True, + first_party=True) m = experiments.commercial_metrics(store) assert m["paid_decisions"] == 0 assert m["settled_but_not_attributable_external"] >= 1 diff --git a/live/guild/tests/test_index_fixes2.py b/live/guild/tests/test_index_fixes2.py new file mode 100644 index 0000000..eafbb14 --- /dev/null +++ b/live/guild/tests/test_index_fixes2.py @@ -0,0 +1,259 @@ +"""Second-round corrections — system-level defects the unit tests missed. + +Every defect here passed its own tests. They were found by asking what the +SYSTEM does end to end, which is a different question from what each function +returns: + + 1. an A2A deep-preflight challenge settled the wrong operation and returned + the wrong product — challenge and card tests both passed + 2. `settlement_mode == "x402"` was treated as money, but the same rail + defaults to Base Sepolia, where a successful settlement pays nothing + 3. any global metric moving could promote any experiment, so unrelated + escrow revenue would "prove" a deep-preflight price change worked + 4. the A2A agent card still opened with passport issuance + +The tests below are end-to-end or attribution-scoped for exactly that reason. +""" +from __future__ import annotations + +import json +import os +import sys + +import pytest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from app import experiments, payments, pricing # noqa: E402 +from app.store import Store # noqa: E402 + + +@pytest.fixture() +def store(tmp_path) -> Store: + pricing.load_runtime({}) + return Store(path=str(tmp_path / "guild.json")) + + +def _settled_event(store: Store, etype="deep_preflight_run", *, key="a2a:net:payer", + ua="a2a:langchain/0.2.1", mainnet=True, confirmed=True, + mode="x402", usd_atomic=20000, **extra): + store.record_event(key, etype, ua=ua, endpoint="preflight_deep", + paid=(mode == "x402"), settlement_mode=mode, + settlement_confirmed=confirmed, + settlement_mainnet=mainnet, + settlement_network="eip155:8453" if mainnet else "eip155:84532", + settlement_amount_atomic=usd_atomic, **extra) + + +# -------------------------------------------------------------------------- +# 1. A2A payment must settle the operation that was QUOTED +# -------------------------------------------------------------------------- +def test_a_deep_preflight_task_rebuilds_as_deep_preflight(store, monkeypatch): + """The defect: _preq_from_task always rebuilt check_request(capability), so + a deep-preflight challenge settled the wrong canonical operation.""" + from app import a2a_x402 + monkeypatch.setattr(a2a_x402, "store", store) + preq = payments.deep_preflight_request("https://x.example/a2a") + task = a2a_x402.build_payment_required_task(preq, preq.cost, + actor="a2a:net:abc", ua="a2a:ua") + stored = store.x402_task_get(task["id"]) + assert stored["operation"] == "deep_preflight" + assert stored["operation_params"]["url"] == "https://x.example/a2a" + rebuilt = a2a_x402._preq_from_task(stored) + assert rebuilt.operation == "deep_preflight" + assert rebuilt.resource_url == preq.resource_url + assert rebuilt.request_hash == preq.request_hash + + +def test_rebuild_uses_only_stored_fields_not_the_submission(store, monkeypatch): + """A caller must not be able to steer settlement onto another operation.""" + from app import a2a_x402 + monkeypatch.setattr(a2a_x402, "store", store) + preq = payments.deep_preflight_request("https://x.example/a2a") + task = a2a_x402.build_payment_required_task(preq, preq.cost) + stored = dict(store.x402_task_get(task["id"])) + # a hostile echo of a cheaper operation + stored_with_noise = {**stored, "capability": "fact-check"} + assert a2a_x402._preq_from_task(stored_with_noise).operation == "deep_preflight" + + +def test_unknown_operation_is_refused_not_defaulted(): + with pytest.raises(ValueError): + payments.request_from_stored("not_a_real_operation", {}) + + +def test_full_a2a_payment_round_trip_returns_the_DEEP_product(store, monkeypatch): + """payment-required -> payment-submitted -> completed, end to end. + + The challenge/card tests passed while this path returned a capability + shortlist to someone who paid for a deep endpoint check.""" + from app import a2a_x402, deepcheck, x402 + + monkeypatch.setattr(a2a_x402, "store", store) + monkeypatch.setattr(deepcheck, "deep_preflight", + lambda s, url: {"tier": "deep", "target": url, + "policy": {"decision": "allow"}}) + preq = payments.deep_preflight_request("https://x.example/a2a") + task = a2a_x402.build_payment_required_task( + preq, preq.cost, actor="a2a:net:payer", ua="a2a:langchain/0.2.1") + stored = store.x402_task_get(task["id"]) + + class _Settled: + record = {"confirmed": True, "mainnet": True, "network": "eip155:8453", + "amount_atomic": 20000, "transaction": "0x" + "a" * 64} + + def finalize(self, body): + return {"settle_response": {}, "extensions": {}} + + out = a2a_x402._produce_for(a2a_x402._preq_from_task(stored), + _Settled(), stored) + assert out["tier"] == "deep", "the caller must receive what they paid for" + assert out["target"] == "https://x.example/a2a" + ev = [e for e in store.events if e["type"] == "deep_preflight_run"] + assert len(ev) == 1, "the settled deep check must be recorded" + assert ev[0]["settlement_mode"] == "x402" + assert ev[0]["settlement_confirmed"] is True + assert ev[0]["settlement_mainnet"] is True + assert ev[0]["key"] == "a2a:net:payer", "settlement must be attributable" + + +def test_a_capability_task_still_returns_the_capability_product(store, monkeypatch): + from app import a2a_x402 + monkeypatch.setattr(a2a_x402, "store", store) + preq = payments.check_request("fact-check") + task = a2a_x402.build_payment_required_task(preq, preq.cost) + stored = store.x402_task_get(task["id"]) + assert a2a_x402._preq_from_task(stored).operation == "best_agent" + + +# -------------------------------------------------------------------------- +# 2. x402 is not the same thing as money +# -------------------------------------------------------------------------- +def test_testnet_settlement_can_never_count_as_revenue(store): + """Base Sepolia is the DEFAULT network. A successful settlement there is a + successful payment of nothing.""" + _settled_event(store, mainnet=False) + m = experiments.commercial_metrics(store, "deep_preflight") + assert m["paid_decisions"] == 0 + assert m["distinct_external_payers"] == 0 + assert m["external_settled_revenue_usd"] == 0.0 + assert m["supporting_testnet_or_unconfirmed_NOT_REVENUE"] == 1 + + +def test_unconfirmed_mainnet_settlement_does_not_count(store): + """A facilitator claiming success is not a chain receipt.""" + _settled_event(store, confirmed=False) + m = experiments.commercial_metrics(store, "deep_preflight") + assert m["paid_decisions"] == 0 + assert m["supporting_testnet_or_unconfirmed_NOT_REVENUE"] == 1 + + +def test_confirmed_mainnet_external_settlement_counts(store): + _settled_event(store) + m = experiments.commercial_metrics(store, "deep_preflight") + assert m["paid_decisions"] == 1 + assert m["distinct_external_payers"] == 1 + assert m["external_settled_revenue_usd"] == pytest.approx(0.02) + + +def test_is_revenue_requires_all_three_conditions(): + base = {"settlement_mode": "x402", "settlement_confirmed": True, + "settlement_mainnet": True} + assert experiments.is_revenue(base) is True + for drop in ("settlement_mode", "settlement_confirmed", "settlement_mainnet"): + bad = {**base, drop: (False if drop != "settlement_mode" else "credits_sandbox")} + assert experiments.is_revenue(bad) is False, drop + + +def test_testnet_volume_cannot_promote(store): + experiments.define(store, "e", hypothesis="h", variable="price:deep_preflight", + baseline={m: 0 for m in experiments.PRIMARY_METRICS}) + for i in range(30): + _settled_event(store, key=f"a2a:net:p{i}", ua=f"a2a:langchain/0.2.{i}", + mainnet=False) + assert experiments.evaluate(store, "e")["decision"] != "promote" + + +# -------------------------------------------------------------------------- +# 3. An experiment must be judged on ITS OWN operation +# -------------------------------------------------------------------------- +def test_unrelated_settlement_cannot_promote_a_deep_preflight_experiment(store): + """The defect: any global primary metric moving promoted any experiment, so + revenue from a completely different product would 'prove' a price change + worked.""" + exp = experiments.define(store, "deep", hypothesis="h", + variable="price:deep_preflight", + baseline={m: 0 for m in experiments.PRIMARY_METRICS}) + exp["min_qualified"] = 1 + store.experiments["deep"] = exp + # qualified exposure = actually shown the deep-preflight price + store.record_event("a2a:net:looker", "paid_offer_challenged", + ua="a2a:langchain/0.2.1", endpoint="x402_challenge", + challenged_operation="deep_preflight") + # ...and a pile of REAL money from an entirely different operation + for i in range(20): + _settled_event(store, etype="evidence_bundle_issued", + key=f"a2a:net:other{i}", ua=f"a2a:langchain/0.3.{i}") + out = experiments.evaluate(store, "deep") + assert out["decision"] != "promote", out + assert out["evidence"]["operation"] == "deep_preflight" + assert out["evidence"]["metrics"]["paid_decisions"] == 0 + + +def test_the_experiments_own_operation_does_promote(store): + exp = experiments.define(store, "deep", hypothesis="h", + variable="price:deep_preflight", + baseline={m: 0 for m in experiments.PRIMARY_METRICS}) + exp["min_qualified"] = 1 + store.experiments["deep"] = exp + _settled_event(store, key="a2a:net:realpayer") + assert experiments.evaluate(store, "deep")["decision"] == "promote" + + +def test_exposure_is_scoped_to_the_experiments_surface(store): + """Exposure is being SHOWN a specific operation's price (correction: + tests/test_index_fixes3.py) — so the impression must name the operation.""" + store.record_event("a2a:net:a", "paid_offer_challenged", + ua="a2a:langchain/0.2.1", endpoint="x402_challenge", + challenged_operation="watch_cycle") + deep = experiments.qualified_exposure(store, "deep_preflight") + watch = experiments.qualified_exposure(store, "watch_cycle") + assert deep["qualified_actors"] == 0 + assert watch["qualified_actors"] == 1 + + +def test_only_one_change_is_applied_per_cycle_globally(store): + """Two prices moving in one cycle makes both results uninterpretable.""" + for key, op in (("a", "deep_preflight"), ("b", "evidence_bundle")): + exp = experiments.define(store, key, hypothesis="h", + variable=f"price:{op}", + baseline={m: 0 for m in experiments.PRIMARY_METRICS}) + exp["min_qualified"] = 1 + store.experiments[key] = exp + store.record_event("a2a:net:x", "paid_offer_challenged", + ua="a2a:langchain/0.2.1", endpoint="x402_challenge", + challenged_operation="deep_preflight") + store.record_event("a2a:net:y", "paid_offer_challenged", + ua="a2a:langchain/0.2.2", endpoint="x402_challenge", + challenged_operation="evidence_bundle") + applied = experiments.apply_next_action(store) + acted = [r for r in applied if r.get("acted")] + assert len(acted) == 1, applied + deferred = [r for r in applied + if r.get("reason") == "deferred_one_change_per_cycle"] + assert len(deferred) == 1 + + +# -------------------------------------------------------------------------- +# 4. The A2A card leads with the decision +# -------------------------------------------------------------------------- +def test_agent_card_description_leads_with_the_endpoint_decision(): + from fastapi.testclient import TestClient + from app.main import app + card = TestClient(app).get("/.well-known/agent-card.json").json() + desc = card["description"] + assert desc.lower().startswith("can i safely use or pay this endpoint") + assert "preflight" in desc.lower() + # passports may still be mentioned — as supporting, not as the opener + head = desc[:200].lower() + assert "passport" not in head diff --git a/live/guild/tests/test_index_fixes3.py b/live/guild/tests/test_index_fixes3.py new file mode 100644 index 0000000..73b79ef --- /dev/null +++ b/live/guild/tests/test_index_fixes3.py @@ -0,0 +1,233 @@ +"""Third-round corrections: the impression boundary, and DID coalescing. + +Both defects were the same species — a claim in a docstring that the code did +not implement, where the gap flattered us: + + 1. exposure to a PAID price was being counted from FREE product use, so the + engine could halve or kill a price nobody had ever been shown; + 2. the module promised endpoint-then-identity deduplication and performed + endpoint deduplication only, so one subject at several addresses inflated + the one number the index is judged on. +""" +from __future__ import annotations + +import os +import sys + +import pytest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from app import experiments, indexops, pricing, trustindex # noqa: E402 +from app.store import Store # noqa: E402 + +EXT_UA = "a2a:langchain/0.2.1" + + +@pytest.fixture() +def store(tmp_path) -> Store: + pricing.load_runtime({}) + return Store(path=str(tmp_path / "guild.json")) + + +def _challenge(store: Store, operation: str, actor: str, ua: str = EXT_UA): + store.record_event(actor, "paid_offer_challenged", ua=ua, + endpoint="x402_challenge", + challenged_operation=operation, price_credits=20) + + +# -------------------------------------------------------------------------- +# 1. Exposure to a PAID price means having been shown that price +# -------------------------------------------------------------------------- +def test_free_preflight_is_not_exposure_to_the_deep_price(store): + """THE defect: a free preflight caller has never been quoted the deep + price, so counting them let the engine act on an offer nobody saw.""" + for i in range(40): + store.record_event(f"a2a:net:free{i}", "preflight_run", + ua=f"a2a:langchain/0.2.{i}", endpoint="preflight") + exp = experiments.qualified_exposure(store, "deep_preflight") + assert exp["qualified_actors"] == 0, exp + assert exp["paid_offers_shown"] == 0 + + +def test_free_preflight_alone_cannot_make_deep_pricing_decisive(store): + e = experiments.define(store, "deep", hypothesis="h", + variable="price:deep_preflight", + baseline={m: 0 for m in experiments.PRIMARY_METRICS}) + e["min_qualified"] = 3 + store.experiments["deep"] = e + for i in range(50): + store.record_event(f"a2a:net:free{i}", "preflight_run", + ua=f"a2a:langchain/0.2.{i}", endpoint="preflight") + out = experiments.evaluate(store, "deep") + assert out["decision"] in ("hold", "insufficient_evidence") + assert out["decision"] != "kill" + before = pricing.price("deep_preflight") + experiments.apply_next_action(store) + assert pricing.price("deep_preflight") == before, \ + "a price nobody was offered must not move" + + +def test_a_genuine_external_deep_challenge_IS_decisive(store): + """The other half: real exposure to the real price must be able to decide.""" + e = experiments.define(store, "deep", hypothesis="h", + variable="price:deep_preflight", + baseline={m: 0 for m in experiments.PRIMARY_METRICS}) + e["min_qualified"] = 2 + store.experiments["deep"] = e + _challenge(store, "deep_preflight", "a2a:net:shown1", "a2a:langchain/0.2.1") + _challenge(store, "deep_preflight", "a2a:net:shown2", "a2a:crewai/1.0") + exp = experiments.qualified_exposure(store, "deep_preflight") + assert exp["qualified_actors"] == 2 + assert exp["paid_offers_shown"] == 2 + assert experiments.evaluate(store, "deep")["decision"] == "kill" + + +def test_a_challenge_for_another_operation_is_not_exposure(store): + _challenge(store, "evidence_bundle", "a2a:net:other") + assert experiments.qualified_exposure( + store, "deep_preflight")["qualified_actors"] == 0 + assert experiments.qualified_exposure( + store, "evidence_bundle")["qualified_actors"] == 1 + + +def test_crawler_challenges_are_not_exposure(store): + _challenge(store, "deep_preflight", "a2a:net:bot", + ua="a2a:AgenstryBot/0.3.0") + assert experiments.qualified_exposure( + store, "deep_preflight")["qualified_actors"] == 0 + + +def test_a_paid_completion_also_counts_as_exposure(store): + """Someone who saw the price and PAID it has certainly been exposed.""" + store.record_event("a2a:net:payer", "deep_preflight_run", ua=EXT_UA, + endpoint="preflight_deep", settlement_mode="x402", + settlement_confirmed=True, settlement_mainnet=True) + exp = experiments.qualified_exposure(store, "deep_preflight") + assert exp["qualified_actors"] == 1 + assert exp["paid_completions"] == 1 + + +def test_a_free_tier_call_of_the_same_shape_is_not_a_completion(store): + store.record_event("a2a:net:freebie", "deep_preflight_run", ua=EXT_UA, + endpoint="preflight_deep", settlement_mode="free") + assert experiments.qualified_exposure( + store, "deep_preflight")["qualified_actors"] == 0 + + +def test_evidence_bundle_and_watch_have_the_same_boundary(store): + for op in ("evidence_bundle", "watch_cycle"): + assert experiments.qualified_exposure(store, op)["qualified_actors"] == 0 + _challenge(store, "watch_cycle", "a2a:net:w1") + assert experiments.qualified_exposure( + store, "watch_cycle")["qualified_actors"] == 1 + assert experiments.qualified_exposure( + store, "evidence_bundle")["qualified_actors"] == 0 + + +def test_portfolio_view_is_labelled_as_not_valid_for_pricing(store): + out = experiments.qualified_exposure(store) + assert out["operation_scope"] == "all_surfaces" + assert "NOT valid for pricing" in out["rule"] + + +# -------------------------------------------------------------------------- +# 2. DID coalescing — one subject, several addresses +# -------------------------------------------------------------------------- +DID_A = "did:key:z6MkExampleAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA" +DID_B = "did:key:z6MkExampleBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB" + + +def test_two_endpoints_with_one_did_are_one_subject(store): + out = indexops.ingest(store, [ + {"endpoint": "https://a.example/a2a", "source": "s1", "did": DID_A}, + {"endpoint": "https://b.example/a2a", "source": "s2", "did": DID_A}, + ]) + assert out["added"] == 1 + assert out["aliased_to_existing_did"] == 1 + assert len(store.trust_index) == 1 + entry = next(iter(store.trust_index.values())) + assert [a["endpoint"] for a in entry["alias_endpoints"]] == \ + ["https://b.example/a2a"] + + +def test_alias_retains_its_own_provenance(store): + indexops.ingest(store, [ + {"endpoint": "https://a.example/a2a", "source": "mcp_registry", "did": DID_A}, + {"endpoint": "https://b.example/a2a", "source": "guild_registration", + "did": DID_A}, + ]) + entry = next(iter(store.trust_index.values())) + alias = entry["alias_endpoints"][0] + assert [s["source"] for s in alias["sources"]] == ["guild_registration"] + # and the canonical entry records that the second source saw the subject + assert {s["source"] for s in entry["sources"]} == {"mcp_registry", + "guild_registration"} + + +def test_different_dids_are_never_merged(store): + indexops.ingest(store, [ + {"endpoint": "https://a.example/a2a", "source": "s", "did": DID_A}, + {"endpoint": "https://b.example/a2a", "source": "s", "did": DID_B}, + ]) + assert len(store.trust_index) == 2 + + +def test_operator_equivalence_is_never_inferred_from_names(store): + """Two endpoints of the SAME company, no declared identity. Guessing would + launder one party's evidence into another's.""" + indexops.ingest(store, [ + {"endpoint": "https://api.acme.example/a2a", "source": "s", + "declared": {"name": "Acme Agent"}}, + {"endpoint": "https://eu.acme.example/a2a", "source": "s", + "declared": {"name": "Acme Agent"}}, + ]) + assert len(store.trust_index) == 2 + view = trustindex.public_view(next(iter(store.trust_index.values()))) + assert "never inferred" in view["identity"]["operator"] + + +def test_reconciliation_migrates_pre_existing_duplicates(store): + """The index shipped keyed on endpoint alone, so duplicates already exist.""" + for host in ("a", "b", "c"): + e = trustindex.new_entry(f"https://{host}.example/a2a", "s", did=DID_A) + e["first_indexed_at"] = f"2026-07-{10 + ord(host) - 97}T00:00:00+00:00" + store.trust_index[e["id"]] = e + assert len(store.trust_index) == 3 + out = indexops.reconcile_identities(store) + assert out["merged_into_canonical"] == 2 + assert len(store.trust_index) == 1 + canonical = next(iter(store.trust_index.values())) + assert canonical["endpoint"] == "https://a.example/a2a", "oldest wins" + assert len(canonical["alias_endpoints"]) == 2 + + +def test_reconciliation_is_idempotent(store): + for host in ("a", "b"): + e = trustindex.new_entry(f"https://{host}.example/a2a", "s", did=DID_A) + store.trust_index[e["id"]] = e + indexops.reconcile_identities(store) + again = indexops.reconcile_identities(store) + assert again["merged_into_canonical"] == 0 + assert len(store.trust_index) == 1 + assert len(next(iter(store.trust_index.values()))["alias_endpoints"]) == 1 + + +def test_reconciliation_is_a_noop_without_declared_identity(store): + for host in ("a", "b"): + e = trustindex.new_entry(f"https://{host}.example/a2a", "s") + store.trust_index[e["id"]] = e + assert indexops.reconcile_identities(store)["merged_into_canonical"] == 0 + assert len(store.trust_index) == 2 + + +def test_summary_reports_subjects_and_endpoints_separately(store): + indexops.ingest(store, [ + {"endpoint": "https://a.example/a2a", "source": "s", "did": DID_A}, + {"endpoint": "https://b.example/a2a", "source": "s", "did": DID_A}, + ]) + summary = trustindex.summarise(store.trust_index.values()) + assert summary["total_entries"] == 1 + assert summary["alias_endpoints_folded"] == 1 + assert summary["distinct_endpoints_known"] == 2 + assert "operator identity is NEVER inferred" in summary["dedupe"] diff --git a/live/guild/tests/test_passport_offers.py b/live/guild/tests/test_passport_offers.py index f9625e2..89507e5 100644 --- a/live/guild/tests/test_passport_offers.py +++ b/live/guild/tests/test_passport_offers.py @@ -56,14 +56,19 @@ def test_manifest_leads_with_passport_and_records_offer(): assert len(offers) == 1 and offers[0]["offer"] == "passport" -def test_agent_card_leads_with_passport_and_records_offer(): +def test_agent_card_leads_with_the_endpoint_decision_and_still_offers_a_passport(): + """Copy contract UPDATED by the product-led pivot (2026-07-31): the card + now leads with allow/caution/block for an endpoint, and the passport is + supporting. The offer telemetry must keep working regardless — a change of + emphasis must not silently drop a funnel stage.""" n0 = len(store.events) r = client.get("/.well-known/agent-card.json", headers={"User-Agent": "card-bot/1.0"}) assert r.status_code == 200 desc = r.json()["description"] assert "passport_offer:agent_card" in desc - assert desc.index("Passport") < desc.index("check:") + assert desc.lower().startswith("can i safely use or pay this endpoint") + assert desc.index("preflight") < desc.index("Passport") offers = _offers_since(n0, "agent_card") assert len(offers) == 1 and offers[0]["offer"] == "passport" # a2a attribution conventions: derived a2a: actor key + tagged real UA diff --git a/live/guild/tests/test_trust_index.py b/live/guild/tests/test_trust_index.py index e2f5f77..5ceb974 100644 --- a/live/guild/tests/test_trust_index.py +++ b/live/guild/tests/test_trust_index.py @@ -435,10 +435,26 @@ def test_recheck_is_capped_per_cycle(store): assert len(calls) == 5 -def test_remote_ingest_is_off_by_default(monkeypatch): +def test_remote_ingest_has_a_working_kill_switch(monkeypatch): + """Remote ingest is now ON for CLEARED sources only (a bounded local cycle + ran clean in production first). The property that still matters is that it + can be stopped in one config change, with no deploy.""" from app import indexsources - monkeypatch.delenv("GUILD_INDEX_INGEST", raising=False) + monkeypatch.setenv("GUILD_INDEX_INGEST", "0") assert indexsources.enabled() is False + monkeypatch.delenv("GUILD_INDEX_INGEST", raising=False) + assert indexsources.enabled() is True + + +def test_only_cleared_sources_are_ingested(): + """A source ships only with a documented public read-only API and terms + that permit automated reading. Everything else is excluded WITH ITS + REASON, not silently omitted.""" + from app import indexsources + assert indexsources.CLEARED_SOURCES == ("mcp_registry",) + for name, why in indexsources.UNAVAILABLE_SOURCES.items(): + assert len(why) > 40, f"{name} must state its actual gate" + assert "a2a_registry" in indexsources.UNAVAILABLE_SOURCES def test_source_adapter_identifies_itself_truthfully(): diff --git a/render.yaml b/render.yaml index 7e3343f..86311c8 100644 --- a/render.yaml +++ b/render.yaml @@ -73,8 +73,13 @@ services: # infrastructure should never start because a container restarted. # Adapters read documented public read-only APIs with a truthful, # contactable User-Agent; no scraping, no auth bypass, no ToS evasion. + # Cleared for the documented, capped MCP Registry adapter only after a + # bounded local-only cycle ran clean in production. Set to "0" to stop + # all remote ingest in one config change, with no deploy. Excluded + # sources and their exact gates: GET /admin/index/cycle -> bounds, and + # app/indexsources.UNAVAILABLE_SOURCES. - key: GUILD_INDEX_INGEST - value: "0" + value: "1" # How long an observation stays "fresh" before the index reports it as # stale and the loop re-probes it. - key: GUILD_INDEX_FRESH_TTL_S