From 133ae19825051c9e3ace81a5dbd9c7c29cdb68cb Mon Sep 17 00:00:00 2001 From: AgentTanuki Date: Fri, 31 Jul 2026 13:54:05 +0100 Subject: [PATCH 1/3] Third and fourth corrections: real impressions on every transport, DID coalescing, and an engine that is not inert Rebased onto main so one branch carries both rounds (the earlier fixes3 branch did not merge; its content is included here unchanged). EXPOSURE TO A PRICE MEANS BEING SHOWN THAT PRICE. qualified_exposure counted free preflight_run events toward the PAID deep-preflight experiment, so the engine could reach its denominator on callers who were never offered anything and then halve a price nobody saw. There is now an explicit impression boundary: a canonical `paid_offer_shown` event carrying `challenged_operation`, recorded identically on HTTP, MCP and A2A. Two impressions qualify, distinguished by `impression` so they are never conflated: a 402 quote, and a price DISPLAYED in a successful response where there is no 402 (a watch is provisioned free and priced per cycle, so its response is the only moment its price is shown - previously a watch pricing experiment could never become decidable at all). Adjacent free products never count, and a free-tier call of a paid operation's shape is not an impression. THE IMPRESSION WAS DEFINED CORRECTLY AND POPULATED ON NO REAL PATH. Events moved; distinct ACTORS - the number the engine gates on - stayed at zero. * HTTP recorded actor=None; the middleware now binds the same stable, privacy-safe actor used for demand dedupe into a request-scoped contextvar the challenge path reads. * Unauthenticated MCP callers collapsed into the literal actor "mcp". The actor is now a purpose-scoped hash of advertised clientInfo. We do NOT invent identities: a caller advertising nothing distinguishing is "mcp:unidentified" with actor_distinct=False and is excluded from actor thresholds rather than given a fake unique id per call. ONE DID IS ONE SUBJECT. The module documented endpoint-then-identity dedupe and performed endpoint dedupe only, inflating the one number the index is judged on. Endpoints declaring the same did:key are now coalesced at ingest into a canonical entry with `alias_endpoints` retaining their own provenance and last observation; summaries report subjects and endpoints separately. reconcile_identities() migrates existing duplicates - oldest wins, deterministic, idempotent, no-op when there is nothing to merge. OPERATOR equivalence is never inferred from names, domains or contact strings: guessing would launder one party's evidence into another's, a worse error than the one being fixed. THE ENGINE HAD NO EXPERIMENT. GET /commercial on deployed 061dcea returned experiments: {} - nothing ever created one, so the engine evaluated an empty dict forever. seed_defaults() runs on the real cycle path and is idempotent three ways: never overwrites an existing experiment (a restart cannot reset its window or baseline), never seeds one whose price an operator has pinned, and seeds exactly ONE. It does not fabricate exposure - a seeded experiment nobody has been offered reports insufficient_evidence, which is the honest state. Also: remote ingest is bounded default-ON for CLEARED_SOURCES only (the documented public MCP Registry API) with GUILD_INDEX_INGEST=0 retained as a no-deploy kill switch; excluded sources are named with their exact gate rather than silently omitted. Tests: 1150 passed, 9 skipped. The transport suite drives real FastAPI, MCP-tool and A2A JSON-RPC paths rather than hand-built events. The 061dcea auto-revert was a DEPLOY-TIMEOUT FALSE NEGATIVE: the gate stopped waiting before Render finished, production then served the merged SHA and was healthy. The revert branch is inert (the ship job skips ship/revert-*) and must not be merged. --- live/guild/app/a2a.py | 12 + live/guild/app/experiments.py | 205 ++++++++--- live/guild/app/indexops.py | 83 ++++- live/guild/app/main.py | 71 ++++ live/guild/app/mcp_server.py | 50 +++ live/guild/app/swarm/runner.py | 12 + 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 ++++++++++++ .../tests/test_paid_impressions_transport.py | 344 ++++++++++++++++++ 11 files changed, 1050 insertions(+), 71 deletions(-) create mode 100644 live/guild/tests/test_index_fixes3.py create mode 100644 live/guild/tests/test_paid_impressions_transport.py 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/experiments.py b/live/guild/app/experiments.py index 8eb4184..189c4e5 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,106 @@ 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_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 + 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 ) -> dict[str, Any]: """The primary metrics. Revenue is REAL money only. @@ -405,6 +458,64 @@ 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 + define(store, key, hypothesis=spec["hypothesis"], + variable=spec["variable"], + baseline=commercial_metrics(store, operation)) + 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. 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..e1f5b75 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 @@ -3113,6 +3177,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..1621660 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 @@ -560,6 +604,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..9af2208 --- /dev/null +++ b/live/guild/tests/test_paid_impressions_transport.py @@ -0,0 +1,344 @@ +"""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 From 2d86cd7d2be34ef8e7821a7315374086cfb5e102 Mon Sep 17 00:00:00 2001 From: AgentTanuki Date: Fri, 31 Jul 2026 13:57:17 +0100 Subject: [PATCH 2/3] Bind every price experiment to its treatment window and exact treatment Experimental-validity defect, and the most consequential of the round: after repricing, evaluate() reset started_at and the baseline, but qualified_exposure() still counted ALL historical impressions for the operation. So ten callers seeing 20 credits and not buying triggered a cut to 10 - and on the next cycle those same ten old-price impressions were counted again to justify cutting 10 to 5, though nobody had been shown 10. Left alone, the loop repeatedly optimises on stale treatment data and walks a price to zero without ever testing any of the prices in between. An arm is (window, price), and the three now move together: * `tested_price_credits` is persisted on define/seed and updated on every reprice, alongside started_at and a re-taken baseline; * exposure used by evaluate requires event.at >= started_at AND event.price_credits == tested_price_credits - an impression is evidence about the price that was actually displayed, and about nothing else; * completions carry the credits actually QUOTED, not today's price. HTTP and MCP stamp the price at request time; A2A uses the credits_cost stored on the payment task, so a payment quoted under the old arm and settled late cannot promote the new one. The money is real - it is simply evidence about the price the payer was shown. Event-time parsing FAILS CLOSED: an event whose timestamp cannot be parsed is excluded, never admitted. An unreadable date must not be able to decide a price. Scoping applies to EXPERIMENT EVALUATION ONLY. The portfolio view at /commercial is deliberately unscoped and still reports everything that happened, with a test proving it was not narrowed by this change. Tests: 1157 passed, 9 skipped (7 new), including the exact regression - old price impressions cannot decide the new window, new-price impressions can, pre-window and unparseable events are excluded, and an old-quoted payment cannot promote the new price. --- live/guild/app/a2a_x402.py | 9 +- live/guild/app/experiments.py | 83 +++++++++-- live/guild/app/main.py | 8 +- live/guild/app/mcp_server.py | 3 + .../tests/test_paid_impressions_transport.py | 139 ++++++++++++++++++ 5 files changed, 229 insertions(+), 13 deletions(-) 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 189c4e5..54c0deb 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(), @@ -158,10 +162,39 @@ def _is_external(event: dict) -> bool: and attribution.is_genuine_external(event)) -def qualified_exposure(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 @@ -223,6 +256,12 @@ def _external(e: dict) -> bool: 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 @@ -258,7 +297,9 @@ def _external(e: dict) -> bool: } -def commercial_metrics(store: Any, operation: Optional[str] = None +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. @@ -282,6 +323,15 @@ 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 + if (tested_price_credits is not None + and e.get("price_credits") is not None + and e.get("price_credits") != 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"): @@ -357,8 +407,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: @@ -393,6 +447,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} @@ -509,7 +568,8 @@ def seed_defaults(store: Any) -> dict[str, Any]: continue define(store, key, hypothesis=spec["hypothesis"], variable=spec["variable"], - baseline=commercial_metrics(store, operation)) + baseline=commercial_metrics(store, operation), + tested_price_credits=pricing.price(operation)) seeded.append(key) return {"seeded": seeded, "already_present": skipped, "note": ("idempotent: an existing experiment is never overwritten, " @@ -599,10 +659,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/main.py b/live/guild/app/main.py index e1f5b75..dbcd449 100644 --- a/live/guild/app/main.py +++ b/live/guild/app/main.py @@ -3093,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) @@ -3128,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 diff --git a/live/guild/app/mcp_server.py b/live/guild/app/mcp_server.py index 1621660..87b12f6 100644 --- a/live/guild/app/mcp_server.py +++ b/live/guild/app/mcp_server.py @@ -562,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() @@ -569,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 diff --git a/live/guild/tests/test_paid_impressions_transport.py b/live/guild/tests/test_paid_impressions_transport.py index 9af2208..f984676 100644 --- a/live/guild/tests/test_paid_impressions_transport.py +++ b/live/guild/tests/test_paid_impressions_transport.py @@ -342,3 +342,142 @@ def test_a_seeded_experiment_without_exposure_is_not_decisive(store): 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" From fe76e3f2f8ac5f29658efe474f6ce697e91d6745 Mon Sep 17 00:00:00 2001 From: AgentTanuki Date: Fri, 31 Jul 2026 14:02:29 +0100 Subject: [PATCH 3/3] Fail closed on exact-treatment scoping and seed the baseline in the arm's own frame Two edge cases in 2d86cd7, both of which failed OPEN. 1. EXACT-TREATMENT SCOPING ADMITTED UNPRICED COMPLETIONS. The filter excluded a completion only when price_credits was present AND different, so a paid completion with no recorded price passed straight through and could promote a price it was never quoted at. When a tested price is supplied, price_credits must now EXIST and equal it. Excluding a real sale understates us; admitting an unattributable one is the failure this whole round exists to prevent. 2. THE SEED BASELINE WAS IN A DIFFERENT FRAME FROM THE COMPARISON. seed_defaults took an ALL-TIME baseline while evaluate() measures the arm (since started_at, at the tested price). Any historical revenue therefore sat in the baseline while a genuine new-arm sale sat in the metrics, and the sale read as "no movement" - an experiment that could never promote on a real result. The baseline is now taken in the arm's own window and at its own price, so an arm opens at zero, which is the truth about a window that has just started. Tests: 1159 passed, 9 skipped (2 new) - a missing-price completion is excluded, and historical revenue before seeding cannot suppress promotion from a genuine new-arm sale. --- live/guild/app/experiments.py | 35 +++++++++++---- .../tests/test_paid_impressions_transport.py | 44 +++++++++++++++++++ 2 files changed, 71 insertions(+), 8 deletions(-) diff --git a/live/guild/app/experiments.py b/live/guild/app/experiments.py index 54c0deb..bc28032 100644 --- a/live/guild/app/experiments.py +++ b/live/guild/app/experiments.py @@ -328,10 +328,15 @@ def commercial_metrics(store: Any, operation: Optional[str] = None, # about the price the payer was actually shown. if not _at_or_after(e, since): continue - if (tested_price_credits is not None - and e.get("price_credits") is not None - and e.get("price_credits") != tested_price_credits): - 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"): @@ -566,10 +571,24 @@ def seed_defaults(store: Any) -> dict[str, Any]: if os.environ.get(pricing._env_key(operation)) is not None: skipped.append({"key": key, "reason": "price_pinned_by_operator"}) continue - define(store, key, hypothesis=spec["hypothesis"], - variable=spec["variable"], - baseline=commercial_metrics(store, operation), - tested_price_credits=pricing.price(operation)) + # 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, " diff --git a/live/guild/tests/test_paid_impressions_transport.py b/live/guild/tests/test_paid_impressions_transport.py index f984676..bf4c45e 100644 --- a/live/guild/tests/test_paid_impressions_transport.py +++ b/live/guild/tests/test_paid_impressions_transport.py @@ -481,3 +481,47 @@ def test_reprice_updates_window_and_treatment_together(store): 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"]