From 22fc9b9427deb1a26026e720c075bbf5fc47feb2 Mon Sep 17 00:00:00 2001 From: AgentTanuki Date: Fri, 31 Jul 2026 13:30:29 +0100 Subject: [PATCH 1/2] Second corrections: A2A payment settled the wrong product, x402 was mistaken for money, experiments were promoted by unrelated revenue Five system-level defects found by independent review of 79265a9. Every one passed its own unit tests; each was only visible end to end. 1. A2A DEEP-PREFLIGHT PAYMENT WAS BROKEN END TO END. _preq_from_task always rebuilt payments.check_request(capability) and handle_payment_submission always returned store.check(...), so a deep-preflight challenge settled the WRONG canonical operation and returned a capability shortlist to someone who had paid for an endpoint check. The task now persists the exact operation and its canonical params at quote time; reconstruction is table-driven through payments.request_from_stored, reads ONLY our own stored fields (a submission cannot steer settlement onto another operation) and refuses an unknown operation instead of defaulting. Settlement dispatches to the product actually bought and records a deep_preflight_run with full settlement metadata and the actor quoted. Covered by a payment-required -> submitted -> completed test, not just challenge/card assertions. 2. settlement_mode == "x402" IS NOT PROOF OF MONEY. The same rail defaults to Base Sepolia, where a successful settlement is a successful payment of nothing, and a facilitator claiming success is not a chain receipt. HTTP/MCP/A2A now record confirmed, mainnet, network, amount and tx from auth.settled.record, and revenue requires ALL of: x402 + confirmed + mainnet + genuine-external attribution. Testnet and unconfirmed settlements are reported separately as supporting and can never promote. 3. EXPERIMENTS WERE NOT INDEPENDENTLY ATTRIBUTABLE. Any global primary metric moving could promote any experiment, so unrelated escrow revenue - or a watch sold for a different offer - would "prove" a deep_preflight price change worked. Each experiment is now bound to its operation; exposure, revenue, payers and repeat use are computed from that operation's events only, and revenue is summed from settled amounts rather than a global total. Also enforced ONE applied change per cycle GLOBALLY, not per experiment: two prices moving at once makes both results uninterpretable, which is the whole point of "one independently measurable change at a time". 4. THE A2A CARD STILL OPENED WITH PASSPORT ISSUANCE. It now leads with the endpoint decision and the measured numbers; the passport follows as supporting, with its register snippet intact so the offer funnel stage is not silently dropped. The existing copy-contract test was updated to the new ordering rather than deleted. 5. REMOTE INGEST WAS STILL DORMANT. A bounded local-only cycle is green in production, so the documented, capped MCP Registry adapter is now cleared and on by default, with GUILD_INDEX_INGEST=0 retained as a one-config-change kill switch. Sources we will NOT ingest are named with their exact gate rather than silently omitted: a2a_registry (the A2A discovery spec states it prescribes no registry API, so any endpoint we used would be undocumented - enabling it would mean scraping) and x402_bazaar (bulk automated reading is not covered by a documented public API contract, and a trade-listing surface is exactly where an unclear permission should be resolved before reading at machine speed). Published at /admin/index/cycle -> bounds and indexsources.active_sources. Also re-applies the index autorun default and POST /admin/index/cycle, which were lost when the previous correction branch was re-cut from main. Tests: 1113 passed, 9 skipped (15 new). --- docs/INTERFACE.md | 1 + live/guild/app/a2a.py | 33 +-- live/guild/app/a2a_x402.py | 69 +++++- live/guild/app/experiments.py | 173 +++++++++----- live/guild/app/indexsources.py | 62 ++++- live/guild/app/main.py | 52 +++-- live/guild/app/mcp_server.py | 23 +- live/guild/app/payments.py | 48 ++++ live/guild/app/swarm/runner.py | 35 ++- live/guild/contract/contract.json | 6 + live/guild/tests/test_index_corrections.py | 8 +- live/guild/tests/test_index_fixes2.py | 253 +++++++++++++++++++++ live/guild/tests/test_passport_offers.py | 9 +- live/guild/tests/test_trust_index.py | 20 +- render.yaml | 7 +- 15 files changed, 683 insertions(+), 116 deletions(-) create mode 100644 live/guild/tests/test_index_fixes2.py 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..c0e4bda 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,7 +936,7 @@ 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) @@ -953,7 +959,8 @@ 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]) 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..8eb4184 100644 --- a/live/guild/app/experiments.py +++ b/live/guild/app/experiments.py @@ -118,7 +118,8 @@ def define(store: Any, key: str, *, hypothesis: str, variable: str, return rec -def qualified_exposure(store: Any) -> dict[str, Any]: +def qualified_exposure(store: Any, operation: Optional[str] = None + ) -> dict[str, Any]: """Genuinely-external actors who reached a decision surface. Uses the SAME central attribution rule as every other honest number in the @@ -127,9 +128,16 @@ def qualified_exposure(store: Any) -> dict[str, Any]: which is exactly how self-traffic gets laundered into a growth metric.""" from . import attribution - decision_surfaces = {"preflight_run", "deep_preflight_run", - "evidence_bundle_issued", "watch_provisioned", - "index_view"} + # Scoped to the experiment's own surface where one is given: exposure to a + # DIFFERENT offer is not exposure to this one. + decision_surfaces = ({"preflight_run", "deep_preflight_run"} + if operation == "deep_preflight" else + {"evidence_bundle_issued"} + if operation == "evidence_bundle" else + {"watch_provisioned"} if operation == "watch_cycle" + else {"preflight_run", "deep_preflight_run", + "evidence_bundle_issued", "watch_provisioned", + "index_view"}) actors: set[str] = set() events = 0 for e in getattr(store, "events", []): @@ -157,71 +165,91 @@ def qualified_exposure(store: Any) -> dict[str, Any]: } -#: 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. +#: 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",), +} -def commercial_metrics(store: Any) -> dict[str, Any]: - """The primary metrics. Revenue is REAL money only. +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? - 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.""" + 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 commercial_metrics(store: Any, operation: Optional[str] = None + ) -> dict[str, Any]: + """The primary metrics. Revenue is REAL money only. + + `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 +258,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 +288,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 +339,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 +427,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 +449,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 +489,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 +500,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/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..6a6efa1 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 @@ -533,7 +534,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 +602,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 +3029,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 +3064,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 +3311,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..f676de9 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: @@ -422,7 +423,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 +520,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..a98ad1f 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,7 +219,7 @@ 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: 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..ca2d868 100644 --- a/live/guild/tests/test_index_corrections.py +++ b/live/guild/tests/test_index_corrections.py @@ -187,7 +187,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 +198,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..cb21bb9 --- /dev/null +++ b/live/guild/tests/test_index_fixes2.py @@ -0,0 +1,253 @@ +"""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 on the deep-preflight surface + store.record_event("a2a:net:looker", "deep_preflight_run", + ua="a2a:langchain/0.2.1", endpoint="preflight_deep") + # ...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): + store.record_event("a2a:net:a", "watch_provisioned", + ua="a2a:langchain/0.2.1", endpoint="watch") + 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", "deep_preflight_run", + ua="a2a:langchain/0.2.1", endpoint="preflight_deep") + store.record_event("a2a:net:y", "evidence_bundle_issued", + ua="a2a:langchain/0.2.2", endpoint="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_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 From 54950788da71b9572b1fa78adcba84e1d65a16cc Mon Sep 17 00:00:00 2001 From: AgentTanuki Date: Fri, 31 Jul 2026 13:41:00 +0100 Subject: [PATCH 2/2] Third corrections: exposure to a price means being SHOWN that price, and one DID is one subject Two more defects where a docstring promised something the code did not do, and the gap flattered us both times. 1. FREE USE WAS COUNTED AS EXPOSURE TO A PAID PRICE. qualified_exposure(store, "deep_preflight") counted free preflight_run events. A free-preflight caller has never been quoted the deep-preflight price, so the engine could reach its denominator on people who were never offered anything and then halve or kill that price. "They used the free thing" is not evidence about a price. There is now an explicit impression boundary. Every 402 / payment-required moment records a canonical `paid_offer_challenged` event carrying `challenged_operation`, recorded identically on HTTP, MCP and A2A through one shared helper per transport so the three cannot drift into measuring different things. Exposure to a paid operation is exactly two things: being shown that operation's price, or completing a PAID call of it (a free-tier call of the same shape is not an impression of the price). Adjacent free products never count. Evidence-bundle and watch experiments get the same boundary. The portfolio view (no operation) is retained for the commercial report and labelled NOT VALID FOR PRICING. 2. THE INDEX PROMISED IDENTITY DEDUPE AND PERFORMED ENDPOINT DEDUPE ONLY. The module documented "endpoint fingerprint FIRST, then declared identity" while ingest keyed solely on endpoint, so one subject at several addresses was several entries - inflating the one number the index is judged on, in a product whose entire claim is not overstating. Declared-DID coalescing now happens at ingest: the canonical entry keeps its endpoint, additional endpoints declaring the same did:key become `alias_endpoints` retaining their own provenance and last observation, and every alias source is also recorded on the canonical entry. Summaries report subjects and endpoints SEPARATELY (total_entries, alias_endpoints_folded, distinct_endpoints_known) so neither can be quoted as the other. reconcile_identities() migrates duplicates that already exist - oldest entry wins as canonical, deterministic and idempotent, run at the start of every cycle and a no-op when there is nothing to merge. OPERATOR equivalence is explicitly NOT inferred. Two endpoints sharing a name, a domain or a contact string are not evidence of one operator, and guessing would launder one party's evidence into another's - a worse error than the one being fixed. Operator stays `unknown` unless a deterministic declared identifier says otherwise, and the public view says so. Also corrects the stale indexops module comment: remote ingest is bounded default-ON for CLEARED_SOURCES only, with GUILD_INDEX_INGEST=0 retained as the one-config-change kill switch. Tests: 1130 passed, 9 skipped (17 new, including "free preflight alone cannot make deep pricing decisive" and "a genuine external deep challenge can"). --- live/guild/app/a2a.py | 8 + live/guild/app/experiments.py | 138 +++++++----- live/guild/app/indexops.py | 83 +++++++- live/guild/app/main.py | 25 +++ live/guild/app/mcp_server.py | 18 ++ live/guild/app/swarm/runner.py | 5 + live/guild/app/trustindex.py | 73 ++++++- live/guild/tests/test_index_corrections.py | 14 +- live/guild/tests/test_index_fixes2.py | 24 ++- live/guild/tests/test_index_fixes3.py | 233 +++++++++++++++++++++ 10 files changed, 550 insertions(+), 71 deletions(-) create mode 100644 live/guild/tests/test_index_fixes3.py diff --git a/live/guild/app/a2a.py b/live/guild/app/a2a.py index c0e4bda..cb3ad73 100644 --- a/live/guild/app/a2a.py +++ b/live/guild/app/a2a.py @@ -940,6 +940,10 @@ 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_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) @@ -964,6 +968,10 @@ 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_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/experiments.py b/live/guild/app/experiments.py index 8eb4184..5198f52 100644 --- a/live/guild/app/experiments.py +++ b/live/guild/app/experiments.py @@ -118,53 +118,6 @@ def define(store: Any, key: str, *, hypothesis: str, variable: str, return rec -def qualified_exposure(store: Any, operation: Optional[str] = None - ) -> dict[str, Any]: - """Genuinely-external actors who reached a decision surface. - - 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.""" - from . import attribution - - # Scoped to the experiment's own surface where one is given: exposure to a - # DIFFERENT offer is not exposure to this one. - decision_surfaces = ({"preflight_run", "deep_preflight_run"} - if operation == "deep_preflight" else - {"evidence_bundle_issued"} - if operation == "evidence_bundle" else - {"watch_provisioned"} if operation == "watch_cycle" - else {"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 - if e.get("fp") or e.get("first_party"): - continue - 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 { - "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."), - } - - #: 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 @@ -205,6 +158,97 @@ def _is_external(event: dict) -> bool: 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 + + def _external(e: dict) -> bool: + if e.get("fp") or e.get("first_party"): + return False + cls = attribution.caller_class(e) + if cls in ("AG_INTERNAL", "AG_TEST", "OPERATOR", "REGISTRY_CRAWLER"): + 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, + "paid_offers_shown": challenged, + "paid_completions": completed, + "rule": rule, + } + + def commercial_metrics(store: Any, operation: Optional[str] = None ) -> dict[str, Any]: """The primary metrics. Revenue is REAL money only. 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/main.py b/live/guild/app/main.py index 6a6efa1..6d9fa94 100644 --- a/live/guild/app/main.py +++ b/live/guild/app/main.py @@ -449,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.""" @@ -456,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 diff --git a/live/guild/app/mcp_server.py b/live/guild/app/mcp_server.py index f676de9..68fa782 100644 --- a/live/guild/app/mcp_server.py +++ b/live/guild/app/mcp_server.py @@ -306,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.""" @@ -397,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", @@ -406,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 diff --git a/live/guild/app/swarm/runner.py b/live/guild/app/swarm/runner.py index a98ad1f..bccab84 100644 --- a/live/guild/app/swarm/runner.py +++ b/live/guild/app/swarm/runner.py @@ -222,6 +222,11 @@ def _run_index_cycle(store: Any) -> dict[str, Any]: 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/tests/test_index_corrections.py b/live/guild/tests/test_index_corrections.py index ca2d868..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"] diff --git a/live/guild/tests/test_index_fixes2.py b/live/guild/tests/test_index_fixes2.py index cb21bb9..eafbb14 100644 --- a/live/guild/tests/test_index_fixes2.py +++ b/live/guild/tests/test_index_fixes2.py @@ -186,9 +186,10 @@ def test_unrelated_settlement_cannot_promote_a_deep_preflight_experiment(store): baseline={m: 0 for m in experiments.PRIMARY_METRICS}) exp["min_qualified"] = 1 store.experiments["deep"] = exp - # qualified exposure on the deep-preflight surface - store.record_event("a2a:net:looker", "deep_preflight_run", - ua="a2a:langchain/0.2.1", endpoint="preflight_deep") + # 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", @@ -210,8 +211,11 @@ def test_the_experiments_own_operation_does_promote(store): def test_exposure_is_scoped_to_the_experiments_surface(store): - store.record_event("a2a:net:a", "watch_provisioned", - ua="a2a:langchain/0.2.1", endpoint="watch") + """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 @@ -226,10 +230,12 @@ def test_only_one_change_is_applied_per_cycle_globally(store): baseline={m: 0 for m in experiments.PRIMARY_METRICS}) exp["min_qualified"] = 1 store.experiments[key] = exp - store.record_event("a2a:net:x", "deep_preflight_run", - ua="a2a:langchain/0.2.1", endpoint="preflight_deep") - store.record_event("a2a:net:y", "evidence_bundle_issued", - ua="a2a:langchain/0.2.2", endpoint="evidence_bundle") + 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 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"]