diff --git a/live/guild/app/a2a.py b/live/guild/app/a2a.py index c0e4bda..d444251 100644 --- a/live/guild/app/a2a.py +++ b/live/guild/app/a2a.py @@ -940,6 +940,12 @@ async def a2a_endpoint(request: Request): store.record_event(actor, "x402_payment_required", ua=ua_tag, endpoint="best_agent", transport="a2a", capability=caller_cap) + store.record_event(actor, "paid_offer_shown", ua=ua_tag, + endpoint="x402_challenge", transport="a2a", + challenged_operation=preq.operation, + impression="challenge_402", + actor_distinct=True, + price_credits=preq.cost) resp = {"jsonrpc": "2.0", "id": id_, "result": task} return _with_extension_header(resp, request) payload = store.check(caller_cap, demand_recorded=True) @@ -964,6 +970,12 @@ async def a2a_endpoint(request: Request): store.record_event(actor, "x402_payment_required", ua=ua_tag, endpoint="preflight_deep", transport="a2a", target=_target[:300]) + store.record_event(actor, "paid_offer_shown", ua=ua_tag, + endpoint="x402_challenge", transport="a2a", + challenged_operation=preq.operation, + impression="challenge_402", + actor_distinct=True, + price_credits=preq.cost) resp = {"jsonrpc": "2.0", "id": id_, "result": task} return _with_extension_header(resp, request) payload = deepcheck.deep_preflight(store, _target) diff --git a/live/guild/app/a2a_x402.py b/live/guild/app/a2a_x402.py index bdb5f11..95edd34 100644 --- a/live/guild/app/a2a_x402.py +++ b/live/guild/app/a2a_x402.py @@ -373,6 +373,10 @@ def _produce_for(preq: PaidRequest, settled: Any, "settlement_tx": (settled.record or {}).get("transaction"), } params = dict(preq.query) + # The credits QUOTED when this task was created — not today's price. A + # payment settled late must count as evidence about the price the payer + # was actually shown. + quoted = task.get("credits_cost") actor = task.get("actor") or "a2a" ua = task.get("ua") or "a2a/x402" if preq.operation == "deep_preflight": @@ -380,7 +384,7 @@ def _produce_for(preq: PaidRequest, settled: Any, out = deepcheck.deep_preflight(store, url) store.record_event(actor, "deep_preflight_run", ua=ua, endpoint="preflight_deep", transport="a2a", - target=url, paid=True, + target=url, paid=True, price_credits=quoted, verdict=(out.get("policy") or {}).get("decision"), **facts) return out @@ -390,7 +394,8 @@ def _produce_for(preq: PaidRequest, settled: Any, store, url, ttl_s=int(params.get("ttl_seconds") or 3600)) store.record_event(actor, "evidence_bundle_issued", ua=ua, endpoint="evidence_bundle", transport="a2a", - target=url, paid=True, **facts) + target=url, paid=True, price_credits=quoted, + **facts) return out return store.check(params.get("capability") or "", demand_recorded=True) diff --git a/live/guild/app/experiments.py b/live/guild/app/experiments.py index 8eb4184..bc28032 100644 --- a/live/guild/app/experiments.py +++ b/live/guild/app/experiments.py @@ -92,7 +92,8 @@ def _now() -> datetime: def define(store: Any, key: str, *, hypothesis: str, variable: str, - baseline: dict[str, Any]) -> dict[str, Any]: + baseline: dict[str, Any], + tested_price_credits: Optional[int] = None) -> dict[str, Any]: """Register a bounded, reversible experiment. Idempotent by key.""" with store.lock, store._txn(): existing = store.experiments.get(key) @@ -103,6 +104,9 @@ def define(store: Any, key: str, *, hypothesis: str, variable: str, "hypothesis": hypothesis, "variable": variable, "baseline": baseline, + # THE EXACT TREATMENT under test. Evidence gathered at a different + # price is evidence about that other price, not this one. + "tested_price_credits": tested_price_credits, "started_at": _now().isoformat(), "window_days": window_days(), "min_qualified": min_qualified(), @@ -118,53 +122,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,7 +162,144 @@ def _is_external(event: dict) -> bool: and attribution.is_genuine_external(event)) -def commercial_metrics(store: Any, operation: Optional[str] = None +def _at_or_after(event: dict, since: Optional[str]) -> bool: + """Is this event inside the treatment window? FAILS CLOSED. + + An event whose timestamp cannot be parsed is EXCLUDED, not admitted: an + unreadable date must never be able to decide a price. Cheap string + comparison first (ISO-8601 sorts lexically for a fixed offset), with a + real parse as the fallback.""" + if not since: + return True + at = event.get("at") + if not isinstance(at, str) or not at: + return False + try: + return (datetime.fromisoformat(at) + >= datetime.fromisoformat(str(since))) + except (TypeError, ValueError): + return False + + +def qualified_exposure(store: Any, operation: Optional[str] = None, + since: Optional[str] = None, + tested_price_credits: Optional[int] = None + ) -> dict[str, Any]: + """Genuinely-external actors who were ACTUALLY OFFERED this paid operation. + + TREATMENT WINDOW AND EXACT TREATMENT (correction 2026-07-31). `since` and + `tested_price_credits` scope exposure to the CURRENT experiment arm. + Without them the loop optimises on stale treatment data: ten callers see 20 + credits and do not buy, the engine cuts to 10, and on the next cycle those + same ten old-price impressions are counted again to justify cutting 10 to + 5 — even though nobody has been shown 10. An impression is evidence about + the price that was actually displayed, and about nothing else. + + 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_shown`` carrying this operation — the caller was shown + the price. Two impressions qualify and are distinguished by the + `impression` field: a 402 / payment-required quote, and a price + DISPLAYED in a successful response where there is no 402 at all (a + watch is provisioned free and priced per cycle, so its response is the + only moment its price is ever shown). 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 in ("paid_offer_shown", + "paid_offer_challenged") + and e.get("challenged_operation") == operation + # distinctness must be KNOWN to count an actor; an + # unidentifiable MCP caller is real traffic but + # cannot be counted toward an actor threshold + and e.get("actor_distinct") is not False) + # 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 + # Same arm only: inside the window AND at the price under test. + if not _at_or_after(e, since): + continue + if (tested_price_credits is not None + and e.get("price_credits") != tested_price_credits): + 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_shown", "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, + since: Optional[str] = None, + tested_price_credits: Optional[int] = None ) -> dict[str, Any]: """The primary metrics. Revenue is REAL money only. @@ -229,6 +323,20 @@ def commercial_metrics(store: Any, operation: Optional[str] = None for e in getattr(store, "events", []): if e.get("type") not in want: continue + # Same arm only. A payment QUOTED at the old price and settled late + # must not promote the new one — the money is real, but it is evidence + # about the price the payer was actually shown. + if not _at_or_after(e, since): + continue + # EXACT TREATMENT, FAIL CLOSED. A completion with NO recorded price + # cannot be shown to belong to this arm, so it is excluded rather than + # admitted. Excluding a real sale understates us; admitting an + # unattributable one would let any historical payment promote a price + # it was never quoted at, which is the failure being fixed. + if tested_price_credits is not None: + price = e.get("price_credits") + if price is None or price != tested_price_credits: + continue key = e.get("key") or "" if e.get("settlement_mode") != SETTLED_MODE: if e.get("settlement_mode") == "credits_sandbox" or e.get("paid"): @@ -304,8 +412,12 @@ def evaluate(store: Any, key: str) -> dict[str, Any]: return {"key": key, "decision": None, "reason": "unknown experiment"} operation = experiment_operation(rec) - exposure = qualified_exposure(store, operation) - metrics = commercial_metrics(store, operation) + since = rec.get("started_at") + tested = rec.get("tested_price_credits") + exposure = qualified_exposure(store, operation, since=since, + tested_price_credits=tested) + metrics = commercial_metrics(store, operation, since=since, + tested_price_credits=tested) baseline = rec.get("baseline") or {} started = rec.get("started_at") try: @@ -340,6 +452,11 @@ def evaluate(store: Any, key: str) -> dict[str, Any]: "(reach, inventory, free checks) cannot rescue this verdict.") evidence = {"operation": operation, + "arm": {"since": since, "tested_price_credits": tested, + "note": ("evidence is scoped to THIS arm: impressions " + "of a different price, and impressions from " + "before this window opened, cannot decide " + "it")}, "exposure": exposure, "metrics": metrics, "baseline": baseline, "elapsed_days": round(elapsed.total_seconds() / 86400, 2), "window_expired": expired} @@ -405,6 +522,79 @@ def next_action(store: Any, key: str) -> dict[str, Any]: "itself needs to change"} +#: The default experiment the service starts with. ONE, deliberately: the +#: engine applies at most one change per cycle, so seeding several would just +#: queue changes that cannot run concurrently anyway, and would make the first +#: result harder to attribute. +DEFAULT_EXPERIMENTS = ( + { + "key": "deep_preflight_price_v1", + "variable": "price:deep_preflight", + "hypothesis": ( + "Externally-owned agents shown the deep-preflight price do not buy " + "at 20 credits ($0.02). If qualified callers see the price and " + "still do not pay, the price is the blocker and halving it should " + "move paid decisions; if it does not, the OFFER is wrong, not the " + "price."), + }, +) + + +def seed_defaults(store: Any) -> dict[str, Any]: + """Ensure the engine has something to learn from. Idempotent. + + An autonomous experiment engine with no experiment is inert — it evaluates + an empty dict forever and reports nothing, which does not satisfy "find the + formula without a human". Seeding happens on the real cycle path so a fresh + deployment starts measuring by itself. + + Three safety properties, because a seeder that runs every cycle is one bug + away from continuously resetting the thing it is meant to measure: + + * NEVER overwrites an existing experiment — `define` returns the existing + record untouched, so the window and baseline survive every restart; + * NEVER seeds an experiment whose price is pinned by an operator, because + the engine could not act on it anyway and a permanently undecidable + experiment is noise; + * seeds ONE experiment, matching the one-change-per-cycle rule. + + It does NOT fabricate exposure. A seeded experiment with no qualified + callers correctly reports `insufficient_evidence` — that is the honest + state of a product nobody has been offered yet.""" + seeded, skipped = [], [] + for spec in DEFAULT_EXPERIMENTS: + key = spec["key"] + if key in (getattr(store, "experiments", {}) or {}): + skipped.append({"key": key, "reason": "already_exists"}) + continue + operation = spec["variable"].split(":", 1)[1] + if os.environ.get(pricing._env_key(operation)) is not None: + skipped.append({"key": key, "reason": "price_pinned_by_operator"}) + continue + # BASELINE IN THE SAME FRAME AS THE COMPARISON. evaluate() measures + # this arm (since started_at, at the tested price), so an all-time + # baseline would compare different frames: historical revenue would + # sit in the baseline while a genuine new-arm sale sat in the metrics, + # and the sale would read as "no movement". The arm opens at zero, + # which is the truth about a window that has just started. + arm_price = pricing.price(operation) + arm_start = _now().isoformat() + rec = define(store, key, hypothesis=spec["hypothesis"], + variable=spec["variable"], + baseline=commercial_metrics( + store, operation, since=arm_start, + tested_price_credits=arm_price), + tested_price_credits=arm_price) + # keep the window and the baseline frame identical + if rec.get("started_at") and rec["started_at"] < arm_start: + rec["started_at"] = arm_start + store.experiments[key] = rec + seeded.append(key) + return {"seeded": seeded, "already_present": skipped, + "note": ("idempotent: an existing experiment is never overwritten, " + "so a restart cannot reset its window or baseline")} + + def apply_next_action(store: Any) -> list[dict[str, Any]]: """Decide AND ACT on every running experiment. The autonomous half. @@ -488,10 +678,15 @@ def apply_next_action(store: Any) -> list[dict[str, Any]]: "reversible_via": pricing._env_key(op), }) live["changes_applied"] = live["changes_applied"][-20:] - # restart the measurement window against a FRESH baseline - live["baseline"] = commercial_metrics( - store, experiment_operation(live)) + # Restart the window, re-take the baseline, and record the NEW + # treatment. All three move together: an arm is (window, price), + # and updating one without the others is how a loop ends up + # optimising on stale treatment data. live["started_at"] = _now().isoformat() + live["tested_price_credits"] = after + live["baseline"] = commercial_metrics( + store, experiment_operation(live), + since=live["started_at"], tested_price_credits=after) live["status"] = "running" live["decision"] = None live["decided_at"] = None 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..dbcd449 100644 --- a/live/guild/app/main.py +++ b/live/guild/app/main.py @@ -168,6 +168,10 @@ async def _lifespan(app: "FastAPI"): # Capture the caller's User-Agent per request so the activity feed can show who # is calling (a framework UA like "python-httpx" / "langchain" vs a browser). +#: Stable, privacy-safe actor for THIS request (see the middleware). Used for +#: paid-offer impressions so a challenge is attributable to a distinct caller. +_req_actor: contextvars.ContextVar = contextvars.ContextVar( + "req_actor", default=None) _ua: contextvars.ContextVar[str] = contextvars.ContextVar("ua", default="") # x402: the payment headers travel via contextvars so meter() can settle paid # reads on the real rail without threading a parameter through every endpoint. @@ -219,6 +223,18 @@ async def _capture_ua(request: Request, call_next): request.scope["path"] = "/mcp/" request.scope["raw_path"] = b"/mcp/" _ua.set(request.headers.get("user-agent", "")) + # Request-scoped ACTOR for paid-offer impressions. _challenge_http has no + # request handle, so it previously recorded actor=None — every genuine + # external HTTP challenge incremented the event count and never the ACTOR + # count, which is the number the experiment engine actually gates on. The + # same stable, privacy-safe identity used for demand dedupe is bound here + # (a purpose-scoped hash of the key, or of client+UA — never a raw IP, + # never a raw key). + try: + _req_actor.set(_http_demand_actor(request, + request.headers.get("x-api-key"))) + except Exception: # noqa: BLE001 — attribution must never break a request + _req_actor.set(None) _xpay_sig.set(request.headers.get("payment-signature", "")) _xpay_v1.set(request.headers.get("x-payment", "")) _fp_flag.set(_is_first_party(request.headers.get("x-guild-source"), @@ -449,6 +465,47 @@ def _rate_limit_key_op(agent_id: str) -> None: # BaseHTTPMiddleware's downstream task; mutations of the shared holder do). +def _record_paid_offer(preq, ua: str, transport: str, + actor: Optional[str] = None) -> None: + """Record that a specific caller was quoted a specific paid operation. + + The canonical impression for the experiment engine. Kept in one function so + HTTP, MCP and A2A cannot drift into recording different things — an + exposure metric assembled differently per transport is not a metric.""" + try: + operation = getattr(preq, "operation", None) + if not operation: + return + store.record_event(actor, "paid_offer_shown", ua=ua, + endpoint="x402_challenge", transport=transport, + challenged_operation=operation, + impression="challenge_402", + price_credits=getattr(preq, "cost", None)) + except Exception: # noqa: BLE001 — telemetry must never break a 402 + pass + + +def _record_price_impression(actor: Optional[str], operation: str, + price_credits: int, ua: str, + transport: str) -> None: + """The caller was SHOWN a price without a 402 — e.g. the per-cycle price in + a successful watch-provisioning response. + + Same canonical event and same boundary as a 402 quote, distinguished by + `impression` so the two are never conflated in analysis. Provisioning + remains free, and this is emphatically not a payment: it records that a + price was displayed to a specific caller, which is the thing an experiment + on that price needs to know.""" + try: + store.record_event(actor, "paid_offer_shown", ua=ua, + endpoint="price_shown", transport=transport, + challenged_operation=operation, + impression="price_displayed", + price_credits=price_credits) + except Exception: # noqa: BLE001 + pass + + def _challenge_http(exc: PaymentChallenge, status: int = 402) -> HTTPException: """One PaymentChallenge → one HTTP 402 with the PAYMENT-REQUIRED header.""" @@ -456,6 +513,13 @@ 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", + actor=_req_actor.get()) try: hdrs = {x402.PAYMENT_REQUIRED_HEADER: exc.header_value()} except Exception: # never mask the 402 @@ -3029,11 +3093,13 @@ def deep_preflight_route(request: Request, response: Response, The free tier (`GET /preflight`) is not degraded to make this attractive: it returns the full live check set and verdict, and always will.""" - facts = meter(payments.deep_preflight_request(url), x_api_key, response) + _preq = payments.deep_preflight_request(url) + _quoted = _preq.cost + facts = meter(_preq, x_api_key, response) out = deepcheck.deep_preflight(store, url) store.record_event(creds.sanitize_actor_key(x_api_key) if x_api_key else None, "deep_preflight_run", ua=_ua.get(), endpoint="preflight_deep", - target=url, + target=url, price_credits=_quoted, paid=(facts["settlement_mode"] == "x402"), verdict=(out.get("policy") or {}).get("decision"), **facts) @@ -3064,10 +3130,12 @@ def evidence_bundle_route(body: dict[str, Any], response: Response, "error": "evidence_issuance_refused", "code": e.code, "detail": str(e), "billing": "NOT CHARGED — issuance failed, so no meter ran"}) + _quoted = preq.cost facts = meter(preq, x_api_key, response) store.record_event(creds.sanitize_actor_key(x_api_key) if x_api_key else None, "evidence_bundle_issued", ua=_ua.get(), endpoint="evidence_bundle", target=url, + price_credits=_quoted, paid=(facts["settlement_mode"] == "x402"), **facts) return bundle @@ -3113,6 +3181,13 @@ def watch_provision_route(body: dict[str, Any], response: Response, raise HTTPException(422, str(e)) store.record_event(creds.sanitize_actor_key(x_api_key), "watch_provisioned", ua=_ua.get(), endpoint="watch", target=url) + # PRICE IMPRESSION. A watch has no 402 — provisioning is free — so the + # only moment the caller is shown the per-cycle price is this response. + # Recorded as an impression, explicitly NOT as a payment, so a watch + # pricing experiment can become decidable at all. Without it the boundary + # could never observe a watch price being offered to anyone. + _record_price_impression(creds.sanitize_actor_key(x_api_key), "watch_cycle", + pricing.price("watch_cycle"), _ua.get(), "http") return {**rec, "price_per_cycle_credits": pricing.price("watch_cycle"), "feed": f"GET /watch/{rec['id']}", "billing": ("charged per recheck ACTUALLY performed; provisioning " diff --git a/live/guild/app/mcp_server.py b/live/guild/app/mcp_server.py index f676de9..87b12f6 100644 --- a/live/guild/app/mcp_server.py +++ b/live/guild/app/mcp_server.py @@ -306,6 +306,48 @@ def _mcp_payment(ctx: "Context | None") -> Optional[PaymentPayload]: return None +def _mcp_actor(ctx, api_key: str = "") -> tuple[str, bool]: + """(actor, distinct) for an MCP caller. + + An authenticated caller is its account. An UNAUTHENTICATED one used to + collapse into the literal string "mcp", so a thousand different clients + counted as one actor and an experiment could never reach its denominator. + + We do not invent an identity. The strongest stable, non-secret signal an + MCP caller offers is the clientInfo it advertised at initialize, so the + actor is a purpose-scoped hash of that — stable for one client, different + across clients, reversible to nothing. When the client advertises nothing + distinguishing (a bare `mcp/remote`), distinctness is genuinely UNKNOWABLE + and we say so rather than manufacturing a unique id per call: `distinct` + comes back False and the caller is not counted toward an actor threshold.""" + if api_key: + return _creds.sanitize_actor_key(api_key), True + ua = _client_ua(ctx) or "" + if not ua or ua in ("mcp/remote", "mcp"): + return "mcp:unidentified", False + import hashlib + fp = hashlib.sha256( + ("agent-guild/mcp-actor/" + ua).encode()).hexdigest()[:12] + return f"mcp:net:{fp}", True + + +def _record_paid_offer(preq, ctx, api_key: str = "") -> None: + """MCP twin of the HTTP paid-offer impression — same event, same fields.""" + try: + operation = getattr(preq, "operation", None) + if not operation: + return + actor, distinct = _mcp_actor(ctx, api_key) + store.record_event( + actor, "paid_offer_shown", ua=_client_ua(ctx), + endpoint="x402_challenge", transport="mcp", + challenged_operation=operation, impression="challenge_402", + actor_distinct=distinct, + 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 +439,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 +449,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 @@ -518,6 +562,8 @@ def guild_preflight_deep(url: str, api_key: str = "", ctx: Context = None) -> di Example: guild_preflight_deep(url="https://some-agent.example/a2a") """ + _quoted = payments.deep_preflight_request(url).cost + def _produce(): out = deepcheck.deep_preflight(store, url) facts = settlement_mode() @@ -525,6 +571,7 @@ def _produce(): _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", + price_credits=_quoted, paid=(facts.get("settlement_mode") == "x402"), verdict=(out.get("policy") or {}).get("decision"), **facts) return out @@ -560,6 +607,12 @@ def guild_watch(url: str, api_key: str, interval_seconds: int = 3600, store.record_event(_creds.sanitize_actor_key(api_key), "watch_provisioned", ua=_client_ua(ctx), endpoint="watch", target=url, transport="mcp") + # A watch has no 402; this response is where the per-cycle price is shown. + store.record_event(_creds.sanitize_actor_key(api_key), "paid_offer_shown", + ua=_client_ua(ctx), endpoint="price_shown", + transport="mcp", challenged_operation="watch_cycle", + impression="price_displayed", actor_distinct=True, + price_credits=pricing.price("watch_cycle")) return {**rec, "price_per_cycle_credits": pricing.price("watch_cycle")} diff --git a/live/guild/app/swarm/runner.py b/live/guild/app/swarm/runner.py index a98ad1f..a376da0 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 @@ -234,6 +239,13 @@ def _run_index_cycle(store: Any) -> dict[str, Any]: out["watch_cycles"] = _run_watch_cycles(store) except Exception as exc: # noqa: BLE001 out["watch_error"] = type(exc).__name__ + try: + # Make sure there IS an experiment. An engine with nothing to learn + # from is inert, and idempotent seeding is the only way a fresh + # deployment starts measuring without a human. + out["seeded"] = _experiments.seed_defaults(store) + except Exception as exc: # noqa: BLE001 + out["seed_error"] = type(exc).__name__ try: # DECIDE AND ACT. Calling evaluate() alone produced a recommendation # nobody read — the loop could see that an offer had failed and was 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..d7afd5a 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_shown", + 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_shown", + 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..51d6d57 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_shown", + 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_shown", + 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_shown", + ua="a2a:langchain/0.2.1", endpoint="x402_challenge", + challenged_operation="deep_preflight") + store.record_event("a2a:net:y", "paid_offer_shown", + 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..8b0a273 --- /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_shown", ua=ua, + endpoint="x402_challenge", + challenged_operation=operation, price_credits=20) + + +# -------------------------------------------------------------------------- +# 1. Exposure to a PAID price means having been shown that price +# -------------------------------------------------------------------------- +def test_free_preflight_is_not_exposure_to_the_deep_price(store): + """THE defect: a free preflight caller has never been quoted the deep + price, so counting them let the engine act on an offer nobody saw.""" + for i in range(40): + store.record_event(f"a2a:net:free{i}", "preflight_run", + ua=f"a2a:langchain/0.2.{i}", endpoint="preflight") + exp = experiments.qualified_exposure(store, "deep_preflight") + assert exp["qualified_actors"] == 0, exp + assert exp["paid_offers_shown"] == 0 + + +def test_free_preflight_alone_cannot_make_deep_pricing_decisive(store): + e = experiments.define(store, "deep", hypothesis="h", + variable="price:deep_preflight", + baseline={m: 0 for m in experiments.PRIMARY_METRICS}) + e["min_qualified"] = 3 + store.experiments["deep"] = e + for i in range(50): + store.record_event(f"a2a:net:free{i}", "preflight_run", + ua=f"a2a:langchain/0.2.{i}", endpoint="preflight") + out = experiments.evaluate(store, "deep") + assert out["decision"] in ("hold", "insufficient_evidence") + assert out["decision"] != "kill" + before = pricing.price("deep_preflight") + experiments.apply_next_action(store) + assert pricing.price("deep_preflight") == before, \ + "a price nobody was offered must not move" + + +def test_a_genuine_external_deep_challenge_IS_decisive(store): + """The other half: real exposure to the real price must be able to decide.""" + e = experiments.define(store, "deep", hypothesis="h", + variable="price:deep_preflight", + baseline={m: 0 for m in experiments.PRIMARY_METRICS}) + e["min_qualified"] = 2 + store.experiments["deep"] = e + _challenge(store, "deep_preflight", "a2a:net:shown1", "a2a:langchain/0.2.1") + _challenge(store, "deep_preflight", "a2a:net:shown2", "a2a:crewai/1.0") + exp = experiments.qualified_exposure(store, "deep_preflight") + assert exp["qualified_actors"] == 2 + assert exp["paid_offers_shown"] == 2 + assert experiments.evaluate(store, "deep")["decision"] == "kill" + + +def test_a_challenge_for_another_operation_is_not_exposure(store): + _challenge(store, "evidence_bundle", "a2a:net:other") + assert experiments.qualified_exposure( + store, "deep_preflight")["qualified_actors"] == 0 + assert experiments.qualified_exposure( + store, "evidence_bundle")["qualified_actors"] == 1 + + +def test_crawler_challenges_are_not_exposure(store): + _challenge(store, "deep_preflight", "a2a:net:bot", + ua="a2a:AgenstryBot/0.3.0") + assert experiments.qualified_exposure( + store, "deep_preflight")["qualified_actors"] == 0 + + +def test_a_paid_completion_also_counts_as_exposure(store): + """Someone who saw the price and PAID it has certainly been exposed.""" + store.record_event("a2a:net:payer", "deep_preflight_run", ua=EXT_UA, + endpoint="preflight_deep", settlement_mode="x402", + settlement_confirmed=True, settlement_mainnet=True) + exp = experiments.qualified_exposure(store, "deep_preflight") + assert exp["qualified_actors"] == 1 + assert exp["paid_completions"] == 1 + + +def test_a_free_tier_call_of_the_same_shape_is_not_a_completion(store): + store.record_event("a2a:net:freebie", "deep_preflight_run", ua=EXT_UA, + endpoint="preflight_deep", settlement_mode="free") + assert experiments.qualified_exposure( + store, "deep_preflight")["qualified_actors"] == 0 + + +def test_evidence_bundle_and_watch_have_the_same_boundary(store): + for op in ("evidence_bundle", "watch_cycle"): + assert experiments.qualified_exposure(store, op)["qualified_actors"] == 0 + _challenge(store, "watch_cycle", "a2a:net:w1") + assert experiments.qualified_exposure( + store, "watch_cycle")["qualified_actors"] == 1 + assert experiments.qualified_exposure( + store, "evidence_bundle")["qualified_actors"] == 0 + + +def test_portfolio_view_is_labelled_as_not_valid_for_pricing(store): + out = experiments.qualified_exposure(store) + assert out["operation_scope"] == "all_surfaces" + assert "NOT valid for pricing" in out["rule"] + + +# -------------------------------------------------------------------------- +# 2. DID coalescing — one subject, several addresses +# -------------------------------------------------------------------------- +DID_A = "did:key:z6MkExampleAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA" +DID_B = "did:key:z6MkExampleBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB" + + +def test_two_endpoints_with_one_did_are_one_subject(store): + out = indexops.ingest(store, [ + {"endpoint": "https://a.example/a2a", "source": "s1", "did": DID_A}, + {"endpoint": "https://b.example/a2a", "source": "s2", "did": DID_A}, + ]) + assert out["added"] == 1 + assert out["aliased_to_existing_did"] == 1 + assert len(store.trust_index) == 1 + entry = next(iter(store.trust_index.values())) + assert [a["endpoint"] for a in entry["alias_endpoints"]] == \ + ["https://b.example/a2a"] + + +def test_alias_retains_its_own_provenance(store): + indexops.ingest(store, [ + {"endpoint": "https://a.example/a2a", "source": "mcp_registry", "did": DID_A}, + {"endpoint": "https://b.example/a2a", "source": "guild_registration", + "did": DID_A}, + ]) + entry = next(iter(store.trust_index.values())) + alias = entry["alias_endpoints"][0] + assert [s["source"] for s in alias["sources"]] == ["guild_registration"] + # and the canonical entry records that the second source saw the subject + assert {s["source"] for s in entry["sources"]} == {"mcp_registry", + "guild_registration"} + + +def test_different_dids_are_never_merged(store): + indexops.ingest(store, [ + {"endpoint": "https://a.example/a2a", "source": "s", "did": DID_A}, + {"endpoint": "https://b.example/a2a", "source": "s", "did": DID_B}, + ]) + assert len(store.trust_index) == 2 + + +def test_operator_equivalence_is_never_inferred_from_names(store): + """Two endpoints of the SAME company, no declared identity. Guessing would + launder one party's evidence into another's.""" + indexops.ingest(store, [ + {"endpoint": "https://api.acme.example/a2a", "source": "s", + "declared": {"name": "Acme Agent"}}, + {"endpoint": "https://eu.acme.example/a2a", "source": "s", + "declared": {"name": "Acme Agent"}}, + ]) + assert len(store.trust_index) == 2 + view = trustindex.public_view(next(iter(store.trust_index.values()))) + assert "never inferred" in view["identity"]["operator"] + + +def test_reconciliation_migrates_pre_existing_duplicates(store): + """The index shipped keyed on endpoint alone, so duplicates already exist.""" + for host in ("a", "b", "c"): + e = trustindex.new_entry(f"https://{host}.example/a2a", "s", did=DID_A) + e["first_indexed_at"] = f"2026-07-{10 + ord(host) - 97}T00:00:00+00:00" + store.trust_index[e["id"]] = e + assert len(store.trust_index) == 3 + out = indexops.reconcile_identities(store) + assert out["merged_into_canonical"] == 2 + assert len(store.trust_index) == 1 + canonical = next(iter(store.trust_index.values())) + assert canonical["endpoint"] == "https://a.example/a2a", "oldest wins" + assert len(canonical["alias_endpoints"]) == 2 + + +def test_reconciliation_is_idempotent(store): + for host in ("a", "b"): + e = trustindex.new_entry(f"https://{host}.example/a2a", "s", did=DID_A) + store.trust_index[e["id"]] = e + indexops.reconcile_identities(store) + again = indexops.reconcile_identities(store) + assert again["merged_into_canonical"] == 0 + assert len(store.trust_index) == 1 + assert len(next(iter(store.trust_index.values()))["alias_endpoints"]) == 1 + + +def test_reconciliation_is_a_noop_without_declared_identity(store): + for host in ("a", "b"): + e = trustindex.new_entry(f"https://{host}.example/a2a", "s") + store.trust_index[e["id"]] = e + assert indexops.reconcile_identities(store)["merged_into_canonical"] == 0 + assert len(store.trust_index) == 2 + + +def test_summary_reports_subjects_and_endpoints_separately(store): + indexops.ingest(store, [ + {"endpoint": "https://a.example/a2a", "source": "s", "did": DID_A}, + {"endpoint": "https://b.example/a2a", "source": "s", "did": DID_A}, + ]) + summary = trustindex.summarise(store.trust_index.values()) + assert summary["total_entries"] == 1 + assert summary["alias_endpoints_folded"] == 1 + assert summary["distinct_endpoints_known"] == 2 + assert "operator identity is NEVER inferred" in summary["dedupe"] diff --git a/live/guild/tests/test_paid_impressions_transport.py b/live/guild/tests/test_paid_impressions_transport.py new file mode 100644 index 0000000..bf4c45e --- /dev/null +++ b/live/guild/tests/test_paid_impressions_transport.py @@ -0,0 +1,527 @@ +"""Paid-offer impressions through the REAL transports. + +The previous round defined the impression event correctly and then failed to +populate it on any real path — the events existed, the counters moved, and the +number the engine gates on (distinct ACTORS) stayed at zero. Hand-constructed +event tests could not see that, because they wrote the events themselves. + +So every test here drives an actual request through FastAPI, the mounted MCP +tool, or the A2A JSON-RPC endpoint, and then asks the experiment engine what it +learned. Three specific regressions are locked: + + * HTTP recorded `actor=None`, so a genuine external challenge never + incremented qualified_actors; + * unauthenticated MCP callers collapsed into the literal actor "mcp", so a + thousand distinct clients counted as one; + * a watch has no 402 at all, so its price could never be observed as offered + to anyone and a watch pricing experiment could never become decidable. +""" +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, pricing # noqa: E402 +from app.store import Store # noqa: E402 + +EXT_UA = "langchain/0.2.1" + + +@pytest.fixture() +def store(tmp_path, monkeypatch) -> Store: + pricing.load_runtime({}) + s = Store(path=str(tmp_path / "guild.json")) + import app.main as main_mod + import app.a2a as a2a_mod + import app.mcp_server as mcp_mod + monkeypatch.setattr(main_mod, "store", s) + monkeypatch.setattr(a2a_mod, "store", s) + monkeypatch.setattr(mcp_mod, "store", s) + return s + + +@pytest.fixture() +def client(): + from fastapi.testclient import TestClient + from app.main import app + return TestClient(app) + + +@pytest.fixture() +def enforced(monkeypatch): + """Turn on billing enforcement so priced reads actually challenge.""" + monkeypatch.setenv("GUILD_BILLING_ENFORCED", "1") + + +def _impressions(store: Store, operation: str): + return [e for e in store.events + if e.get("type") == "paid_offer_shown" + and e.get("challenged_operation") == operation] + + +# -------------------------------------------------------------------------- +# HTTP +# -------------------------------------------------------------------------- +def test_http_challenge_carries_a_real_distinct_actor(store, client, enforced): + """The defect: actor=None, so events moved and qualified_actors never did.""" + r = client.get("/preflight/deep", params={"url": "https://x.example/a2a"}, + headers={"user-agent": EXT_UA}) + assert r.status_code == 402, r.status_code + shown = _impressions(store, "deep_preflight") + assert len(shown) == 1 + actor = shown[0]["key"] + assert actor and actor != "anon", "the challenge must be attributable" + assert actor.startswith("http:") + assert "langchain" not in actor, "the actor must not embed raw caller data" + + +def test_http_challenges_from_different_callers_are_different_actors( + store, client, enforced): + for ua in ("langchain/0.2.1", "crewai/1.0", "llamaindex/0.9"): + client.get("/preflight/deep", params={"url": "https://x.example/a2a"}, + headers={"user-agent": ua}) + actors = {e["key"] for e in _impressions(store, "deep_preflight")} + assert len(actors) == 3, actors + + +def test_http_free_preflight_records_no_paid_impression(store, client): + r = client.get("/preflight", params={"url": "https://x.example/a2a"}, + headers={"user-agent": EXT_UA}) + assert r.status_code == 200 + assert _impressions(store, "deep_preflight") == [] + + +def test_free_preflight_alone_cannot_make_deep_pricing_decisive_over_http( + store, client): + 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 + for i in range(30): + client.get("/preflight", params={"url": f"https://x{i}.example/a2a"}, + headers={"user-agent": f"langchain/0.2.{i}"}) + assert experiments.qualified_exposure( + store, "deep_preflight")["qualified_actors"] == 0 + before = pricing.price("deep_preflight") + experiments.apply_next_action(store) + assert pricing.price("deep_preflight") == before + + +def test_genuine_http_challenges_make_deep_pricing_decidable( + store, client, enforced): + 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 + for ua in ("langchain/0.2.1", "crewai/1.0"): + client.get("/preflight/deep", params={"url": "https://x.example/a2a"}, + headers={"user-agent": ua}) + exp = experiments.qualified_exposure(store, "deep_preflight") + assert exp["qualified_actors"] == 2, exp + assert experiments.evaluate(store, "deep")["decision"] == "kill" + + +# -------------------------------------------------------------------------- +# Watch — priced, but never behind a 402 +# -------------------------------------------------------------------------- +def _funded(store: Store, name="w"): + rec = store.register_agent(name, ["x"], {}) + raw = rec.get("api_key") + store.credit(store._account_key(raw), 1000, reason="test") + return raw + + +def test_http_watch_response_records_the_price_impression(store, client): + raw = _funded(store) + r = client.post("/watch", json={"url": "https://a.example/a2a"}, + headers={"x-api-key": raw, "user-agent": EXT_UA}) + assert r.status_code == 200 + assert r.json()["price_per_cycle_credits"] == pricing.price("watch_cycle") + shown = _impressions(store, "watch_cycle") + assert len(shown) == 1 + assert shown[0]["impression"] == "price_displayed" + assert shown[0]["price_credits"] == pricing.price("watch_cycle") + + +def test_a_price_impression_is_never_recorded_as_a_payment(store, client): + raw = _funded(store) + client.post("/watch", json={"url": "https://a.example/a2a"}, + headers={"x-api-key": raw, "user-agent": EXT_UA}) + shown = _impressions(store, "watch_cycle")[0] + assert not shown.get("paid") + assert shown.get("settlement_mode") is None + m = experiments.commercial_metrics(store, "watch_cycle") + assert m["paid_decisions"] == 0, "provisioning is free and must stay free" + + +def test_a_watch_experiment_can_become_decidable(store, client): + """Previously impossible: no 402 meant no impression, ever.""" + e = experiments.define(store, "w", hypothesis="h", + variable="price:watch_cycle", + baseline={m: 0 for m in experiments.PRIMARY_METRICS}) + e["min_qualified"] = 2 + store.experiments["w"] = e + for i in range(2): + raw = _funded(store, f"w{i}") + client.post("/watch", json={"url": f"https://a{i}.example/a2a"}, + headers={"x-api-key": raw, "user-agent": f"langchain/0.2.{i}"}) + exp = experiments.qualified_exposure(store, "watch_cycle") + assert exp["qualified_actors"] == 2, exp + assert experiments.evaluate(store, "w")["decision"] in ("kill", "promote") + + +# -------------------------------------------------------------------------- +# MCP +# -------------------------------------------------------------------------- +def _tool(name): + import app.mcp_server as m + tool = getattr(m, name) + for attr in ("fn", "func", "__wrapped__"): + f = getattr(tool, attr, None) + if callable(f): + return f + return tool + + +class _Ctx: + """Minimal stand-in for the MCP Context clientInfo handshake.""" + + def __init__(self, name, version="1.0"): + class _CI: + pass + ci = _CI() + ci.name, ci.version = name, version + + class _P: + clientInfo = ci + + class _S: + client_params = _P() + self.session = _S() + + +def test_unauthenticated_mcp_callers_are_not_all_one_actor(store, monkeypatch): + """The defect: every unauthenticated caller collapsed into literal "mcp".""" + import app.mcp_server as m + a, a_distinct = m._mcp_actor(_Ctx("alpha-client")) + b, b_distinct = m._mcp_actor(_Ctx("beta-client")) + assert a != b + assert a_distinct and b_distinct + assert a.startswith("mcp:net:") + assert "alpha-client" not in a, "the actor must not embed the raw client id" + + +def test_the_same_mcp_client_is_a_stable_actor(store): + import app.mcp_server as m + assert m._mcp_actor(_Ctx("same"))[0] == m._mcp_actor(_Ctx("same"))[0] + + +def test_an_unidentifiable_mcp_caller_is_explicitly_not_distinct(store): + """We do not invent identities. When distinctness is unknowable we say so, + and the caller does not count toward an actor threshold.""" + import app.mcp_server as m + actor, distinct = m._mcp_actor(None) + assert distinct is False + assert actor == "mcp:unidentified" + + +def test_indistinct_mcp_impressions_do_not_reach_the_threshold(store): + for _ in range(50): + store.record_event("mcp:unidentified", "paid_offer_shown", + ua="mcp/remote", endpoint="x402_challenge", + challenged_operation="deep_preflight", + impression="challenge_402", actor_distinct=False) + assert experiments.qualified_exposure( + store, "deep_preflight")["qualified_actors"] == 0 + + +def test_mcp_watch_tool_records_the_price_impression(store): + raw = _funded(store, "mcpw") + out = _tool("guild_watch")(url="https://a.example/a2a", api_key=raw, + ctx=_Ctx("langchain")) + assert out["price_per_cycle_credits"] == pricing.price("watch_cycle") + assert len(_impressions(store, "watch_cycle")) == 1 + + +# -------------------------------------------------------------------------- +# A2A +# -------------------------------------------------------------------------- +def _a2a(client, text, ua=EXT_UA): + return client.post("/a2a", headers={"user-agent": ua}, json={ + "jsonrpc": "2.0", "id": "1", "method": "message/send", + "params": {"message": {"role": "user", + "parts": [{"kind": "text", "text": text}]}}}) + + +def test_a2a_free_preflight_records_no_paid_impression(store, client, monkeypatch): + import app.a2a as a2a_mod + monkeypatch.setattr(a2a_mod.preflight, "run", + lambda url, store=None: {"verdict": "no_failed_checks", + "failed": [], "unknowns": [], + "checks": []}) + r = _a2a(client, "preflight: https://x.example/a2a") + assert r.status_code == 200 + assert _impressions(store, "deep_preflight") == [] + + +def test_a2a_deep_preflight_challenge_records_a_distinct_actor( + store, client, monkeypatch): + import app.a2a as a2a_mod + monkeypatch.setattr(a2a_mod, "_x402_a2a_active", lambda: True) + r = _a2a(client, "deep-preflight: https://x.example/a2a") + assert r.status_code == 200 + shown = _impressions(store, "deep_preflight") + assert len(shown) == 1 + assert shown[0]["key"].startswith("a2a:") + assert shown[0]["impression"] == "challenge_402" + + +# -------------------------------------------------------------------------- +# Seeding — an engine with no experiment is inert +# -------------------------------------------------------------------------- +def test_a_fresh_store_cycle_seeds_exactly_one_experiment(store, monkeypatch): + """Deployed 061dcea returned experiments: {} — the engine had nothing to + learn from and could never satisfy 'find the formula without a human'.""" + from app.swarm import runner + monkeypatch.setenv("GUILD_INDEX_AUTORUN", "1") + monkeypatch.setattr(runner, "_run_watch_cycles", lambda s, cap=10: {}) + monkeypatch.setattr("app.indexops.recheck_due", + lambda s, **kw: {"checked": 0}) + monkeypatch.setattr("app.indexops.ingest", lambda s, r=None: {"added": 0}) + assert store.experiments == {} + out = runner._run_index_cycle(store) + assert out["seeded"]["seeded"] == ["deep_preflight_price_v1"] + assert len(store.experiments) == 1 + rec = store.experiments["deep_preflight_price_v1"] + assert rec["variable"] == "price:deep_preflight" + assert rec["baseline"]["operation_scope"] == "deep_preflight" + + +def test_seeding_is_idempotent_and_never_resets_the_window(store, monkeypatch): + """A seeder that runs every cycle is one bug away from continuously + resetting the thing it is meant to measure.""" + experiments.seed_defaults(store) + rec = dict(store.experiments["deep_preflight_price_v1"]) + started, baseline = rec["started_at"], rec["baseline"] + for _ in range(3): + out = experiments.seed_defaults(store) + assert out["seeded"] == [] + live = store.experiments["deep_preflight_price_v1"] + assert live["started_at"] == started + assert live["baseline"] == baseline + assert len(store.experiments) == 1 + + +def test_a_restart_does_not_reset_the_experiment(store, tmp_path): + experiments.seed_defaults(store) + started = store.experiments["deep_preflight_price_v1"]["started_at"] + reloaded = Store(path=str(tmp_path / "guild.json")) + experiments.seed_defaults(reloaded) + assert reloaded.experiments["deep_preflight_price_v1"]["started_at"] == started + + +def test_an_operator_pinned_price_is_not_seeded(store, monkeypatch): + monkeypatch.setenv("GUILD_PRICE_DEEP_PREFLIGHT", "20") + out = experiments.seed_defaults(store) + assert out["seeded"] == [] + assert out["already_present"][0]["reason"] == "price_pinned_by_operator" + + +def test_a_seeded_experiment_without_exposure_is_not_decisive(store): + """Seeding must not fabricate exposure: the honest state of a product + nobody has been offered is insufficient_evidence, not a verdict.""" + experiments.seed_defaults(store) + out = experiments.evaluate(store, "deep_preflight_price_v1") + assert out["decision"] in ("hold", "insufficient_evidence") + before = pricing.price("deep_preflight") + experiments.apply_next_action(store) + assert pricing.price("deep_preflight") == before + + +# -------------------------------------------------------------------------- +# Treatment window + exact treatment — the stale-evidence defect +# -------------------------------------------------------------------------- +def _impression(store: Store, actor: str, operation: str, price: int, + at: str = None): + store.record_event(actor, "paid_offer_shown", ua="langchain/0.2.1", + endpoint="x402_challenge", + challenged_operation=operation, + impression="challenge_402", price_credits=price) + if at: + store.events[-1]["at"] = at + + +def test_old_price_impressions_cannot_decide_the_new_price_window(store): + """THE defect: ten callers see 20 credits and do not buy, the engine cuts + to 10, and the next cycle reuses those same ten impressions to justify + cutting 10 to 5 — even though nobody has been shown 10.""" + e = experiments.define(store, "deep", hypothesis="h", + variable="price:deep_preflight", + baseline={m: 0 for m in experiments.PRIMARY_METRICS}, + tested_price_credits=20) + e["min_qualified"] = 2 + store.experiments["deep"] = e + for i in range(4): + _impression(store, f"a2a:net:old{i}", "deep_preflight", 20) + + first = experiments.apply_next_action(store) + assert first[0]["acted"] is True + new_price = pricing.price("deep_preflight") + assert new_price == 10 + + # The SAME old-price impressions must not decide the new arm. + exposure = experiments.qualified_exposure( + store, "deep_preflight", + since=store.experiments["deep"]["started_at"], + tested_price_credits=new_price) + assert exposure["qualified_actors"] == 0, exposure + second = experiments.apply_next_action(store) + assert second[0].get("acted") is not True + assert pricing.price("deep_preflight") == new_price, \ + "a second cut on stale evidence is exactly the defect" + + +def test_impressions_of_the_new_price_do_decide_the_new_window(store): + e = experiments.define(store, "deep", hypothesis="h", + variable="price:deep_preflight", + baseline={m: 0 for m in experiments.PRIMARY_METRICS}, + tested_price_credits=20) + e["min_qualified"] = 2 + store.experiments["deep"] = e + for i in range(2): + _impression(store, f"a2a:net:old{i}", "deep_preflight", 20) + experiments.apply_next_action(store) + assert pricing.price("deep_preflight") == 10 + # now two callers actually see 10 + for i in range(2): + _impression(store, f"a2a:net:new{i}", "deep_preflight", 10) + out = experiments.apply_next_action(store) + assert out[0]["acted"] is True + assert pricing.price("deep_preflight") == 5 + + +def test_an_impression_from_before_the_window_is_excluded(store): + experiments.define(store, "deep", hypothesis="h", + variable="price:deep_preflight", + baseline={m: 0 for m in experiments.PRIMARY_METRICS}, + tested_price_credits=20) + _impression(store, "a2a:net:ancient", "deep_preflight", 20, + at="2020-01-01T00:00:00+00:00") + exposure = experiments.qualified_exposure( + store, "deep_preflight", + since=store.experiments["deep"]["started_at"], + tested_price_credits=20) + assert exposure["qualified_actors"] == 0 + + +def test_unparseable_event_times_fail_closed(store): + experiments.define(store, "deep", hypothesis="h", + variable="price:deep_preflight", + baseline={m: 0 for m in experiments.PRIMARY_METRICS}, + tested_price_credits=20) + _impression(store, "a2a:net:bad", "deep_preflight", 20, at="not-a-date") + exposure = experiments.qualified_exposure( + store, "deep_preflight", + since=store.experiments["deep"]["started_at"], + tested_price_credits=20) + assert exposure["qualified_actors"] == 0, \ + "an unreadable timestamp must never be able to decide a price" + + +def test_a_payment_quoted_at_the_old_price_cannot_promote_the_new_one(store): + """The money is real; it is simply evidence about the price the payer was + actually shown.""" + experiments.define(store, "deep", hypothesis="h", + variable="price:deep_preflight", + baseline={m: 0 for m in experiments.PRIMARY_METRICS}, + tested_price_credits=10) + store.record_event("a2a:net:payer", "deep_preflight_run", + ua="langchain/0.2.1", endpoint="preflight_deep", + price_credits=20, # quoted under the OLD arm + settlement_mode="x402", settlement_confirmed=True, + settlement_mainnet=True, settlement_amount_atomic=20000) + m = experiments.commercial_metrics( + store, "deep_preflight", + since=store.experiments["deep"]["started_at"], + tested_price_credits=10) + assert m["paid_decisions"] == 0 + assert m["external_settled_revenue_usd"] == 0.0 + + +def test_the_portfolio_report_is_not_scoped_away(store): + """Scoping applies to EXPERIMENT EVALUATION only — the commercial report + must still show everything that happened.""" + store.record_event("a2a:net:payer", "deep_preflight_run", + ua="langchain/0.2.1", endpoint="preflight_deep", + price_credits=20, settlement_mode="x402", + settlement_confirmed=True, settlement_mainnet=True, + settlement_amount_atomic=20000) + assert experiments.commercial_metrics(store)["paid_decisions"] == 1 + assert experiments.commercial_metrics( + store, "deep_preflight")["paid_decisions"] == 1 + + +def test_reprice_updates_window_and_treatment_together(store): + e = experiments.define(store, "deep", hypothesis="h", + variable="price:deep_preflight", + baseline={m: 0 for m in experiments.PRIMARY_METRICS}, + tested_price_credits=20) + e["min_qualified"] = 1 + store.experiments["deep"] = e + started = e["started_at"] + _impression(store, "a2a:net:one", "deep_preflight", 20) + experiments.apply_next_action(store) + live = store.experiments["deep"] + assert live["tested_price_credits"] == 10 + assert live["started_at"] >= started + assert live["baseline"]["operation_scope"] == "deep_preflight" + + +def test_a_completion_with_no_recorded_price_is_excluded(store): + """Fail closed: a payment we cannot place in this arm must not promote it. + Excluding a real sale understates us; admitting an unattributable one lets + any historical payment promote a price it was never quoted at.""" + experiments.define(store, "deep", hypothesis="h", + variable="price:deep_preflight", + baseline={m: 0 for m in experiments.PRIMARY_METRICS}, + tested_price_credits=10) + store.record_event("a2a:net:payer", "deep_preflight_run", + ua="langchain/0.2.1", endpoint="preflight_deep", + settlement_mode="x402", settlement_confirmed=True, + settlement_mainnet=True, settlement_amount_atomic=20000) + m = experiments.commercial_metrics( + store, "deep_preflight", + since=store.experiments["deep"]["started_at"], + tested_price_credits=10) + assert m["paid_decisions"] == 0 + assert m["external_settled_revenue_usd"] == 0.0 + + +def test_historical_revenue_cannot_suppress_a_new_arm_sale(store): + """Seed baseline must be in the SAME frame as the comparison, or a real + new-arm sale reads as 'no movement' against all-time history.""" + store.record_event("a2a:net:old", "deep_preflight_run", + ua="langchain/0.2.1", endpoint="preflight_deep", + price_credits=20, settlement_mode="x402", + settlement_confirmed=True, settlement_mainnet=True, + settlement_amount_atomic=99000) + experiments.seed_defaults(store) + rec = store.experiments["deep_preflight_price_v1"] + assert rec["baseline"]["paid_decisions"] == 0, rec["baseline"] + assert rec["baseline"]["external_settled_revenue_usd"] == 0.0 + + rec["min_qualified"] = 1 + store.experiments["deep_preflight_price_v1"] = rec + store.record_event("a2a:net:new", "deep_preflight_run", + ua="crewai/1.0", endpoint="preflight_deep", + price_credits=rec["tested_price_credits"], + settlement_mode="x402", settlement_confirmed=True, + settlement_mainnet=True, settlement_amount_atomic=20000) + out = experiments.evaluate(store, "deep_preflight_price_v1") + assert out["decision"] == "promote", out["evidence"]