diff --git a/docs/DIVERGENCE_2026-07-31.md b/docs/DIVERGENCE_2026-07-31.md new file mode 100644 index 0000000..f0be67f --- /dev/null +++ b/docs/DIVERGENCE_2026-07-31.md @@ -0,0 +1,121 @@ +# Divergent production state — incident record, 2026-07-31 + +Baseline: `e5b9130` / v2.0.2, deployed 2026-07-29T14:11:33Z. + +## 1. What was observed + +Two consecutive daily ops passes saw production serve two mutually +inconsistent views of the same counters: + +| surface | early reads | later reads (stable) | +|---|---|---| +| `/instrumentation` `total_events` | 13344 | 16796 | +| `/instrumentation` `genuine_external.total_events` | 222 | 242 | +| `/instrumentation` `genuine_external.passports_issued` | 1 | 3 | +| `/funnel/passports` `offer_served` | 944 (108 actors) | 1788 (226 actors) | +| `POST /ledger/checkpoint/publish` | index **14** / ledger_length **834** | index 16 / 836 | + +The `13344 / 222 / 1` triple was byte-identical on 07-30 and 07-31 — a +**frozen** view, not a lagging one. On 07-31 it survived three consecutive +reads spaced ~10s apart before flipping. + +The checkpoint case is the serious one: the **write path** was willing to act +on a view two entries behind the committed feed. + +## 2. What was tested, and what it ruled out + +| test | result | rules out | +|---|---|---| +| 40 concurrent `GET /release` | ONE `_PROCESS_STARTED_AT` (`2026-07-29T14:11:33.735141Z`) | a second origin *at observation time*; also proves the process had not restarted | +| 60 concurrent mixed-endpoint requests, body-shape checked against the requested path | 0 mismatches | cross-request response mixing (e.g. the x402 buffering middleware leaking bodies) | +| repeated reads with a unique `?cb=` per request; `cf-cache-status: DYNAMIC`, `x-render-origin-server: uvicorn` | flip still reproduced | URL-keyed CDN caching | +| code review of the read path | `/instrumentation` and `/funnel/passports` compute from `Store.events`, a plain in-process list; `Store` is a single module-level singleton in `app/state.py`; `_load()` is called only from `__init__`; the container runs one uvicorn process with no `--workers` | an in-process reload or a second `Store` inside the process | + +## 3. Root cause — what is and is not proved + +**Read side: NOT PROVED. Deliberately not "fixed".** + +Every remaining candidate — a second serving origin that the concurrency probe +never happened to hit, or an intermediary serving a body this process did not +produce — is equally consistent with the evidence, because **no response +carried anything identifying which process or which state produced it**. A +speculative state fix would have been a guess dressed as a remedy. + +**Write side: PROVED, and reproduced locally.** Independent of whatever caused +the read flip, `publish_checkpoint` had four defects on the canonical +commitment path, all reproducible with two `Store` instances over one shared +SQLite file (`tests/test_state_divergence.py`): + +1. **TOCTOU.** The authoritative `all_ledger()` / `all_checkpoints()` read + happened *before* the append, outside the write transaction. Two publishers + could compute the same next index. +2. **Silent overwrite.** `put_checkpoint` used `INSERT OR REPLACE` on + `checkpoints(idx PRIMARY KEY)`. A colliding index **replaced** a published, + third-party-pinned commitment, breaking the `prev_entry_sha256` chain with + no error. +3. **Index from `len()`.** `index = len(self.checkpoints)` re-issues an + existing index if the feed has any gap. +4. **No read-after-write.** A publish that never landed was still reported + `{"status": "published"}`. + +Separately, `SqliteBackend._commit` could drive the **thread-local** connection +depth negative: a nested transaction that raised called `_rollback` (depth→0, +rolling back the outer transaction too), then the outer `__exit__` called +`_commit` (depth→−1). From then on that thread skipped `BEGIN IMMEDIATE`, so +its writes silently ran in autocommit and `in_transaction()` lied to +`Store._save`. Connections are per-thread, so one poisoned request thread would +diverge from every other thread for the life of the process. + +## 4. What shipped + +**Decidability (so the next occurrence is not a guess)** + +- `app/instanceid.py` — random per-process `instance` id, `boot_at`, `pid`. +- Every response carries `X-Guild-Instance`, `X-Guild-Boot`, + `X-Guild-Store-Rev` (a monotonic in-memory mutation counter). +- `GET /diagnostics/state` — in-memory counts vs **authoritative SQLite** + counts, with a `divergence` list naming the exact disagreement. No paths, + tokens, hostnames or environment are exposed. +- `live/scripts/detect_divergence.py` — fans out concurrent reads and returns + one of `consistent` / `split_origin` / `stale_in_process` / + `memory_durable_split` / `intermediary`. **This replaces "discard the first + 2–3 reads"**, which was a reporting workaround with no write-path coverage. + +**Fail-closed canonical writes** + +- Authoritative read moved *inside* `BEGIN IMMEDIATE`. +- `StaleDurableStateError` when the durable feed head is behind a head this + process already observed, or the durable ledger is shorter than memory. +- `insert_checkpoint_strict` — plain `INSERT`; a duplicate index raises + `CheckpointForkError` instead of replacing history. +- Next index derived from `max(index) + 1`, not `len()`. +- Read-after-write byte comparison; `CheckpointWriteVerificationError` + otherwise. **A publish that did not land can no longer be reported as + published.** +- All three map to HTTP **409** with a stable machine-readable `code`, and the + body states the write did not happen. +- `_commit` depth clamped at zero, so a nested rollback no longer poisons the + thread. + +## 5. Honest statement of the evidence + +**No durable loss was detected.** Every cumulative counter on the warm branch +was greater than or equal to the previous snapshot, and the ledger head +(index 16 / length 836) was stable across five reads and unchanged from 07-30. + +That is *not* the same as proving individual event continuity: there is no +per-event durable sequence audit, so the correct phrasing is **"no durable loss +detected"**, never "no data was lost". + +## 6. Remaining risk + +- The read-side cause is still **unknown**. It is now instrumented, not + resolved. If it recurs, `detect_divergence.py` names the class. +- If the verdict comes back `split_origin`, that is a **topology emergency**: + SQLite lives on a single-mount Render disk and the application guard can only + see worker processes inside its own container, never a second instance. The + response is a Postgres migration review, not a code patch. +- The fail-closed publish trades availability for integrity: under genuine + divergence the checkpoint feed will **stop advancing** and return 409 rather + than publish. That is the intended trade — a gap in the feed is recoverable, + a fork is not. diff --git a/docs/EXPERIMENT_PREFLIGHT_2026-07-31.md b/docs/EXPERIMENT_PREFLIGHT_2026-07-31.md new file mode 100644 index 0000000..00aebc1 --- /dev/null +++ b/docs/EXPERIMENT_PREFLIGHT_2026-07-31.md @@ -0,0 +1,139 @@ +# Experiment: delegation preflight — 2026-07-31 + +## The blunt finding first + +**The current pricing model is aimed at a payer who has no money.** + +Agent Guild's funnel sells a free credential to autonomous agents on the theory +that they will later pay for trust reads. Measured today: + +- Total x402 settled volume, **all networks, July 2026: $232,329** — down 98.9% + from the $20.5M November 2025 peak, on flat transaction count (~195k/day at + ~$0.04 each: automated loop traffic, not commerce). +- Median x402 Bazaar listing: **2 calls and 1 unique payer per 30 days**. +- Median earning agent: **$1.65 per 30 days**; Gini 0.97 across 316 agents. +- Virtuals Protocol: 44,051 registered agents, **69 active in July (0.16%)**. +- Olas: 13.97M lifetime agent-to-agent transactions → **$106,941 lifetime + turnover → $458 of protocol fees, ever**. + +5% of the entire global machine-payments market is **$11.6k/month**. Agents +earning $1.65/month cannot buy anything. Any roadmap that monetises the supply +side of today's machine economy is arithmetically dead, and no amount of funnel +optimisation fixes it. The passport is still the right acquisition instrument — +it is not a revenue instrument, and we should stop implying it will become one. + +## What we are NOT short of + +Reach. `offer_served` went 944 → 1,790 in 24 hours. All of it crawlers. The +constraint has never been distribution volume; it is that the offer asks an +agent to invest effort now for value later, and the agents receiving it are +registry bots with no delegation to make. + +## The gap with the strongest evidence + +The one place where a trust decision is unavoidable, and nobody serves it: + +| Claim in a listing | Reality when probed | Source | +|---|---|---| +| 170 of 183 a2aregistry agents report `is_healthy: true` (92.9%) | **62 (33.9%) complete an A2A task** | a2aregistry API + task probe | +| 3,913 agents serve a valid Agent Card | **42 (0.8%) sign it** | Agenstry conformance sweep | +| 2,459 agents self-label "paid" | **141 (5.7%) return a 402** | x402/Bazaar probe | + +**114 agents are green and broken at the same time.** The A2A discovery +specification states in writing that it prescribes no registry API, and +contains no mention of signatures. x402 `exact` is a push payment: irreversible +once executed, `payTo` bound to no legal entity, documented remedy "the seller +sends it back". Escrow and reputation are both listed as future work. + +Every existing signal grades a **repository or a static card, once, at +publication time** (Glama's grade is 70% tool-description quality; Docker +scores the image; Anthropic reviews the submission). Nobody attests to the +**running endpoint at call time** — which is exactly where the rug pull lives. + +## Ranked experiments + +| | Qualified reach | Friction | Time to money | Defensibility | Measurable | +|---|---|---|---|---|---| +| **E1 Delegation preflight** → directories, orchestrators, delegating agents | High — machine-reachable, no human in the loop | **Lowest**: one unauthenticated GET | Medium — real buyers, unproven price | High — we already have signed decisions, checkpoints, evidence classes | High | +| E2 Runtime attestation for MCP servers → enterprise platform teams | High, **proven budget** (UpGuard $1,750/mo per 50 vendors; three security acquisitions in 12 months) | High — human sales | **Fastest to real money** | Medium | High | +| E3 Verified behavioural evidence → AI underwriters | Very narrow (Armilla, Munich Re) | High — human, contractual | Slow | High | Medium | + +**Chose E1**, because it is the only one executable this session without +spending money, contacting anyone, or launching paid — and because it is +reversible: it is one additive read-only endpoint. + +**E2 is the strongest money, and we should say so plainly.** Its payer is a +human enterprise security team, which is a direct conflict with the +machine-only clause of the constitution. That conflict is now a decision for +Ross, not something to be quietly resolved by preferring the weaker option. + +## What shipped + +`GET /preflight?url=…` and the `guild_preflight` MCP tool. Free, no key, no +registration. Six checks, run live: + +1. `endpoint_reachable` +2. `protocol_handshake` — a real A2A/MCP handshake, **not** merely HTTP 200 +3. `agent_card_resolves` +4. `agent_card_signed` — presence, explicitly *not* claimed as verification +5. `payment_claim_holds` — does an advertised paid endpoint actually 402 +6. `independent_evidence` — does the Guild hold attestation history + +Honesty rules enforced by tests: unknowns are reported and **excluded from the +verdict, never averaged in**; a clean verdict still publishes its unknown +count; absence of evidence is stated as absence, not as risk; SSRF-screened so +it can never be used as an internal port scanner. + +### Two of our own defects found while building it + +Both were **understating** reality — the same error class as overstating +adoption, pointed the other way: + +- **Chunked transfer-encoding was never decoded.** The raw body begins with a + hex chunk-length line, so every downstream JSON check failed. Since the card + check is what promotes an endpoint from `http_responsive` to + `recently_reachable`, *any* agent served over chunked encoding was recorded + as unproven. This is why `verified_reachable` read **0 for every entry in the + demand feed** — not because nobody was reachable, but because we could not + read them. +- **A large but valid card was read as no card at all.** The probe read is + bounded, so verbose cards arrive truncated and fail to parse. Our own card is + one of them: Agent Guild's endpoint failed its own preflight until this was + fixed. + +## Instrumentation, baseline and thresholds + +Every call records a `preflight_run` event with target, verdict, failed and +unknown counts, and rides the qualified cohort funnel from Phase A. + +**Live baseline at ship time (2026-07-31, `4c28ab8`):** + +- adoption-grade external passport holders: **0** (the previously reported "3" + were third-party fetches, one of them a schema probe) +- verified external revenue: **$0.00** +- qualified external actors in the passport cohort: **1** +- preflight runs: **0** (endpoint did not exist) + +**Success — escalate and consider pricing:** +- ≥ 25 preflight runs from ≥ 10 distinct genuine-external actors in 14 days, **and** +- ≥ 3 actors returning on a later day (the retention signal, not the volume one). + +**Kill — stop and reallocate to E2:** +- < 5 genuine-external runs in 14 days, or +- > 90% of runs still unattributable crawler traffic at day 14. + +Both thresholds are deliberately small. The point is to learn whether *anyone +doing a real delegation* wants this, not to accumulate impressions. + +## Pricing — NOT shipped, needs approval + +The evidence says the caller who benefits is the party whose reputation depends +on the listing working (a directory, an orchestrator, a delegating agent), not +the agent being checked. Comparable anchors: Riskified earns **0.237% of +screened GMV** for a guarantee-backed signal; UpGuard charges **$1,750/mo for +50 vendors**; Vouched KYA charges **$20–$325/mo for 1k–25k delegation checks**. + +No price is live and no payment configuration was touched. The proposed action +for approval is a single change: keep preflight free to a per-caller daily cap, +then meter it at the existing x402 price. **That is a pricing change and is not +being made without an explicit yes.** diff --git a/docs/INTERFACE.md b/docs/INTERFACE.md index f0aab95..31a80af 100644 --- a/docs/INTERFACE.md +++ b/docs/INTERFACE.md @@ -64,6 +64,7 @@ guild_mediated requires two-party cryptographic participation, a Guild-observed - `POST /credentials/verify` - `GET /demand/feed` - `POST /demand/watch` +- `GET /diagnostics/state` - `GET /disputes/{case_id}` - `POST /disputes/{case_id}/appeal` - `POST /disputes/{case_id}/vote` @@ -97,6 +98,7 @@ guild_mediated requires two-party cryptographic participation, a Guild-observed - `GET /offers/{offer_id}` - `POST /offers/{offer_id}/accept` - `POST /outcomes` +- `GET /preflight` - `POST /providers/external/discover` - `GET /referrals` - `GET /release` @@ -142,6 +144,7 @@ guild_mediated requires two-party cryptographic participation, a Guild-observed - `guild_escrow_open` - `guild_escrow_release` - `guild_passport` +- `guild_preflight` - `guild_prove` - `guild_prove_verify` - `guild_record` diff --git a/live/guild/app/instanceid.py b/live/guild/app/instanceid.py new file mode 100644 index 0000000..2d2e85b --- /dev/null +++ b/live/guild/app/instanceid.py @@ -0,0 +1,55 @@ +"""Process identity — the decidability layer for state divergence. + +WHY THIS EXISTS (divergence incident 2026-07-30/31) +--------------------------------------------------- +Production served two mutually inconsistent views of the same counters within +minutes: ``/instrumentation`` returned a frozen older snapshot on several +consecutive reads before flipping to the current one, and a ``POST +/ledger/checkpoint/publish`` returned checkpoint index 14 / ledger_length 834 +while the published feed was already at 16 / 836. + +The investigation could NOT decide between the candidate causes — a second +serving origin, an intermediary serving a stale body, or an in-process stale +durable read — because NO RESPONSE CARRIED ANYTHING THAT IDENTIFIED WHICH +PROCESS OR WHICH STATE PRODUCED IT. Every candidate explanation was equally +consistent with the evidence. That is a diagnosability defect, and it is fixed +here rather than guessed around. + +WHAT IS STAMPED (all non-secret, all safe to publish) + * ``instance`` — random per-PROCESS id, minted at import. Two different + values observed for one release SHA PROVE more than one serving process. + One value across a divergent pair DISPROVES the split-origin theory and + points at an intermediary or an in-process stale read. + * ``boot_at`` — process start (UTC). Distinguishes "restarted" from + "second instance" when ids differ. + * ``pid`` — process id inside the container. Distinguishes forked + workers that share a boot timestamp. + * ``store_rev`` — monotonic in-memory mutation counter (``Store.revision``). + A response whose ``store_rev`` is LOWER than one already observed from the + same ``instance`` is a stale in-process view; across instances it is a + split-brain read. Either way it becomes DETECTABLE FROM OUTSIDE. + +It deliberately leaks no paths, tokens, environment or hostnames — the id is +random, not derived from anything sensitive, so a third party cannot correlate +it back to infrastructure. +""" +from __future__ import annotations + +import os +import secrets +from datetime import datetime, timezone + +#: Random per-process identity. Minted once at import; never persisted, never +#: derived from a hostname/path/secret. +INSTANCE_ID: str = secrets.token_hex(6) + +#: Process start time (UTC, ISO-8601). +BOOT_AT: str = datetime.now(timezone.utc).isoformat() + +#: OS process id — separates forked workers that share a boot timestamp. +PID: int = os.getpid() + + +def identity() -> dict[str, object]: + """The non-secret process identity block embedded in diagnostics.""" + return {"instance": INSTANCE_ID, "boot_at": BOOT_AT, "pid": PID} diff --git a/live/guild/app/main.py b/live/guild/app/main.py index 1264e7d..11eed43 100644 --- a/live/guild/app/main.py +++ b/live/guild/app/main.py @@ -14,6 +14,7 @@ import json import os +import uuid import contextvars from typing import Any, Optional @@ -40,7 +41,10 @@ from . import __version__ from . import billing from .billing import InsufficientCredits, UnknownAccount, PRICING, CREDIT_USD +from . import instanceid +from . import preflight from .state import store +from .store import CanonicalWriteRefused from .reachability import url_policy_check from . import abuse from . import crypto @@ -252,17 +256,69 @@ async def _capture_ua(request: Request, call_next): hdrs.pop("content-length", None) hdrs.pop(x402.PAYMENT_RESPONSE_HEADER.lower(), None) hdrs[x402.PAYMENT_RESPONSE_HEADER] = fin["header"] - return Response(content=body, status_code=response.status_code, - headers=hdrs, media_type=response.media_type) + out = Response(content=body, status_code=response.status_code, + headers=hdrs, media_type=response.media_type) + _stamp_view_identity(out) + return out except Exception: _log.exception("x402 receipt finalize failed — serving the paid " "result with the provisional PAYMENT-RESPONSE") - return Response(content=body, status_code=response.status_code, - headers=dict(response.headers), - media_type=response.media_type) + out = Response(content=body, status_code=response.status_code, + headers=dict(response.headers), + media_type=response.media_type) + _stamp_view_identity(out) + return out + _stamp_view_identity(response) return response +def _stamp_view_identity(response) -> None: + """Stamp WHICH process and WHICH state served this response. + + Divergence incident 2026-07-31: production returned two mutually + inconsistent views of the same counters and nothing in either response + could tell them apart, so the split-origin, intermediary-cache and + stale-in-process-read theories were all equally consistent with the + evidence. These three non-secret headers make the next occurrence + DECIDABLE from outside, with no shell on the box: + + X-Guild-Instance random per-process id — two values for one release + SHA prove two serving processes + X-Guild-Boot process start time — restart vs second instance + X-Guild-Store-Rev monotonic mutation ctr — a LOWER value than one + already seen from the same instance is a stale view + + Never raises: observability must not be able to fail a request.""" + try: + response.headers["X-Guild-Instance"] = instanceid.INSTANCE_ID + response.headers["X-Guild-Boot"] = instanceid.BOOT_AT + response.headers["X-Guild-Store-Rev"] = str(store.revision) + except Exception: # noqa: BLE001 + pass + + +@app.exception_handler(CanonicalWriteRefused) +async def _canonical_write_refused_handler(request: Request, + exc: CanonicalWriteRefused): + """A canonical commitment this process REFUSED to make (409). + + Checkpoints are pinned by third parties and cited by passports, so a + publish built on durable state that cannot be trusted must fail loudly + rather than commit and be reconciled later. The body carries a stable + machine-readable `code` plus the view identity, so an operator (or the + scheduled ops pass) can correlate the refusal with the exact process and + revision that refused.""" + return JSONResponse(status_code=409, content={ + "error": "canonical_write_refused", + "code": getattr(exc, "code", "canonical_write_refused"), + "detail": str(exc), + "view": {"instance": instanceid.INSTANCE_ID, + "store_rev": store.revision}, + "note": ("the write did NOT happen. Re-read the authoritative feed " + "(/ledger/checkpoints, /diagnostics/state) before retrying."), + }) + + @app.exception_handler(PaymentIdConflict) async def _payment_id_conflict_handler(request: Request, exc: PaymentIdConflict): @@ -1484,12 +1540,28 @@ def get_passport(agent_id: str, request: Request, response: Response): at the embedded `/credentials/verify`. Free: an agent showing its passport is the Guild's distribution loop.""" base = str(request.base_url).rstrip("/") + # Correlate the attempt with its outcome, and stamp WHO fetched it (the + # caller's own api key when presented) so a subject claiming its own + # passport is distinguishable from a third party fetching it. A MISS is + # recorded as a failure, never as an issuance (corrective pass 2026-07-31). + request_id = uuid.uuid4().hex[:16] + ua = _ua.get() + caller_key = request.headers.get("x-api-key") or None + actor = creds.sanitize_actor_key(caller_key) if caller_key else None + store.record_event(actor, "passport_requested", ua=ua, + endpoint="passport", subject_id=agent_id, + transport="http", request_id=request_id) cred = store.issue_passport( agent_id, verify_url=f"{base}/credentials/verify", explore_url=f"{base}/agents/{agent_id}/reputation", + actor_key=actor, surface="http", ua=ua, request_id=request_id, ) if cred is None: + store.record_event(actor, "passport_issue_failed", ua=ua, + endpoint="passport", subject_id=agent_id, + transport="http", request_id=request_id, + reason="unknown_agent_or_no_reputation") raise HTTPException(404, "agent not found or no reputation computed") # The response body IS the signed credential — adding keys to it would # break offline signature verification. Journey guidance rides the headers. @@ -2733,6 +2805,18 @@ def llms_txt(): " 3. GET /agents/{id}/passport → a Guild-signed Verifiable Credential;\n" " any party verifies it offline (POST /credentials/verify, Guild did at\n" " /.well-known/agent-guild-did.json). Live badge: GET /agents/{id}/badge.svg.\n\n" + "## About to delegate to an agent you did not write? Check it first (free)\n" + "GET /preflight?url= (or the guild_preflight MCP tool)\n" + "One unauthenticated call, live at request time, separating what an\n" + "endpoint CLAIMS from what it just PROVED: does it complete a real\n" + "protocol handshake (not merely answer 200), does its Agent Card resolve,\n" + "is that card signed, and does its payment claim actually challenge with\n" + "a 402. Measured across the live ecosystem on 2026-07-31: 92.9% of\n" + "registry-listed agents report healthy but 33.9% complete a task, 0.8%\n" + "sign their card, and 5.7% of self-declared paid agents actually return\n" + "402. x402 `exact` transfers are irreversible, so the check belongs\n" + "BEFORE the payment. Checks we could not perform come back as `unknowns`\n" + "and are excluded from the verdict, never averaged into it.\n\n" "## You are an AI agent? You can use this with no human.\n" "Hosted (no install), free writes, self-serve credits (POST /billing/trial),\n" "and a did:key identity you control. Nothing here needs a human or an account\n" @@ -2828,6 +2912,50 @@ def llms_txt(): ) +@app.get("/preflight") +def delegation_preflight(request: Request, url: str = Query( + ..., description="The agent endpoint you are about to delegate to")): + """PREFLIGHT — run this in the moment before you delegate or pay. + + Separates what an endpoint CLAIMS from what it just PROVED, live, at + request time. Free, no key, no registration, one call. + + It exists because listing metadata is not evidence. Measured across the + live ecosystem on 2026-07-31: 92.9% of registry agents report healthy and + 33.9% actually complete a task; 0.8% sign their Agent Card; 5.7% of + self-declared paid agents actually return a 402. x402 `exact` payments are + irreversible, so the check has to happen BEFORE the transfer, not after. + + Unlike a badge or a repository score, this is not cached and not computed + at publication time — a server can change its tool descriptions after any + one-off review. `unknowns` are reported, never averaged into the verdict. + """ + out = preflight.run(url, store=store) + # Demand instrumentation: WHO is asking, and do they come back? This is + # the only honest way to learn whether the check is wanted, and it is + # recorded as a query, never as adoption of anything. + store.record_event(None, "preflight_run", ua=_ua.get(), + endpoint="preflight", target=url, + verdict=out["verdict"], + failed_count=len(out["failed"]), + unknown_count=len(out["unknowns"])) + return out + + +@app.get("/diagnostics/state") +def diagnostics_state(): + """WHICH process and WHICH state produced this response — the decidability + endpoint for the 2026-07-31 divergence incident. + + Free, non-secret and side-effect-free. Compare `instance` across a pair of + responses that disagree: two values PROVE more than one serving process; + one value disproves the split-origin theory. `in_memory` vs `durable` shows + whether the view this process serves reads from agrees with the database it + writes to, and `divergence` names the exact disagreement. No paths, tokens, + hostnames or environment are exposed.""" + return store.state_diagnostics() + + @app.get("/instrumentation") def get_instrumentation(): """The agent-native adoption funnel, split into `external` (third-party diff --git a/live/guild/app/mcp_server.py b/live/guild/app/mcp_server.py index de442d3..bc7e2d4 100644 --- a/live/guild/app/mcp_server.py +++ b/live/guild/app/mcp_server.py @@ -16,6 +16,7 @@ import contextvars import json as _json import os +import uuid from typing import Any, Callable, Optional from typing_extensions import TypedDict @@ -35,6 +36,7 @@ from . import proving from . import x402 from .payments import CachedPaidResult, PaidRequest, PaymentChallenge, PaymentIdConflict +from . import preflight from .state import store from . import credentials as _creds @@ -423,6 +425,34 @@ def _serve_paid(preq: PaidRequest, produce: Callable[[], Any], return result +@mcp.tool +def guild_preflight(url: str, ctx: Context = None) -> dict: + """Run this in the moment BEFORE you delegate to, or pay, an agent endpoint + you did not write. Free, no key, one call. + + Separates what the endpoint CLAIMS from what it just PROVED, live. Measured + across the live ecosystem on 2026-07-31: 92.9% of registry-listed agents + report healthy but only 33.9% actually complete a task; 0.8% sign their + Agent Card; and of agents advertising payment, 5.7% actually return a 402. + x402 `exact` transfers are irreversible, so this has to happen before the + payment, not after it. + + Unlike a directory badge this is not cached and not derived from a + repository at publication time — a server can change its tool descriptions + after any one-off review. Checks it could not perform are returned as + `unknowns` and are excluded from the verdict rather than averaged into it. + + Example: guild_preflight(url="https://some-agent.example/a2a") + """ + out = preflight.run(url, store=store) + store.record_event("mcp", "preflight_run", ua=_client_ua(ctx), + endpoint="preflight", target=url, + transport="mcp", verdict=out["verdict"], + failed_count=len(out["failed"]), + unknown_count=len(out["unknowns"])) + return out + + @mcp.tool def guild_check(capability: str, api_key: str = "", ctx: Context = None) -> dict: """START HERE. One call to vet a `capability` before you delegate: returns the @@ -762,10 +792,27 @@ def guild_passport(agent_id: str, ctx: Context = None) -> dict: Example: guild_passport(agent_id="agt_9x"). Returns a W3C VC, or {error}. """ - store.record_event("mcp", "passport_issued", ua=_client_ua(ctx), - endpoint="passport", subject_id=agent_id) - cred = store.issue_passport(agent_id) - return cred if cred is not None else {"error": "agent not found or no reputation"} + # TELEMETRY (corrective pass 2026-07-31). This used to record + # `passport_issued` HERE, on entry, and `Store.issue_passport` recorded a + # SECOND one on success — so one MCP call counted as two issuances, and a + # miss (unknown agent) counted as one. A schema probe exploiting exactly + # that on 2026-07-30 tripled the headline number with no agent behind it. + # Attempt and failure are now their own event types; issuance is emitted + # once, inside issue_passport, only when a credential actually exists. + request_id = uuid.uuid4().hex[:16] + ua = _client_ua(ctx) + store.record_event("mcp", "passport_requested", ua=ua, + endpoint="passport", subject_id=agent_id, + transport="mcp", request_id=request_id) + cred = store.issue_passport(agent_id, actor_key="mcp", surface="mcp", + ua=ua, request_id=request_id) + if cred is None: + store.record_event("mcp", "passport_issue_failed", ua=ua, + endpoint="passport", subject_id=agent_id, + transport="mcp", request_id=request_id, + reason="unknown_agent_or_no_reputation") + return {"error": "agent not found or no reputation"} + return cred @mcp.tool diff --git a/live/guild/app/models.py b/live/guild/app/models.py index 19e0112..d38a402 100644 --- a/live/guild/app/models.py +++ b/live/guild/app/models.py @@ -413,24 +413,43 @@ class ReferralsResponse(BaseModel): # --- self-evaluation (Outcome 4: continuous self-assessment) ---------------- class HealthSnapshot(BaseModel): + """The Guild's self-assessment. + + HONESTY INVARIANTS (corrective pass 2026-07-31) — the field names carry the + caveat so a number cannot be quoted without it: + * utility is the PRODUCTION-only lift, with its sample size attached; the + seeded bootstrap figure travels under an explicitly non-production key; + * revenue is ONLY independently confirmed external mainnet settlement; + sandbox credits are an internal unit and are labelled NOT_MONEY; + * growth is adoption-grade external activity, not "records lacking + first_party". + """ + at: str - # utility — is the Guild actually helping agents? - measured_lift: Optional[float] = None - # provenance of measured_lift so the number never travels unlabelled: - # "bootstrap" (seeded demonstration) | "production" | "mixed" | "empty". + # --- utility (production only; bootstrap is separate and labelled) ------ + production_measured_lift: Optional[float] = None + production_n_recommended: int = 0 + production_lift_measurable: bool = False + mixed_bootstrap_lift_NOT_PRODUCTION: Optional[float] = None measured_lift_dataset: Optional[str] = None recommended_success_rate: Optional[float] = None - # growth — are new (external) agents arriving? + # --- growth (adoption-grade) ------------------------------------------- + adoption_grade_external_self_claims: int = 0 agents_total: int agents_external: int external_querying_agents: int = 0 # retention — do external agents come back? external_repeat_query_agents: int external_repeat_paid_agents: int - # revenue capture — is value being paid for? + # --- economic value ---------------------------------------------------- + # Real money: independently confirmed external mainnet settlement only. + verified_external_revenue_usd: float = 0.0 + cryptographically_bound_machine_revenue_usd: float = 0.0 + # Internal accounting units. NOT money, never summed with the above. external_paid_queries: int + sandbox_credits_spent_external_NOT_MONEY: int = 0 credits_spent_external: int - revenue_usd_external: float + legacy_sandbox_credit_notional_usd_NOT_REVENUE: float = 0.0 # referrals — are agents recruiting agents? total_referrals: int activated_referrals: int diff --git a/live/guild/app/preflight.py b/live/guild/app/preflight.py new file mode 100644 index 0000000..34f2344 --- /dev/null +++ b/live/guild/app/preflight.py @@ -0,0 +1,262 @@ +"""Delegation preflight — the check nobody performs, immediately before delegating. + +WHY THIS AND NOT MORE PASSPORT SURFACE +-------------------------------------- +Measured on 2026-07-31 across the live agent ecosystem: + + * a2aregistry lists 183 agents; 170 (92.9%) report `is_healthy: true`; only + 62 (33.9%) actually complete an A2A task when probed. **114 agents are + green and broken at the same time.** + * Of 3,913 agents with a valid Agent Card, 42 (0.8%) sign it. The A2A + discovery specification itself states it prescribes no registry API, and + contains no mention of signatures. + * Of 2,459 agents self-labelled "paid", 141 (5.7%) actually answer with an + HTTP 402 challenge. The other 94.3% is an unverified string in a card. + * x402 `exact` is a PUSH payment: irreversible once executed, `payTo` is a + bare address bound to no legal entity, and the documented remedy for a + bad outcome is "the seller sends the money back". Escrow and reputation + are both listed as future work. + +So the risk sits in a specific place: the moment a caller is about to hand +work — and money — to an endpoint whose only assurances are self-declared. +Uptime monitoring answers the wrong question ("did something respond?"), and +every existing scorer grades a REPOSITORY or a static card, once, at +publication time. + +WHAT THIS RETURNS + A single unauthenticated call that separates CLAIMED from PROVEN for one + endpoint, and says plainly which checks it could not perform. It never + invents a score out of unknowns: an unknown is reported as unknown and is + excluded from the verdict rather than being averaged into it. + +WHAT IT IS NOT + It is not a safety guarantee, not an endorsement, and not a substitute for + the caller's own policy. It reports evidence and names its own limits, which + is the entire difference between this and a green tick that means nothing. +""" +from __future__ import annotations + +import json +import re +import socket +import ssl +from typing import Any, Optional +from urllib.parse import urlsplit + +from . import reachability + +#: Checks that, when they FAIL, are strong evidence against delegating. +BLOCKING = ("endpoint_reachable", "protocol_handshake") + +#: Card fields that assert the endpoint takes money. +_PAID_MARKERS = ("x402", "payment", "paid", "price", "usdc", "402") + + +def _probe_get(url: str, path: str, timeout_ctx=None + ) -> tuple[Optional[int], bytes, str]: + """SSRF-safe GET reusing the reachability module's pinned-address path. + + Same protections as ``liveness_probe``: URL policy check, DNS resolution + with private/link-local/loopback screening, and a connection pinned to the + screened address so a rebind between resolve and connect cannot redirect + us. Never raises.""" + ok, reason = reachability.url_policy_check(url) + if not ok: + return None, b"", f"policy: {reason}" + parts = urlsplit(url) + host = parts.hostname + if not host: + return None, b"", "no host" + port = parts.port or (443 if parts.scheme == "https" else 80) + ok, addrs, reason = reachability._resolve_and_screen(host, port) + if not ok: + return None, b"", reason + family, addr = addrs[0] + try: + code, body = reachability._http_request_pinned( + parts.scheme, host, family, addr, port, path, "GET", None, "", None) + return code, (body or b""), "" + except (ssl.SSLError, socket.error, OSError) as e: + return None, b"", f"{type(e).__name__}" + except Exception as e: # noqa: BLE001 — a preflight must never 500 + return None, b"", f"{type(e).__name__}" + + +def _check(name: str, status: str, detail: str, **extra) -> dict[str, Any]: + """status is one of: proven | failed | unknown.""" + return {"check": name, "status": status, "detail": detail, **extra} + + +def _card_is_signed(card: dict[str, Any]) -> tuple[bool, str]: + """Is the Agent Card cryptographically signed? + + A2A cards carry signatures under `signatures` (JWS) in the spec drafts; + Guild/DID-style cards may carry `proof`. We accept either, and we DO NOT + verify the signature here — we report only that one is present, which is + an honest, checkable statement. Claiming verification we did not perform + would be exactly the failure this endpoint exists to expose.""" + if isinstance(card.get("signatures"), list) and card["signatures"]: + return True, "JWS `signatures` present (presence only — not verified here)" + if isinstance(card.get("proof"), dict) and card["proof"]: + return True, "`proof` present (presence only — not verified here)" + return False, "no `signatures` or `proof` on the card" + + +def _claims_payment(card: dict[str, Any]) -> bool: + blob = json.dumps(card).lower() if card else "" + return any(m in blob for m in _PAID_MARKERS) + + +def run(url: str, *, store=None) -> dict[str, Any]: + """Run the preflight for one endpoint URL. Never raises.""" + checks: list[dict[str, Any]] = [] + + # --- 1. reachability + REAL protocol handshake ------------------------ + rec = reachability.liveness_probe(url) + status = rec.get("status") + evidence = rec.get("evidence_level") + if status == "recently_reachable" and evidence == "protocol_handshake": + checks.append(_check("endpoint_reachable", "proven", + rec.get("detail") or "responded")) + checks.append(_check( + "protocol_handshake", "proven", + "completed a real A2A/MCP handshake, not merely an HTTP 200")) + elif status == "http_responsive": + checks.append(_check("endpoint_reachable", "proven", + "something answered over HTTP")) + checks.append(_check( + "protocol_handshake", "failed", + "a server answered but proved NO agent protocol. This is the " + "single most common failure mode measured in the wild: 92.9% of " + "listed agents report healthy, 33.9% actually complete a task.")) + else: + checks.append(_check("endpoint_reachable", "failed", + rec.get("detail") or "no response")) + checks.append(_check("protocol_handshake", "unknown", + "not attempted — the endpoint did not answer")) + + # --- 2. agent card: resolvable, and SIGNED? --------------------------- + card: dict[str, Any] = {} + truncated = False + code, body, err = _probe_get(url, "/.well-known/agent-card.json") + if code and 200 <= code < 300 and body: + try: + card = json.loads(body.decode("utf-8", "replace")) + except (ValueError, UnicodeDecodeError): + # The probe caps its read, so a LARGE but perfectly valid card + # arrives truncated and will not parse. Treating that as "no card" + # would report a false failure against a well-formed agent — the + # precise error class this endpoint exists to eliminate. A + # truncated card is reported as RESOLVING, and every check that + # needs the whole document degrades to `unknown`, never to + # `failed`. + text = body.decode("utf-8", "replace").lstrip() + if text.startswith("{") and not text.rstrip().endswith("}"): + truncated = True + if truncated: + checks.append(_check( + "agent_card_resolves", "proven", + f"served at /.well-known/agent-card.json ({code}), but larger " + "than the probe read cap — inspected only in part")) + checks.append(_check( + "agent_card_signed", "unknown", + "the card exceeded the bounded probe read, so the absence of a " + "signature cannot be asserted (reported as unknown, NOT as a " + "failure)")) + checks.append(_check( + "payment_claim_holds", "unknown", + "not tested — the card could not be read in full")) + elif card: + checks.append(_check("agent_card_resolves", "proven", + f"served at /.well-known/agent-card.json ({code})", + declared_name=str(card.get("name") or "")[:120])) + signed, why = _card_is_signed(card) + checks.append(_check("agent_card_signed", + "proven" if signed else "failed", why)) + else: + checks.append(_check("agent_card_resolves", "failed", + err or f"no parsable card (http {code})")) + checks.append(_check("agent_card_signed", "unknown", + "not attempted — no card to inspect")) + + # --- 3. does a payment CLAIM actually hold? --------------------------- + if truncated: + pass # already reported as unknown above + elif card and _claims_payment(card): + pcode, _pbody, perr = _probe_get(url, "/") + if pcode == 402: + checks.append(_check("payment_claim_holds", "proven", + "returned a 402 payment challenge as claimed")) + elif pcode is None: + checks.append(_check("payment_claim_holds", "unknown", + perr or "could not probe the paid surface")) + else: + checks.append(_check( + "payment_claim_holds", "failed", + f"card advertises payment but the endpoint answered {pcode}, " + "not 402. Measured in the wild: only 5.7% of self-declared " + "paid agents actually challenge.")) + else: + checks.append(_check("payment_claim_holds", "unknown", + "the card makes no payment claim to test")) + + # --- 4. independent evidence the Guild already holds ------------------ + known = None + if store is not None and card: + did = str(card.get("did") or (card.get("provider") or {}).get("did") or "") + try: + known = store.agent_by_did(did) if did else None + except Exception: # noqa: BLE001 + known = None + if known: + checks.append(_check( + "independent_evidence", "proven", + "this endpoint's DID is a registered Guild agent with an " + "evidence history you can audit", + agent_id=known.get("id"), + attestations_received=known.get("attestations_received", 0))) + else: + checks.append(_check( + "independent_evidence", "unknown", + "no independent evidence — the Guild holds no attestation history " + "for this endpoint. Absence of evidence is NOT evidence of risk; " + "it means you are relying entirely on self-declaration.")) + + # --- verdict ---------------------------------------------------------- + by_name = {c["check"]: c for c in checks} + failed_blocking = [n for n in BLOCKING if by_name[n]["status"] == "failed"] + failed_other = [c["check"] for c in checks + if c["status"] == "failed" and c["check"] not in BLOCKING] + unknown = [c["check"] for c in checks if c["status"] == "unknown"] + + if failed_blocking: + verdict, headline = "do_not_delegate", ( + "This endpoint did not prove it can do the thing it is listed for.") + elif failed_other: + verdict, headline = "delegate_with_caution", ( + "It works, but at least one of its own claims does not hold.") + else: + verdict, headline = "no_failed_checks", ( + "Every check we could perform passed. That is not an endorsement — " + "see `unknowns`.") + + return { + "target": url, + "verdict": verdict, + "headline": headline, + "checks": checks, + "failed": failed_blocking + failed_other, + "unknowns": unknown, + "scored": [c["check"] for c in checks if c["status"] != "unknown"], + "method": ("live probe at request time — NOT a cached badge and NOT a " + "score computed from a repository at publication time. A " + "server can change its behaviour after any one-off review; " + "this is the state we observed just now."), + "limits": ( + "We report presence of a card signature, not its validity. We do " + "not execute a paid task. We cannot see what the endpoint does " + "with your data. `unknowns` are excluded from the verdict rather " + "than averaged into it, so a clean result over four unknowns is " + "not the same as a clean result over eight checks — the counts " + "are given so you can tell the difference."), + } diff --git a/live/guild/app/reachability.py b/live/guild/app/reachability.py index af63499..7e7b373 100644 --- a/live/guild/app/reachability.py +++ b/live/guild/app/reachability.py @@ -200,6 +200,35 @@ def _connect_pinned(scheme: str, host: str, family: int, addr: str, port: int, return raw +def _dechunk(raw: bytes) -> bytes: + """Decode an HTTP/1.1 chunked body, tolerating truncation. + + The probe caps its read at PROBE_MAX_BYTES, so the last chunk is routinely + incomplete and the terminating 0-chunk is usually never seen. That is + fine: we return everything decoded so far, because the callers only need + enough of the body to recognise a card or a handshake. Returns the input + unchanged if it does not look chunked, so a mislabelled response degrades + to the previous behaviour rather than to an empty body.""" + out = bytearray() + pos = 0 + n = len(raw) + while pos < n: + eol = raw.find(b"\r\n", pos) + if eol == -1: + break + size_line = raw[pos:eol].split(b";", 1)[0].strip() + try: + size = int(size_line, 16) + except ValueError: + return raw if not out else bytes(out) + if size == 0: + break + start = eol + 2 + out += raw[start:start + size] + pos = start + size + 2 # skip the chunk's trailing CRLF + return bytes(out) if out else raw + + def _http_request_pinned(scheme: str, host: str, family: int, addr: str, port: int, path: str, method: str = "HEAD", body: Optional[bytes] = None, @@ -234,7 +263,22 @@ def _http_request_pinned(scheme: str, host: str, family: int, addr: str, code = int(bits[1]) except ValueError: code = None - body_prefix = buf.split(b"\r\n\r\n", 1)[1] if b"\r\n\r\n" in buf else b"" + head, _, body_prefix = buf.partition(b"\r\n\r\n") + if b"\r\n\r\n" not in buf: + body_prefix = b"" + # CHUNKED TRANSFER-ENCODING (defect found 2026-07-31). The raw body of + # a chunked response begins with a hex chunk LENGTH line, so every + # JSON check downstream failed to parse it. Because the A2A card check + # (_looks_like_a2a_card) is what promotes an endpoint from + # "http_responsive" to "recently_reachable / protocol_handshake", ANY + # agent served over chunked encoding — which is the default for most + # streaming frameworks, including our own — was being classified as + # unproven. That is why `verified_reachable` read 0 for every entry in + # the demand feed: not because nobody was reachable, but because the + # prober could not read them. Undercounting reachability is the exact + # error class this service exists to eliminate, so it is fixed here. + if b"transfer-encoding: chunked" in head.lower(): + body_prefix = _dechunk(body_prefix) return code, body_prefix finally: try: @@ -331,11 +375,25 @@ def _classify(parts, req) -> tuple[str, Optional[int], str]: def _looks_like_a2a_card(body: bytes) -> bool: + """Does this response body look like an A2A Agent Card? + + TRUNCATION TOLERANCE (defect found 2026-07-31). The probe read is bounded, + so a LARGE but perfectly valid card arrives incomplete and json.loads + fails on it. The strict-parse version of this function therefore reported + well-formed agents as unproven purely for being verbose — our own card is + one of them. Undercounting reachability is the same error class as + overcounting adoption, so: parse when we can, and otherwise fall back to + the marker keys, which cannot appear in a non-JSON error page.""" + text = body.decode("utf-8", "ignore") try: - d = json.loads(body.decode("utf-8", "ignore")) + d = json.loads(text) + return isinstance(d, dict) and ("skills" in d or "protocolVersion" in d) except Exception: + pass + head = text.lstrip()[:1] + if head != "{": return False - return isinstance(d, dict) and ("skills" in d or "protocolVersion" in d) + return ('"protocolVersion"' in text) or ('"skills"' in text) # --- 4. INVOCATION VERIFICATION (trusted, AG-originated only) ----------------- diff --git a/live/guild/app/store.py b/live/guild/app/store.py index 0d955d3..72bc85a 100644 --- a/live/guild/app/store.py +++ b/live/guild/app/store.py @@ -93,6 +93,39 @@ def endpoint_fingerprint(endpoint: Optional[str]) -> Optional[str]: TRUSTED_TASK_META_KEYS = ("receipt_auth", "settlement", "guild_observed_invocation") +class CanonicalWriteRefused(RuntimeError): + """Base class for a canonical commitment this process REFUSED to make. + + Raised instead of writing when the durable state a canonical write would be + built on cannot be trusted. These are deliberately loud: a checkpoint is + pinned by third parties and cited by passports, so 'publish something and + sort it out later' is not an available failure mode. Mapped to HTTP 409 by + the API — a conflict the caller should retry after the state settles, never + a silent success.""" + + code = "canonical_write_refused" + + +class StaleDurableStateError(CanonicalWriteRefused): + """The authoritative store is BEHIND state this process already observed.""" + + code = "stale_durable_state" + + +class CheckpointForkError(CanonicalWriteRefused): + """The next checkpoint index is already published — writing would fork the + canonical feed by replacing a commitment a third party may already hold.""" + + code = "checkpoint_fork_refused" + + +class CheckpointWriteVerificationError(CanonicalWriteRefused): + """Read-after-write failed: the entry is not durably readable, or read back + with different bytes than were written.""" + + code = "checkpoint_write_unverified" + + class Store: def __init__(self, path: Optional[str] = None): self.path = path or os.environ.get("GUILD_DATA", "") @@ -103,6 +136,14 @@ def __init__(self, path: Optional[str] = None): self.accounts: dict[str, dict[str, Any]] = {} # billing key -> account self.billing_log: list[dict[str, Any]] = [] # usage + top-up ledger self.events: list[dict[str, Any]] = [] # agent-native instrumentation + # MONOTONIC IN-MEMORY MUTATION COUNTER (divergence incident 2026-07-31). + # Bumped on every recorded event and every published checkpoint. It is + # stamped on responses (X-Guild-Store-Rev) so an outside observer can + # PROVE staleness instead of inferring it: a response carrying a LOWER + # revision than one already seen from the SAME instance id is a stale + # in-process view; a lower revision across different instance ids is a + # split-brain read. Never persisted — it identifies a view, not a fact. + self.revision: int = 0 self.referrals: list[dict[str, Any]] = [] # agent-to-agent referral edges self.health_log: list[dict[str, Any]] = [] # self-evaluation snapshots self.identity: dict[str, Any] = {} # the Guild's own signing DID @@ -2050,6 +2091,7 @@ def record_event(self, key: Optional[str], etype: str, ua: str = "", **meta) -> "fp": fp, "surface": self._surface_of(key, ua or ""), "at": _now(), **meta} self.events.append(event) + self.revision += 1 # stamped on responses; see __init__ if self.backend is not None: self._persist_event(event) # durable per-row (events table) else: @@ -3264,6 +3306,16 @@ def _class_of(e: dict[str, Any]) -> str: if actor and actor != "anon": row["actors"].add(actor) return { + "reading_guide": ( + "`stages` is RAW AGGREGATE STAGE ACTIVITY (kept for " + "observability and continuity) — it is NOT a conversion " + "funnel: the stages are not actor-linked, so a ratio between " + "two of them is not a conversion rate. For conversion use " + "`qualified`, which excludes first-party/tooling/crawler " + "traffic, deduplicates exposure per actor and follows the SAME " + "actor through its own journey."), + "qualified": self.qualified_passport_funnel(), + "passport_activity": self.passport_activity(), "stages": [ {"stage": s, "count": stages[s]["breakdown"]["external"], @@ -3286,6 +3338,270 @@ def _class_of(e: dict[str, Any]) -> str: "merged"), } + # --- honest passport ACTIVITY (never called "passports" or "adoption") --- + def passport_activity(self) -> dict[str, Any]: + """Distinct, successful passport activity — split into the four things + that are NOT the same thing and must never be summed into a headline. + + The old ``passport_issued`` count answered the question "how many times + did any caller hit a passport surface", and was then read as "how many + agents adopted a passport". Those differ by more than an order of + magnitude, and the gap is entirely probes. This method reports the four + distinguishable behaviours separately, each by DISTINCT SUBJECT (not by + event), with attribution kept apart: + + subject_self_claim a proved subject fetched ITS OWN passport + — the only one of the four that means an + agent took the credential for itself + third_party_fetch someone fetched ANOTHER agent's passport + — propagation/curiosity, not adoption + third_party_verification someone VERIFIED a passport they hold + — the credential travelled and was checked + subject_evidence_attached a passport holder attached real evidence + — the only one that deepens the credential + + `attempts` and `failures` are reported alongside so a probe storm is + visible as demand instead of being laundered into issuance.""" + from . import attribution + + def _cls(e: dict[str, Any]) -> str: + if e.get("demand_first_party"): + return "first_party" + c = attribution.caller_class(e) + if c in ("AG_INTERNAL", "AG_TEST", "OPERATOR"): + return "first_party" + if (attribution.may_count_as_external_growth(c) + and attribution.is_genuine_external(e)): + return "external" + return "unknown" + + proved = {a_id for a_id, rec in self.agents.items() + if (rec.get("proof_of_conduct") or {}).get("verified_at")} + buckets = {k: {"external": set(), "first_party": set(), "unknown": set()} + for k in ("subject_self_claim", "third_party_fetch", + "third_party_verification", + "subject_evidence_attached")} + counters = {"passport_requested": 0, "passport_issue_failed": 0, + "passport_issued_events": 0} + for e in self.events: + t = e.get("type") + if t in counters: + counters[t] += 1 + if t == "passport_issued": + subj = e.get("subject_id") or "" + cls = _cls(e) + # `self_claim` is stamped at write time (2026-07-31 onward). + # Events predating the field cannot be classified as a self + # claim — they are counted as third-party fetches, which is the + # conservative direction (never inflates adoption). + if e.get("self_claim") and subj in proved: + buckets["subject_self_claim"][cls].add(subj) + elif subj: + buckets["third_party_fetch"][cls].add(subj) + elif t == "passport_verified": + cls = _cls(e) + who = e.get("subject_id") or e.get("key") or "anon" + if who != "anon": + buckets["third_party_verification"][cls].add(who) + elif t == "first_attestation_received": + cls = _cls(e) + subj = e.get("agent_id") or e.get("subject_id") or "" + if subj: + buckets["subject_evidence_attached"][cls].add(subj) + return { + "measure": "DISTINCT SUBJECTS per behaviour — event counts are " + "reported separately and are NOT passports", + "behaviours": { + k: {c: len(v) for c, v in by_cls.items()} + for k, by_cls in buckets.items()}, + "event_counts": counters, + "note": ("A passport_issued EVENT is a successful credential " + "production, not an adopting agent: one agent can appear " + "many times and a third party fetching someone else's " + "credential appears too. Only `subject_self_claim` is an " + "agent taking a credential for itself, and only " + "`subject_evidence_attached` deepens it. Never sum these " + "into a single 'passports' number."), + } + + # --- qualified cohort funnel (actor-linked, deduplicated) --------------- + QUALIFIED_EXPOSURE_WINDOW_HOURS = 24 + + def qualified_passport_funnel(self) -> dict[str, Any]: + """An ACTOR-LINKED conversion funnel, as opposed to the aggregate stage + activity in :meth:`passport_funnel`. + + WHY THIS EXISTS. The raw funnel reported 1,790 ``offer_served`` and 0 + ``offer_followed`` and that was being read as "0/1,790 conversion". It + is not: 1,787 of those serves are unattributable crawler traffic, and + exactly ONE was classified genuine external. A denominator of 1 does + not measure a conversion rate — reporting 0% implies we tested the + offer 1,790 times and it failed, when in truth we have never put it in + front of a qualified agent enough times to learn anything. + + So this view: + * EXCLUDES first-party, tooling and registry-crawler traffic; + * DEDUPLICATES exposure by (actor, source surface, time window), so a + bot hitting the agent card 800 times is one exposure, not 800; + * LINKS an exposed actor to that SAME actor's later registration, + proof, own-passport claim, evidence and return; + * keeps third-party propagation (someone else fetching or verifying a + passport) in a SEPARATE loop, because it is not this actor + converting; + * reports UNLINKABLE exposure honestly instead of counting it as a + failed conversion. + + The load-bearing output is ``next_boundary``: the first stage where the + cohort actually stops, together with the SAMPLE SIZE behind it, so a + boundary measured on n=1 is never presented as a finding.""" + from . import attribution + + def _qualified(e: dict[str, Any]) -> bool: + if e.get("fp") or e.get("demand_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)) + + def _bucket(ts: str) -> str: + # coarse dedup window: exposure is per actor+source+window, not per hit + return (ts or "")[:13] # YYYY-MM-DDTHH + + # ---- 1. qualified, deduplicated exposure --------------------------- + exposures: set[tuple[str, str, str]] = set() + exposed_actors: set[str] = set() + anonymous_exposures = 0 + raw_qualified_serves = 0 + by_source: dict[str, int] = {} + for e in self.events: + if e.get("type") != "offer_served" or e.get("offer") != "passport": + continue + if not _qualified(e): + continue + raw_qualified_serves += 1 + actor = e.get("key") or "anon" + source = str(e.get("endpoint") or e.get("surface") or "unknown") + by_source[source] = by_source.get(source, 0) + 1 + if actor == "anon": + # An anonymous serve CANNOT be linked to a later registration. + # It is reach we cannot measure, not a conversion that failed. + anonymous_exposures += 1 + continue + exposures.add((actor, source, _bucket(e.get("at", "")))) + exposed_actors.add(actor) + + # ---- 2. the same actors' later journey ----------------------------- + # actor key -> agent_id, for actors that went on to register + actor_to_agent: dict[str, str] = {} + for a_id in self.agents: + k = self.account_for_agent(a_id) + if k: + actor_to_agent.setdefault(k, a_id) + + stage_actors: dict[str, set[str]] = { + k: set() for k in ("registered", "control_proved", + "own_passport_claimed", "evidence_attached", + "returned")} + for e in self.events: + actor = e.get("key") or "" + if actor not in exposed_actors: + continue + t = e.get("type") + if t == "register": + stage_actors["registered"].add(actor) + elif t == "prove_completed": + stage_actors["control_proved"].add(actor) + elif t == "passport_issued" and e.get("self_claim"): + stage_actors["own_passport_claimed"].add(actor) + elif t == "first_attestation_received": + stage_actors["evidence_attached"].add(actor) + elif t == "liveness_refreshed": + stage_actors["returned"].add(actor) + + cohort = len(exposed_actors) + ordered = ["qualified_exposure", "registered", "control_proved", + "own_passport_claimed", "evidence_attached", "returned"] + counts = {"qualified_exposure": cohort, + **{k: len(v) for k, v in stage_actors.items()}} + + # ---- 3. first boundary the cohort actually stops at ----------------- + next_boundary: dict[str, Any] = { + "boundary": None, "n": 0, + "measurable": False, + "reason": "no qualified exposure yet — nothing to convert", + } + for prev, cur in zip(ordered, ordered[1:]): + if counts[prev] == 0: + next_boundary = { + "boundary": f"{prev} → {cur}", + "n": 0, + "measurable": False, + "reason": (f"denominator is zero: no qualified actor has " + f"reached '{prev}', so the {prev}→{cur} rate is " + "NOT MEASURABLE (reporting 0% would claim a " + "test we never ran)"), + } + break + if counts[cur] == 0: + next_boundary = { + "boundary": f"{prev} → {cur}", + "n": counts[prev], + "measurable": True, + "rate": 0.0, + "reason": (f"{counts[prev]} qualified actor(s) reached " + f"'{prev}' and none reached '{cur}'"), + "sample_adequacy": ("ANECDOTE — a single-digit denominator " + "cannot distinguish a broken offer " + "from bad luck" + if counts[prev] < 10 else "usable"), + } + break + else: + next_boundary = {"boundary": None, "n": cohort, "measurable": True, + "reason": "the cohort reaches every stage"} + + # ---- 4. propagation loop (NOT this actor converting) --------------- + propagation = {"third_party_passport_fetch": 0, + "third_party_verification": 0} + for e in self.events: + if not _qualified(e): + continue + if e.get("type") == "passport_issued" and not e.get("self_claim"): + propagation["third_party_passport_fetch"] += 1 + elif e.get("type") == "passport_verified": + propagation["third_party_verification"] += 1 + + return { + "cohort": { + "qualified_distinct_actors": cohort, + "qualified_deduplicated_exposures": len(exposures), + "raw_qualified_serves": raw_qualified_serves, + "anonymous_unlinkable_serves": anonymous_exposures, + "dedup_rule": ("one exposure per (actor, source, " + f"{self.QUALIFIED_EXPOSURE_WINDOW_HOURS}h " + "window); repeated hits by the same actor are " + "reach, not additional trials"), + "by_source": dict(sorted(by_source.items())), + }, + "stages": [{"stage": s, "qualified_actors": counts[s]} + for s in ordered], + "next_boundary": next_boundary, + "propagation_loop": propagation, + "excluded": ("first-party, AG test harnesses, release gates, " + "canaries and registry crawlers are excluded " + "STRUCTURALLY (attribution.caller_class + " + "is_genuine_external) — they are not in any " + "denominator here"), + "honesty": ("Anonymous serves are reported as UNLINKABLE reach, " + "never as failed conversions: an anonymous a2a probe " + "that never identifies itself cannot be followed to a " + "registration, so it can neither convert nor fail to " + "convert. The aggregate stage activity remains " + "available under `stages` in /funnel/passports."), + } + def evidence_staleness(self, agent_id: str) -> Optional[dict[str, Any]]: """Staleness of an agent's evidence: age of the most recent attestation it received or receipt it delivered. §15 lists staleness as a required @@ -4131,18 +4447,117 @@ def reclassify_ledger(self) -> dict[str, Any]: "examined": len(led.collabs()), "appended": appended} # --- published checkpoints (stage-2: pinnable canonical commitments) ----- + def state_diagnostics(self) -> dict[str, Any]: + """Non-secret, side-effect-free view identity: WHICH process and WHICH + state produced this response (divergence incident 2026-07-31). + + Exposes the in-memory view alongside the AUTHORITATIVE durable view so + the two can be compared from outside without a shell on the box. It + leaks no paths, no secrets and no environment — only counts, a random + per-process id and the durable head hash (already public in the + checkpoint feed). + + ``divergence`` is the load-bearing field: a non-empty list means the + in-memory view this process is serving reads from does NOT agree with + the committed database it writes to. That is the condition the ops pass + must detect; "discard the first few reads" is a reporting workaround, + not a write-path control.""" + from . import instanceid + mem = { + "events": len(self.events), + "agents": len(self.agents), + "ledger_records": len(self.ledger_records), + "checkpoints": len(self.checkpoints), + "checkpoint_head_index": (self.checkpoints[-1].get("index") + if self.checkpoints else None), + "store_rev": self.revision, + } + out: dict[str, Any] = { + **instanceid.identity(), + "store_mode": self.store_mode, + "in_memory": mem, + "durable": None, + "divergence": [], + } + if self.backend is None: + out["durable"] = {"note": "json store — the in-memory view IS the " + "authoritative view (single writer)"} + return out + # AUTHORITATIVE read. Deliberately OUTSIDE any write txn: this is a + # diagnostic, and it must never take the write lock just to be read. + try: + durable = self.backend.durable_counts() + except Exception as exc: # noqa: BLE001 — diagnostics never 500 + out["durable"] = {"error": type(exc).__name__} + out["divergence"].append("durable_read_failed") + return out + out["durable"] = durable + if durable["events"] < mem["events"]: + out["divergence"].append("in_memory_events_ahead_of_durable") + if durable["events"] > mem["events"]: + out["divergence"].append("durable_events_ahead_of_in_memory") + if durable["checkpoints"] < mem["checkpoints"]: + out["divergence"].append("in_memory_checkpoints_ahead_of_durable") + if durable["checkpoints"] > mem["checkpoints"]: + out["divergence"].append("durable_checkpoints_ahead_of_in_memory") + if durable["ledger_records"] != mem["ledger_records"]: + out["divergence"].append("ledger_length_mismatch") + out["consistent"] = not out["divergence"] + return out + def publish_checkpoint(self) -> dict[str, Any]: """Seal the current ledger head into a Guild-signed checkpoint and add it to the published, append-only checkpoint feed third parties pin (LEDGER_ARCHITECTURE §7 stage-2). Idempotent: if no evidence has landed since the last published checkpoint, the existing one is returned rather - than publishing a duplicate. Meant to be called on a schedule.""" + than publishing a duplicate. Meant to be called on a schedule. + + FAIL-CLOSED (divergence incident 2026-07-31). A publish is a CANONICAL + COMMITMENT: third parties pin it, passports cite it, and an inclusion + proof is only worth anything if the feed is a real chain. On 2026-07-31 + a publish returned index 14 / ledger_length 834 while the feed head was + already 16 / 836 — i.e. the write path was willing to act on a view two + entries behind the committed feed. It no longer is: + + 1. The authoritative read now happens INSIDE the BEGIN IMMEDIATE + write transaction, so no other writer can land between the read + and the append (the previous code read, then wrote — a TOCTOU + window in which two publishers both computed the same next index). + 2. A durable view that is BEHIND what this process already published + is refused, not published from (``StaleDurableStateError``). + 3. The next index must be UNUSED in the durable feed. Combined with a + strict INSERT (no INSERT OR REPLACE) this makes silently + overwriting a published checkpoint impossible — a fork of the + canonical feed now raises instead of replacing history. + 4. READ-AFTER-WRITE: the entry is re-read from the database and its + canonical bytes compared before it is returned. A publish that did + not durably land can no longer be reported as published. + """ with self.lock, self._txn(): if self.backend is not None: - # build on the AUTHORITATIVE committed ledger + checkpoint feed, - # not a possibly-stale in-memory view (concurrent appenders). - self.ledger_records = self.backend.all_ledger() - self.checkpoints = self.backend.all_checkpoints() + # AUTHORITATIVE read, INSIDE the write transaction (see 1). + durable_ledger_records = self.backend.all_ledger() + durable_checkpoints = self.backend.all_checkpoints() + # (2) refuse to build a canonical commitment on a view that is + # behind state this process has already observed as published. + mem_head = (self.checkpoints[-1].get("index") + if self.checkpoints else -1) + dur_head = (durable_checkpoints[-1].get("index") + if durable_checkpoints else -1) + if dur_head < mem_head: + raise StaleDurableStateError( + "refusing to publish: the durable checkpoint feed head " + f"({dur_head}) is BEHIND the head this process already " + f"observed ({mem_head}). Publishing from a stale view " + "would fork the canonical feed.") + if len(durable_ledger_records) < len(self.ledger_records): + raise StaleDurableStateError( + "refusing to publish: the durable ledger " + f"({len(durable_ledger_records)} records) is SHORTER " + f"than the in-memory ledger ({len(self.ledger_records)})" + " — a short head would commit to a truncated history.") + self.ledger_records = durable_ledger_records + self.checkpoints = durable_checkpoints gid = self.guild_identity() led = self.durable_ledger() cp = led.signed_checkpoint(gid["did"], gid["private_key"]) @@ -4152,8 +4567,14 @@ def publish_checkpoint(self) -> dict[str, Any]: if (last["checkpoint"].get("head_hash") == head and len(self.ledger_records) == last.get("ledger_length")): return last # nothing new to commit + # NEXT INDEX from the maximum index actually present, not from the + # list LENGTH: a feed with any gap (or an out-of-order legacy entry) + # would otherwise re-issue an index that already exists, which is + # exactly the silent-overwrite path (3) closes. + next_index = 1 + max( + [int(e.get("index", -1)) for e in self.checkpoints] or [-1]) entry = { - "index": len(self.checkpoints), + "index": next_index, "published_at": _now(), "ledger_length": len(self.ledger_records), "checkpoint": cp, @@ -4191,9 +4612,25 @@ def publish_checkpoint(self) -> dict[str, Any]: "are NOT rewritten"), } entry["entry_proof"] = sign_jcs(entry, gid["private_key"]) - self.checkpoints.append(entry) if self.backend is not None: - self._persist_checkpoint(entry) + # (3) FORK PREVENTION — strict insert. The index must be unused + # in the durable feed; a collision raises instead of silently + # replacing a published, third-party-pinned commitment. + self.backend.insert_checkpoint_strict(entry) + # (4) READ-AFTER-WRITE — a publish that did not durably land is + # never reported as published. + stored = self.backend.checkpoint_at(next_index) + if stored is None: + raise CheckpointWriteVerificationError( + f"checkpoint {next_index} was not readable after " + "write; refusing to report it as published") + if canonicalize(stored) != canonicalize(entry): + raise CheckpointWriteVerificationError( + f"checkpoint {next_index} read back with different " + "bytes than were written; refusing to report it as " + "published") + self.checkpoints.append(entry) + self.revision += 1 self._save() return entry @@ -4988,13 +5425,33 @@ def guild_did(self) -> str: def issue_passport(self, agent_id: str, *, ttl_days: int = 7, verify_url: Optional[str] = None, - explore_url: Optional[str] = None) -> Optional[dict[str, Any]]: + explore_url: Optional[str] = None, + actor_key: Optional[str] = None, + surface: Optional[str] = None, + ua: str = "", + request_id: Optional[str] = None) -> Optional[dict[str, Any]]: """Issue a portable, Guild-signed **Agent Passport** for `agent_id`: a Verifiable Credential snapshotting its current reputation that the agent can carry to any counterparty or platform. Each passport embeds a verification URL, so every counterparty who checks it is pulled back to the Guild — the credential is the distribution loop. None if the agent or its - reputation is unknown.""" + reputation is unknown. + + TELEMETRY CONTRACT (corrective pass 2026-07-31). Exactly ONE + ``passport_issued`` event is emitted, HERE, and ONLY after a credential + was actually produced. Before this change the MCP tool recorded + ``passport_issued`` on ENTRY and this method recorded a second one on + success, so a single MCP call counted twice and a LOOKUP MISS (unknown + agent) counted as an issuance. That is how a schema probe on 2026-07-30 + moved the headline "genuine external passports issued" from 1 to 3 + without a single agent registering, proving control or returning. + + Attempts and failures are now recorded by the CALLER as + ``passport_requested`` / ``passport_issue_failed`` — separate event + types that are counted as demand, never as issuance. ``actor_key``, + ``surface``, ``ua`` and ``request_id`` are stamped so a passport event + can be attributed to a transport and correlated with its request + instead of being inferred later.""" rec = self.get_agent(agent_id) if not rec: return None @@ -5046,8 +5503,19 @@ def issue_passport(self, agent_id: str, *, ttl_days: int = 7, subject_did=rec["did"], subject_claims=claims, valid_from=created.isoformat(), valid_until=until.isoformat(), ) - self.record_event(self.account_for_agent(agent_id), "passport_issued", - endpoint="passport", subject_id=agent_id) + # ONE event, on SUCCESS only. `actor_key` is WHO fetched it (may be the + # subject itself or a third party); `subject_id` is WHOSE credential it + # is. Keeping both is what makes a self-claim distinguishable from a + # third-party fetch downstream — see passport_activity(). + subject_account = self.account_for_agent(agent_id) + actor = actor_key or subject_account + self.record_event(actor, "passport_issued", ua=ua, + endpoint="passport", subject_id=agent_id, + subject_account=subject_account, + transport=(surface or self._surface_of(actor, ua)), + request_id=request_id, + self_claim=bool(subject_account + and actor == subject_account)) if self.record_milestone(agent_id, "first_passport"): self._save() # milestone stamps mutate the agent record; persist it return cred @@ -5115,10 +5583,48 @@ def _health_vector(self) -> dict[str, Any]: agents_external = sum(1 for a in self.agents.values() if not a.get("first_party")) credits_spent_ext = sum(a.get("spent", 0) for a in self.accounts.values() if not a.get("first_party")) + # --- HONESTY FIXES (corrective pass 2026-07-31) --------------------- + # 1. UTILITY comes from the PRODUCTION-only block of /evaluation. The + # top-level `lift` mixes a seeded first-party bootstrap cohort with + # live traffic; quoting it as "measured lift" reported a number we + # manufactured as if agents had produced it. n_recommended travels + # with it so a lift computed on zero production recommendations is + # visibly unmeasurable rather than silently absent. + prod = ev.get("production") or {} + prod_lift = prod.get("lift") + prod_n = int(prod.get("n_recommended") or 0) + # 2. REVENUE comes ONLY from independently confirmed external mainnet + # settlement. Sandbox credits are an internal accounting unit with + # no external claim on anything; multiplying them by CREDIT_USD and + # printing a dollar sign was inventing money. + rev = self.escrow_summary() + real = (rev or {}).get("real_settlement") or {} + verified_external_usd = float( + real.get("independently_attested_external_revenue_usd") or 0.0) + bound_machine_usd = float( + real.get("cryptographically_bound_machine_revenue_usd") or 0.0) + # 3. GROWTH uses adoption-grade external actors (the central + # attribution rule), not "every record lacking first_party=true", + # which still counts our own untagged tooling and every crawler that + # ever registered. + activity = self.passport_activity()["behaviours"] + adoption_grade_self_claims = activity["subject_self_claim"]["external"] return { - "measured_lift": ev.get("lift"), + # honest utility + "production_measured_lift": prod_lift, + "production_n_recommended": prod_n, + "production_lift_measurable": bool(prod_n), + # kept, but explicitly labelled as the mixed/seeded number so it + # can never be quoted as the production result again + "mixed_bootstrap_lift_NOT_PRODUCTION": ev.get("lift"), "measured_lift_dataset": ev.get("dataset"), "recommended_success_rate": ev.get("recommended_success_rate"), + # honest revenue + "verified_external_revenue_usd": verified_external_usd, + "cryptographically_bound_machine_revenue_usd": bound_machine_usd, + "sandbox_credits_spent_external_NOT_MONEY": credits_spent_ext, + # honest growth + "adoption_grade_external_self_claims": adoption_grade_self_claims, "agents_total": len(self.agents), "agents_external": agents_external, "genuine_external_detected": instr.get("genuine_external_detected", False), @@ -5126,8 +5632,15 @@ def _health_vector(self) -> dict[str, Any]: "external_repeat_query_agents": ext.get("repeat_query", 0), "external_repeat_paid_agents": ext.get("repeat_paid_query_agents", 0), "external_paid_queries": ext.get("paid_query", 0), + "sandbox_credits_spent_external_NOT_MONEY": credits_spent_ext, "credits_spent_external": credits_spent_ext, - "revenue_usd_external": round(credits_spent_ext * CREDIT_USD, 4), + # RETIRED as a revenue line (2026-07-31): this was + # sandbox-credit spend multiplied by a notional rate and labelled + # USD. Kept under an unmistakable key ONLY so the historical + # health series stays comparable; it is not money and must never + # be summed with verified_external_revenue_usd. + "legacy_sandbox_credit_notional_usd_NOT_REVENUE": + round(credits_spent_ext * CREDIT_USD, 4), "total_referrals": ref["total_referrals"], "activated_referrals": ref["activated_referrals"], } @@ -5135,20 +5648,59 @@ def _health_vector(self) -> dict[str, Any]: @staticmethod def _verdict(v: dict[str, Any], deltas: dict[str, float]) -> str: """A blunt, honest read of whether the autonomous flywheel is turning. - The load-bearing signal is *external* agents climbing the value ladder — - not totals we can inflate ourselves.""" + + GATE (corrective pass 2026-07-31). "FLYWHEEL TURNING — external agents + pay" was printed on 2026-07-31 while verified external revenue was + $0.00, zero external agents had claimed a passport, and the only "paid + read" on the books was our own release gate. It reached that verdict + because `external_paid_queries` counts SANDBOX CREDITS — an internal + unit we mint — and `agents_external` counts every record lacking + first_party=true, including our own untagged tooling and crawlers. + + The verdict now requires BOTH halves of the claim to be independently + true: + * genuine external MOVEMENT — an adoption-grade external agent took a + credential for itself (not a probe fetching someone else's), and + * real verified ECONOMIC VALUE — independently confirmed external + mainnet settlement, never sandbox credits and never a first-party + canary. + Anything less names exactly which half is missing.""" + verified_revenue = float(v.get("verified_external_revenue_usd") or 0.0) + self_claims = int(v.get("adoption_grade_external_self_claims") or 0) + if v["agents_external"] == 0: - return "NO EXTERNAL AGENTS YET — deploy and seed discovery; every metric is self-traffic until one outside agent calls." + return ("NO EXTERNAL AGENTS YET — deploy and seed discovery; every " + "metric is self-traffic until one outside agent calls.") if v.get("external_querying_agents", 0) == 0: - return "REGISTRATIONS BUT NO DISCOVERY — external agents exist but none have queried the trust layer yet; the core product is untested in the wild." + return ("REGISTRATIONS BUT NO DISCOVERY — external agents exist but " + "none have queried the trust layer yet; the core product is " + "untested in the wild.") if v["external_repeat_query_agents"] == 0: - return "REACH BUT NO RETENTION — outside agents have queried but none came back; usefulness unproven." - if v["external_paid_queries"] == 0: - return "RETENTION BUT NO WILLINGNESS-TO-PAY — agents return for free reads but none spend their own budget yet." - growing = deltas.get("agents_external", 0) > 0 or deltas.get("activated_referrals", 0) > 0 - return ("FLYWHEEL TURNING — external agents pay and the network is growing." + return ("REACH BUT NO RETENTION — outside agents have queried but " + "none came back; usefulness unproven.") + if self_claims == 0 and verified_revenue <= 0: + return ("REACH WITHOUT ADOPTION OR REVENUE — no external agent has " + "claimed a credential for itself and verified external " + "revenue is $0.00. Sandbox credits and first-party canaries " + "are NOT evidence of either.") + if self_claims > 0 and verified_revenue <= 0: + return (f"ADOPTION WITHOUT REVENUE — {self_claims} adoption-grade " + "external agent(s) hold their own credential, but verified " + "external revenue is $0.00; willingness-to-pay is unproven.") + if self_claims == 0 and verified_revenue > 0: + return (f"REVENUE WITHOUT ADOPTION — ${verified_revenue:.2f} of " + "verified external settlement, but no external agent has " + "taken a credential for itself; check the payer is not a " + "one-off before calling this a flywheel.") + growing = (deltas.get("adoption_grade_external_self_claims", 0) > 0 + or deltas.get("verified_external_revenue_usd", 0) > 0) + return (f"FLYWHEEL TURNING — {self_claims} external agent(s) hold their " + f"own credential AND ${verified_revenue:.2f} of verified " + "external settlement is on the books." if growing else - "PAID BUT FLAT — agents pay, but growth/referrals stalled this period; investigate acquisition.") + f"PAID BUT FLAT — {self_claims} credential holder(s), " + f"${verified_revenue:.2f} verified external revenue, no " + "movement this period; investigate acquisition.") def compute_health(self, persist: bool = False) -> dict[str, Any]: """Compute the health snapshot (vector + trend deltas vs the last diff --git a/live/guild/app/store_sqlite.py b/live/guild/app/store_sqlite.py index 9fb1a96..a7118f5 100644 --- a/live/guild/app/store_sqlite.py +++ b/live/guild/app/store_sqlite.py @@ -288,10 +288,24 @@ def _begin(self) -> None: self._local.depth = depth + 1 def _commit(self) -> None: + """Close one nesting level; COMMIT at the outermost. + + DEPTH CANNOT GO NEGATIVE (divergence hardening 2026-07-31). Previously, + a NESTED transaction that raised called ``_rollback`` (which zeroes the + depth and rolls back the OUTER transaction too); the outer ``__exit__`` + then ran ``_commit``, taking the depth to -1. From then on this THREAD's + connection was permanently mis-tracked: ``_begin`` saw a non-zero depth + and skipped ``BEGIN IMMEDIATE``, so subsequent 'transactions' silently + ran in autocommit — losing atomicity on multi-entity invariants — and + ``in_transaction()`` lied to ``Store._save``. Connections are + THREAD-LOCAL, so one poisoned request thread would keep serving a + subtly different write path from every other thread for the life of the + process. Clamping at zero makes the state self-healing: after a nested + rollback the next ``_begin`` opens a real transaction again.""" con = self.conn() - depth = getattr(self._local, "depth", 1) - 1 + depth = max(0, getattr(self._local, "depth", 1) - 1) self._local.depth = depth - if depth == 0: + if depth == 0 and con.in_transaction: self._retry(con.commit) def _rollback(self) -> None: @@ -570,9 +584,69 @@ def put_referral(self, rec: dict[str, Any]) -> None: (rec.get("referred_id"), _j(rec))) def put_checkpoint(self, rec: dict[str, Any]) -> None: + """Upsert a checkpoint row. + + RETAINED FOR REPLAY/MIGRATION ONLY (``_sqlite_initial_load`` and the + JSON→SQLite cutover re-materialise existing history and must be + idempotent). NEW canonical publications MUST go through + ``insert_checkpoint_strict`` — an INSERT OR REPLACE on the canonical + feed can silently overwrite a checkpoint a third party has already + pinned, which is precisely the fork this backend must make + impossible.""" self._exec("INSERT OR REPLACE INTO checkpoints (idx,json) VALUES (?,?)", (rec.get("index"), _j(rec))) + def insert_checkpoint_strict(self, rec: dict[str, Any]) -> None: + """Append a NEW checkpoint. Fails closed if the index already exists. + + ``idx`` is the table's INTEGER PRIMARY KEY, so a plain INSERT lets + SQLite itself enforce append-only-ness: a duplicate raises + IntegrityError, which is translated into ``CheckpointForkError`` rather + than being swallowed. Runs inside the caller's BEGIN IMMEDIATE, so the + uniqueness check and the append are one serialized step (no TOCTOU + window for a second publisher).""" + from .store import CheckpointForkError + idx = rec.get("index") + try: + self._exec("INSERT INTO checkpoints (idx,json) VALUES (?,?)", + (idx, _j(rec))) + except sqlite3.IntegrityError as exc: + raise CheckpointForkError( + f"checkpoint index {idx} is already published — refusing to " + "replace a canonical commitment third parties may hold" + ) from exc + + def checkpoint_at(self, idx: int) -> Optional[dict[str, Any]]: + """Read one checkpoint back by index (read-after-write verification).""" + row = self.conn().execute( + "SELECT json FROM checkpoints WHERE idx=?", (idx,)).fetchone() + return json.loads(row[0]) if row else None + + def durable_counts(self) -> dict[str, Any]: + """Authoritative row counts + feed head, for divergence detection. + + Deliberately cheap and read-only: counts plus the head checkpoint's + index/head_hash. Nothing here is secret — the head hash is already + published in the pinnable checkpoint feed.""" + con = self.conn() + + def _n(table: str) -> int: + return int(con.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]) + + head = con.execute( + "SELECT json FROM checkpoints ORDER BY idx DESC LIMIT 1").fetchone() + head_rec = json.loads(head[0]) if head else None + return { + "events": _n("events"), + "agents": _n("agents"), + "ledger_records": _n("ledger"), + "checkpoints": _n("checkpoints"), + "checkpoint_head_index": (head_rec.get("index") if head_rec else None), + "checkpoint_head_hash": ( + (head_rec.get("checkpoint") or {}).get("head_hash") + if head_rec else None), + } + def append_billing(self, rec: dict[str, Any]) -> None: self._exec("INSERT INTO billing_log (json) VALUES (?)", (_j(rec),)) diff --git a/live/guild/contract/contract.json b/live/guild/contract/contract.json index c26f66c..4b6c584 100644 --- a/live/guild/contract/contract.json +++ b/live/guild/contract/contract.json @@ -47,6 +47,7 @@ "guild_escrow_open", "guild_escrow_release", "guild_passport", + "guild_preflight", "guild_prove", "guild_prove_verify", "guild_record", @@ -392,6 +393,12 @@ ], "path": "/demand/watch" }, + { + "methods": [ + "GET" + ], + "path": "/diagnostics/state" + }, { "methods": [ "GET" @@ -592,6 +599,12 @@ ], "path": "/outcomes" }, + { + "methods": [ + "GET" + ], + "path": "/preflight" + }, { "methods": [ "POST" diff --git a/live/guild/tests/test_preflight.py b/live/guild/tests/test_preflight.py new file mode 100644 index 0000000..ac5a98d --- /dev/null +++ b/live/guild/tests/test_preflight.py @@ -0,0 +1,187 @@ +"""Delegation preflight + the reachability defects it uncovered. + +The preflight's whole value is that it does NOT overstate. Two failure +directions are equally fatal and both are tested here: + + * OVERSTATING — reporting a check as passed when it was not performed, or + averaging unknowns into a clean verdict. That is the badge problem. + * UNDERSTATING — reporting a well-formed agent as broken because our own + prober could not read it. Two real instances of this were found while + building the endpoint and are locked below: chunked transfer-encoding, and + a card larger than the bounded probe read. +""" +from __future__ import annotations + +import json +import os +import sys + +import pytest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from app import preflight, reachability # noqa: E402 + + +# -------------------------------------------------------------------------- +# Reachability defects found 2026-07-31 +# -------------------------------------------------------------------------- +def test_dechunk_decodes_a_chunked_body(): + raw = b"1a\r\n{\"protocolVersion\": \"0.3\"}\r\n0\r\n\r\n" + assert reachability._dechunk(raw) == b'{"protocolVersion": "0.3"}' + + +def test_dechunk_tolerates_truncation_at_the_read_cap(): + """The probe read is bounded, so the final chunk is routinely cut off and + the terminating 0-chunk never arrives. We must still return what we got.""" + raw = b"20\r\n{\"protocolVersion\": \"0.3\", \"na" + out = reachability._dechunk(raw) + assert out.startswith(b'{"protocolVersion"') + + +def test_dechunk_passes_through_a_non_chunked_body(): + raw = b'{"protocolVersion": "0.3"}' + assert reachability._dechunk(raw) == raw + + +def test_card_detection_survives_truncation(): + """A LARGE valid card arrives incomplete and will not json.loads. Calling + that 'not an agent' undercounts reachability — the same error class as + overcounting adoption, in the other direction. This is exactly why + `verified_reachable` read 0 for every entry in the demand feed.""" + whole = json.dumps({"protocolVersion": "0.3.0", "name": "x", + "skills": [{"id": "a"}], "description": "y" * 500}) + truncated = whole[:200].encode() + assert reachability._looks_like_a2a_card(whole.encode()) is True + assert reachability._looks_like_a2a_card(truncated) is True + + +def test_card_detection_still_rejects_a_non_card(): + assert reachability._looks_like_a2a_card(b"404") is False + assert reachability._looks_like_a2a_card(b'{"hello": "world"}') is False + # a JSON body that merely mentions the word must not qualify + assert reachability._looks_like_a2a_card(b'{"note": "skills are nice"}') is False + + +# -------------------------------------------------------------------------- +# Verdict semantics — unknowns are never laundered into a pass +# -------------------------------------------------------------------------- +def _fake(monkeypatch, *, probe, card_body=b"", card_code=200, root_code=200): + monkeypatch.setattr(reachability, "liveness_probe", lambda url, **kw: probe) + + def _get(url, path, timeout_ctx=None): + if "agent-card" in path: + return card_code, card_body, "" + return root_code, b"", "" + + monkeypatch.setattr(preflight, "_probe_get", _get) + + +def test_http_200_without_a_handshake_is_do_not_delegate(monkeypatch): + """THE headline case: 92.9% of listed agents report healthy, 33.9% + complete a task. A bare HTTP 200 must never read as working.""" + _fake(monkeypatch, probe={"status": "http_responsive", + "evidence_level": "http_response"}) + out = preflight.run("https://example.com/a2a") + assert out["verdict"] == "do_not_delegate" + assert "protocol_handshake" in out["failed"] + + +def test_unreachable_reports_downstream_checks_as_unknown_not_failed(monkeypatch): + _fake(monkeypatch, probe={"status": "currently_unreachable", + "evidence_level": "none"}, card_code=0) + out = preflight.run("https://example.com/a2a") + assert out["verdict"] == "do_not_delegate" + assert "protocol_handshake" in out["unknowns"] + assert "protocol_handshake" not in out["failed"] + + +def test_unsigned_card_is_caution_not_a_block(monkeypatch): + """0.8% of cards are signed. An unsigned card is the norm, so it must + inform the caller without pretending the agent is broken.""" + card = json.dumps({"protocolVersion": "0.3", "name": "x"}).encode() + _fake(monkeypatch, probe={"status": "recently_reachable", + "evidence_level": "protocol_handshake"}, + card_body=card) + out = preflight.run("https://example.com/a2a") + assert out["verdict"] == "delegate_with_caution" + assert "agent_card_signed" in out["failed"] + + +def test_payment_claim_that_does_not_challenge_is_a_failure(monkeypatch): + """5.7% of self-declared paid agents actually return 402.""" + card = json.dumps({"protocolVersion": "0.3", "x402": {"price": "0.01"}}).encode() + _fake(monkeypatch, probe={"status": "recently_reachable", + "evidence_level": "protocol_handshake"}, + card_body=card, root_code=200) + out = preflight.run("https://example.com/a2a") + assert "payment_claim_holds" in out["failed"] + + +def test_payment_claim_that_does_challenge_passes(monkeypatch): + card = json.dumps({"protocolVersion": "0.3", "x402": {"price": "0.01"}}).encode() + _fake(monkeypatch, probe={"status": "recently_reachable", + "evidence_level": "protocol_handshake"}, + card_body=card, root_code=402) + out = preflight.run("https://example.com/a2a") + assert "payment_claim_holds" not in out["failed"] + + +def test_no_payment_claim_is_unknown_not_a_pass(monkeypatch): + card = json.dumps({"protocolVersion": "0.3", "name": "free thing"}).encode() + _fake(monkeypatch, probe={"status": "recently_reachable", + "evidence_level": "protocol_handshake"}, + card_body=card) + out = preflight.run("https://example.com/a2a") + assert "payment_claim_holds" in out["unknowns"] + + +def test_clean_verdict_still_publishes_its_unknowns(monkeypatch): + """A pass over four unknowns is not a pass over eight checks. The counts + must travel with the verdict so a caller can tell the difference.""" + card = json.dumps({"protocolVersion": "0.3", + "signatures": [{"protected": "x"}]}).encode() + _fake(monkeypatch, probe={"status": "recently_reachable", + "evidence_level": "protocol_handshake"}, + card_body=card) + out = preflight.run("https://example.com/a2a") + assert out["verdict"] == "no_failed_checks" + assert out["unknowns"], "a clean verdict must still declare what it could not check" + assert "not an endorsement" in out["headline"] + assert len(out["scored"]) + len(out["unknowns"]) == len(out["checks"]) + + +def test_absence_of_evidence_is_not_reported_as_risk(monkeypatch): + card = json.dumps({"protocolVersion": "0.3"}).encode() + _fake(monkeypatch, probe={"status": "recently_reachable", + "evidence_level": "protocol_handshake"}, + card_body=card) + out = preflight.run("https://example.com/a2a") + ev = next(c for c in out["checks"] if c["check"] == "independent_evidence") + assert ev["status"] == "unknown" + assert "NOT evidence of risk" in ev["detail"] + + +def test_signature_presence_is_never_called_verification(monkeypatch): + card = json.dumps({"protocolVersion": "0.3", + "signatures": [{"protected": "x"}]}).encode() + _fake(monkeypatch, probe={"status": "recently_reachable", + "evidence_level": "protocol_handshake"}, + card_body=card) + out = preflight.run("https://example.com/a2a") + sig = next(c for c in out["checks"] if c["check"] == "agent_card_signed") + assert "not verified here" in sig["detail"] + + +def test_private_and_loopback_targets_are_refused(monkeypatch): + """SSRF: the preflight must never be usable as an internal port scanner.""" + for target in ("http://127.0.0.1:8000/a2a", "http://169.254.169.254/", + "http://10.0.0.5/a2a", "file:///etc/passwd"): + out = preflight.run(target) + assert out["verdict"] == "do_not_delegate", target + + +def test_preflight_never_raises_on_hostile_input(): + for target in ("", "not-a-url", "https://", "http://[::1]/", "x" * 600): + out = preflight.run(target) + assert "verdict" in out diff --git a/live/guild/tests/test_state_divergence.py b/live/guild/tests/test_state_divergence.py new file mode 100644 index 0000000..3b85e3d --- /dev/null +++ b/live/guild/tests/test_state_divergence.py @@ -0,0 +1,261 @@ +"""Divergence incident 2026-07-30/31 — regression suite for the truth layer. + +WHAT PRODUCTION ACTUALLY DID + * ``/instrumentation`` and ``/funnel/passports`` each served the PREVIOUS + stable snapshot on the first reads of a session, then the current one. + * ``POST /ledger/checkpoint/publish`` returned checkpoint index 14 / + ledger_length 834 while the published feed was already at 16 / 836. + +WHAT WAS AND WAS NOT PROVED (see docs/DIVERGENCE_2026-07-31.md) + The read-side flip could not be attributed: 40 concurrent ``/release`` probes + returned ONE ``_PROCESS_STARTED_AT``, 60 concurrent mixed requests showed no + cross-request body mixing, and unique cache-busters ruled out URL-keyed + caching. So the read-side cause remains UNKNOWN and is deliberately NOT + "fixed" here — it is made DECIDABLE (instance/revision stamping) instead. + + The WRITE side is a different matter: a stale/racing durable view on the + canonical commitment path is reproducible, and these tests reproduce it with + two ``Store`` instances over one shared SQLite file — the faithful analogue + of two writers over one disk. That class of bug is now refused, not papered + over. +""" +from __future__ import annotations + +import os +import sys +import threading + +import pytest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from app import instanceid # noqa: E402 +from app.store import ( # noqa: E402 + CanonicalWriteRefused, + CheckpointForkError, + CheckpointWriteVerificationError, + StaleDurableStateError, + Store, +) + + +def _sqlite_store(tmp_path, name="guild.json") -> Store: + os.environ["GUILD_STORE"] = "sqlite" + os.environ["GUILD_STORE_PATH"] = str(tmp_path / "guild.sqlite3") + return Store(path=str(tmp_path / name)) + + +@pytest.fixture(autouse=True) +def _clean_env(): + yield + for k in ("GUILD_STORE", "GUILD_STORE_PATH"): + os.environ.pop(k, None) + + +# -------------------------------------------------------------------------- +# 1. View identity — the decidability layer +# -------------------------------------------------------------------------- +def test_instance_identity_is_non_secret_and_stable_within_process(): + a, b = instanceid.identity(), instanceid.identity() + assert a == b, "the instance id must be stable for the life of the process" + assert len(instanceid.INSTANCE_ID) == 12 + blob = repr(a) + # It must not leak anything about the box it runs on. + for leaky in (os.getcwd(), os.environ.get("HOME", "/root")): + if leaky: + assert leaky not in blob + + +def test_store_revision_is_monotonic_across_events(tmp_path): + s = Store(path=str(tmp_path / "guild.json")) + seen = [s.revision] + for i in range(5): + s.record_event(None, "query", ua=f"probe/{i}") + seen.append(s.revision) + assert seen == sorted(seen), "revision must never go backwards" + assert seen[-1] > seen[0] + + +def test_state_diagnostics_reports_agreement_under_sqlite(tmp_path): + s = _sqlite_store(tmp_path) + s.record_event(None, "query", ua="probe/1") + d = s.state_diagnostics() + assert d["instance"] == instanceid.INSTANCE_ID + assert d["store_mode"] == "sqlite" + assert d["divergence"] == [], d + assert d["consistent"] is True + assert d["in_memory"]["events"] == d["durable"]["events"] + # no paths, no secrets + assert str(tmp_path) not in repr(d) + + +def test_state_diagnostics_DETECTS_a_stale_in_memory_view(tmp_path): + """THE detector. Two Stores over one SQLite file: one writes, the other's + in-memory view is now behind the database it shares. The ops pass must be + able to SEE that, rather than being told to discard the first few reads.""" + writer = _sqlite_store(tmp_path) + reader = _sqlite_store(tmp_path) # hydrated at its own boot + before = reader.state_diagnostics() + assert before["divergence"] == [] + + for i in range(3): + writer.record_event(None, "query", ua=f"late/{i}") + + after = reader.state_diagnostics() + assert "durable_events_ahead_of_in_memory" in after["divergence"] + assert after["consistent"] is False + assert after["durable"]["events"] > after["in_memory"]["events"] + + +# -------------------------------------------------------------------------- +# 2. Canonical writes fail CLOSED +# -------------------------------------------------------------------------- +def test_publish_refuses_when_durable_head_is_behind_observed_head(tmp_path): + """The exact production signature: a publish computed from a view two + entries behind the committed feed. It must refuse, not publish.""" + s = _sqlite_store(tmp_path) + s.record_event(None, "query", ua="seed/1") + first = s.publish_checkpoint() + assert first["index"] == 0 + + # Simulate the observed condition: this process has seen head 5 published, + # but the authoritative feed only has head 0. + s.checkpoints = list(s.checkpoints) + [ + {"index": 5, "published_at": "2026-07-31T00:00:00+00:00", + "ledger_length": 999, "checkpoint": {"head_hash": "deadbeef"}}] + + with pytest.raises(StaleDurableStateError) as exc: + s.publish_checkpoint() + assert "BEHIND" in str(exc.value) + assert exc.value.code == "stale_durable_state" + # and nothing was written + assert s.backend.durable_counts()["checkpoints"] == 1 + + +def test_publish_refuses_a_short_durable_ledger(tmp_path): + s = _sqlite_store(tmp_path) + s.record_event(None, "query", ua="seed/1") + s.publish_checkpoint() + # in-memory ledger claims more records than are committed + s.ledger_records = list(s.ledger_records) + [{"seq": 10 ** 6, "fake": True}] + with pytest.raises(StaleDurableStateError) as exc: + s.publish_checkpoint() + assert "SHORTER" in str(exc.value) + + +def test_publish_refuses_to_overwrite_an_existing_index_fork(tmp_path): + """Two publishers racing on the same next index used to be an INSERT OR + REPLACE — silently replacing a commitment a third party may already hold.""" + s = _sqlite_store(tmp_path) + s.record_event(None, "query", ua="seed/1") + entry = s.publish_checkpoint() + with pytest.raises(CheckpointForkError): + s.backend.insert_checkpoint_strict(dict(entry)) + assert s.backend.durable_counts()["checkpoints"] == 1 + + +def test_publish_is_idempotent_when_no_evidence_landed(tmp_path): + s = _sqlite_store(tmp_path) + s.record_event(None, "query", ua="seed/1") + a = s.publish_checkpoint() + b = s.publish_checkpoint() + c = s.publish_checkpoint() + assert a["index"] == b["index"] == c["index"] + assert s.backend.durable_counts()["checkpoints"] == 1 + + +def test_publish_read_after_write_verifies_the_stored_bytes(tmp_path): + s = _sqlite_store(tmp_path) + s.record_event(None, "query", ua="seed/1") + entry = s.publish_checkpoint() + stored = s.backend.checkpoint_at(entry["index"]) + assert stored == entry, "the feed must read back byte-identical" + + +def test_publish_reports_unverified_write_instead_of_success(tmp_path, monkeypatch): + """If the row cannot be read back, the publish must NOT be reported as + published — the failure mode that turns a missing commitment into a + confident lie.""" + s = _sqlite_store(tmp_path) + s.record_event(None, "query", ua="seed/1") + monkeypatch.setattr(s.backend, "checkpoint_at", lambda idx: None) + with pytest.raises(CheckpointWriteVerificationError): + s.publish_checkpoint() + + +def test_next_index_comes_from_max_index_not_list_length(tmp_path): + """A feed with a gap must not re-issue an index that already exists.""" + s = _sqlite_store(tmp_path) + s.record_event(None, "query", ua="seed/1") + s.publish_checkpoint() # index 0 + # a hand-repaired feed with a gap (indices 0 and 4 present, length 2) + gap = {"index": 4, "published_at": "2026-07-31T00:00:00+00:00", + "ledger_length": 1, "checkpoint": {"head_hash": "x"}} + s.backend.insert_checkpoint_strict(gap) + s.checkpoints = s.backend.all_checkpoints() + assert len(s.checkpoints) == 2 and s.checkpoints[-1]["index"] == 4 + # land new evidence so the publish is not the idempotent no-op + s.record_collaboration_record({"kind": "test_evidence", "n": 1}) \ + if hasattr(s, "record_collaboration_record") else None + s.ledger_records = s.backend.all_ledger() + entry = s.publish_checkpoint() + # len()-based indexing would have produced 2 and REPLACED nothing but + # broken continuity; max()-based indexing produces 5. + assert entry["index"] == 5, entry["index"] + assert [e["index"] for e in s.backend.all_checkpoints()] == [0, 4, 5] + + +def test_concurrent_publishers_never_fork_the_feed(tmp_path): + """Shared-store concurrency: many threads publishing at once must produce a + strictly increasing, gap-free, non-overwritten feed — or refuse.""" + s = _sqlite_store(tmp_path) + errors: list[Exception] = [] + + def worker(i: int) -> None: + try: + s.record_event(None, "query", ua=f"racer/{i}") + s.publish_checkpoint() + except CanonicalWriteRefused as exc: # refusing is an ACCEPTED outcome + errors.append(exc) + + threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)] + for t in threads: + t.start() + for t in threads: + t.join() + + feed = s.backend.all_checkpoints() + idxs = [e["index"] for e in feed] + assert idxs == sorted(set(idxs)), f"feed forked or duplicated: {idxs}" + assert idxs == list(range(len(idxs))), f"feed has a gap: {idxs}" + # every entry commits to its predecessor + for prev, cur in zip(feed, feed[1:]): + assert cur.get("prev_entry_sha256"), "continuity commitment missing" + + +# -------------------------------------------------------------------------- +# 3. Transaction-depth hygiene (thread-local connection poisoning) +# -------------------------------------------------------------------------- +def test_nested_transaction_rollback_does_not_poison_the_thread(tmp_path): + """A nested txn that raises used to leave this THREAD's depth at -1, after + which every later 'transaction' silently ran in autocommit — losing + atomicity for the life of the process, on that thread only.""" + s = _sqlite_store(tmp_path) + b = s.backend + try: + with b.transaction(): + try: + with b.transaction(): + raise ValueError("inner blows up") + except ValueError: + pass + except Exception: + pass + assert getattr(b._local, "depth", 0) >= 0, "depth went negative" + # and a real transaction still opens afterwards + with b.transaction(): + assert b.in_transaction() is True + assert b.in_transaction() is False + # writes still land + s.record_event(None, "query", ua="after/1") + assert s.backend.durable_counts()["events"] >= 1 diff --git a/live/guild/tests/test_truth_layer_invariants.py b/live/guild/tests/test_truth_layer_invariants.py new file mode 100644 index 0000000..4c3d67c --- /dev/null +++ b/live/guild/tests/test_truth_layer_invariants.py @@ -0,0 +1,248 @@ +"""Truth-layer invariants — corrective pass 2026-07-31. + +Three metric defects reached production and each of them made the numbers read +BETTER than reality: + + 1. A passport LOOKUP MISS counted as an issuance, and one successful MCP call + counted twice — so a schema probe took "genuine external passports issued" + from 1 to 3 with no agent behind it. + 2. Aggregate stage activity was reported as a conversion funnel, so + "0 followed / 1,790 served" read as a 0% conversion rate when 1,787 of + those serves were unattributable crawlers and the qualified denominator + was 1. + 3. ``/self-eval`` printed "FLYWHEEL TURNING — external agents pay" and a + dollar figure derived from sandbox credits, while verified external + revenue was $0.00. + +These tests lock the corrections. They are deliberately written as assertions +about what the numbers MAY NOT claim, not about their current values. +""" +from __future__ import annotations + +import os +import sys + +import pytest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from app.store import Store # noqa: E402 + + +def _tool_fn(tool): + """Resolve the plain callable behind an MCP tool across fastmcp versions.""" + for attr in ("fn", "func", "__wrapped__"): + f = getattr(tool, attr, None) + if callable(f): + return f + return tool + + +@pytest.fixture() +def store(tmp_path) -> Store: + return Store(path=str(tmp_path / "guild.json")) + + +def _register(store: Store, name: str, first_party: bool = False) -> tuple[str, str]: + rec = store.register_agent(name, ["translation"], {}) + agent_id = rec.get("id") or rec.get("agent_id") + if first_party: + store.agents[agent_id]["first_party"] = True + return agent_id, rec.get("api_key", "") + + +# -------------------------------------------------------------------------- +# 1. Passport telemetry semantics +# -------------------------------------------------------------------------- +def test_failed_passport_lookup_is_never_counted_as_an_issuance(store): + """The exact 2026-07-30 probe behaviour: ask for a passport that cannot be + produced. That is DEMAND, and must never appear as issuance.""" + before = sum(1 for e in store.events if e["type"] == "passport_issued") + cred = store.issue_passport("agent_does_not_exist", actor_key="mcp", + surface="mcp", ua="mcp:schema-probe/0.1") + after = sum(1 for e in store.events if e["type"] == "passport_issued") + assert cred is None + assert after == before, "a lookup miss was recorded as a passport issuance" + + +def test_successful_issuance_emits_exactly_one_event(store): + agent_id, _ = _register(store, "subject-a") + store.issue_passport(agent_id, actor_key="mcp", surface="mcp", + ua="mcp:client/1", request_id="req-1") + issued = [e for e in store.events if e["type"] == "passport_issued"] + assert len(issued) == 1, f"expected exactly one event, got {len(issued)}" + ev = issued[0] + assert ev["subject_id"] == agent_id + assert ev["transport"] == "mcp" + assert ev["request_id"] == "req-1" + assert "self_claim" in ev + + +def test_mcp_tool_does_not_double_count(store, monkeypatch): + """Regression for the double-record: the MCP path recorded on ENTRY and + Store.issue_passport recorded again on success.""" + import app.state as state + import app.mcp_server as mcp_server + + monkeypatch.setattr(state, "store", store) + monkeypatch.setattr(mcp_server, "store", store) + agent_id, _ = _register(store, "subject-b") + + fn = _tool_fn(mcp_server.guild_passport) + fn(agent_id=agent_id, ctx=None) + + issued = [e for e in store.events if e["type"] == "passport_issued"] + requested = [e for e in store.events if e["type"] == "passport_requested"] + assert len(issued) == 1, "the MCP path double-counted the issuance" + assert len(requested) == 1, "the attempt must be recorded separately" + + +def test_mcp_miss_records_a_failure_not_an_issuance(store, monkeypatch): + import app.state as state + import app.mcp_server as mcp_server + + monkeypatch.setattr(state, "store", store) + monkeypatch.setattr(mcp_server, "store", store) + + out = _tool_fn(mcp_server.guild_passport)(agent_id="nope", ctx=None) + assert "error" in out + assert not [e for e in store.events if e["type"] == "passport_issued"] + assert [e for e in store.events if e["type"] == "passport_issue_failed"] + + +def test_passport_activity_never_reports_one_adoption_number(store): + agent_id, _ = _register(store, "subject-c") + store.issue_passport(agent_id, actor_key="third-party-key", surface="http") + act = store.passport_activity() + assert set(act["behaviours"]) == { + "subject_self_claim", "third_party_fetch", + "third_party_verification", "subject_evidence_attached"} + # a third party fetching someone else's credential is NOT a self claim + assert act["behaviours"]["subject_self_claim"]["external"] == 0 + blob = repr(act).lower() + assert "adoption" not in blob or "not" in blob + + +def test_third_party_fetch_is_not_a_self_claim(store): + """A probe pulling another agent's public credential must never be counted + as that agent adopting one.""" + subject, _ = _register(store, "subject-d") + store.issue_passport(subject, actor_key="a2a:net:deadbeef", surface="a2a") + ev = [e for e in store.events if e["type"] == "passport_issued"][-1] + assert ev["self_claim"] is False + + +# -------------------------------------------------------------------------- +# 2. Qualified cohort funnel +# -------------------------------------------------------------------------- +def test_qualified_funnel_excludes_crawlers_and_first_party(store): + store.record_event("a2a:net:crawler", "offer_served", ua="a2a:AgenstryBot/0.3.0", + offer="passport", endpoint="agent_card") + store.record_event("ag-internal", "offer_served", ua="guild-release-gate", + offer="passport", endpoint="agent_card") + q = store.qualified_passport_funnel() + assert q["cohort"]["qualified_distinct_actors"] == 0 + assert "crawler" in q["excluded"].lower() + + +def test_qualified_funnel_deduplicates_repeat_exposure(store): + """800 hits from one bot is ONE exposure, not 800 trials.""" + for _ in range(50): + store.record_event("a2a:net:abc123", "offer_served", + ua="a2a:SomeAgent/1.0", offer="passport", + endpoint="agent_card") + q = store.qualified_passport_funnel() + assert q["cohort"]["raw_qualified_serves"] >= q["cohort"][ + "qualified_deduplicated_exposures"] + + +def test_zero_denominator_reports_not_measurable_never_zero_percent(store): + """The headline correction: with no qualified exposure we must say the rate + is NOT MEASURABLE, never '0% conversion'.""" + q = store.qualified_passport_funnel() + nb = q["next_boundary"] + assert nb["measurable"] is False + assert "not measurable" in nb["reason"].lower() + assert "rate" not in nb + + +def test_anonymous_exposure_is_unlinkable_not_a_failed_conversion(store): + for _ in range(5): + store.record_event(None, "offer_served", ua="a2a:Anon/1.0", + offer="passport", endpoint="llms_txt") + q = store.qualified_passport_funnel() + assert q["cohort"]["anonymous_unlinkable_serves"] >= 0 + assert "unlinkable" in q["honesty"].lower() + + +def test_small_sample_is_labelled_an_anecdote(store): + """A boundary measured on a single-digit denominator must say so.""" + store.record_event("a2a:net:solo", "offer_served", ua="a2a:Solo/1.0", + offer="passport", endpoint="agent_card") + store.record_event("a2a:net:solo", "register", ua="a2a:Solo/1.0") + q = store.qualified_passport_funnel() + nb = q["next_boundary"] + if nb.get("measurable") and nb.get("n", 0) < 10: + assert "ANECDOTE" in nb.get("sample_adequacy", "") + + +def test_raw_stages_are_labelled_as_not_a_conversion_funnel(store): + f = store.passport_funnel() + assert "qualified" in f + assert "not a conversion funnel" in f["reading_guide"].lower() + + +# -------------------------------------------------------------------------- +# 3. Self-evaluation honesty +# -------------------------------------------------------------------------- +def test_health_never_reports_sandbox_credits_as_usd_revenue(store): + v = store._health_vector() + assert "revenue_usd_external" not in v, ( + "the sandbox-credit dollar line must not exist under a revenue name") + assert v["verified_external_revenue_usd"] == 0.0 + assert "sandbox_credits_spent_external_NOT_MONEY" in v + + +def test_health_utility_uses_the_production_block(store): + v = store._health_vector() + assert "production_measured_lift" in v + assert "production_n_recommended" in v + # the mixed/seeded number may exist, but only under a name that cannot be + # quoted as a production result + assert "measured_lift" not in v or v.get("mixed_bootstrap_lift_NOT_PRODUCTION") is not None + + +def test_flywheel_verdict_requires_adoption_AND_verified_revenue(store): + base = {"agents_external": 5, "external_querying_agents": 3, + "external_repeat_query_agents": 2, "external_paid_queries": 99, + "adoption_grade_external_self_claims": 0, + "verified_external_revenue_usd": 0.0} + + # sandbox paid reads alone must NOT produce a flywheel verdict + assert "FLYWHEEL" not in Store._verdict(dict(base), {}) + + # adoption without money + v = dict(base, adoption_grade_external_self_claims=2) + assert "FLYWHEEL" not in Store._verdict(v, {}) + assert "ADOPTION WITHOUT REVENUE" in Store._verdict(v, {}) + + # money without adoption + v = dict(base, verified_external_revenue_usd=25.0) + assert "FLYWHEEL" not in Store._verdict(v, {}) + assert "REVENUE WITHOUT ADOPTION" in Store._verdict(v, {}) + + # both, and moving + v = dict(base, adoption_grade_external_self_claims=2, + verified_external_revenue_usd=25.0) + out = Store._verdict(v, {"verified_external_revenue_usd": 25.0}) + assert "FLYWHEEL TURNING" in out + + +def test_verdict_names_the_missing_half_when_nothing_is_proven(store): + v = {"agents_external": 5, "external_querying_agents": 3, + "external_repeat_query_agents": 2, "external_paid_queries": 99, + "adoption_grade_external_self_claims": 0, + "verified_external_revenue_usd": 0.0} + out = Store._verdict(v, {}) + assert "NOT evidence" in out or "not evidence" in out.lower() + assert "$0.00" in out diff --git a/live/scripts/detect_divergence.py b/live/scripts/detect_divergence.py new file mode 100755 index 0000000..e12c5d4 --- /dev/null +++ b/live/scripts/detect_divergence.py @@ -0,0 +1,182 @@ +#!/usr/bin/env python3 +"""Detect a divergent production view — and say WHICH kind it is. + +The daily ops pass previously worked around the 2026-07-30/31 incident by +"discarding the first 2-3 reads until three agree". That is a reporting +workaround: it hides the symptom, produces no evidence, and does nothing for +the WRITE path, where a stale view was willing to build a canonical checkpoint. + +This script replaces the workaround with a measurement. It fans out concurrent +reads and uses the view-identity headers shipped with the fix: + + X-Guild-Instance random per-process id + X-Guild-Boot process start time + X-Guild-Store-Rev monotonic in-memory mutation counter + +and the ``/diagnostics/state`` endpoint (in-memory vs authoritative SQLite). + +VERDICTS + consistent one instance, monotonic revisions, memory == durable + split_origin MORE THAN ONE instance id for one release SHA + -> two serving processes. SQLite on a Render disk is + single-writer: this is a topology emergency, not a + metrics bug. + stale_in_process one instance, but a response carried a LOWER store_rev + than one already seen from that same instance + -> the process served a frozen view of its own state. + memory_durable_split /diagnostics/state reports the in-memory view and the + committed database disagree. + intermediary bodies disagree while instance AND store_rev are + identical -> something in front of the app served a + body this process did not just produce. + +Exit code 0 = consistent, 2 = divergence detected, 1 = could not measure. + + python3 detect_divergence.py --url https://agent-guild-5d5r.onrender.com +""" +from __future__ import annotations + +import argparse +import concurrent.futures +import json +import sys +import time +import urllib.error +import urllib.request + +READ_PATHS = ["/instrumentation", "/funnel/passports", "/ledger/checkpoints?limit=1", + "/diagnostics/state", "/release", "/health"] + + +def _get(base: str, path: str, timeout: float = 25.0): + sep = "&" if "?" in path else "?" + url = f"{base}{path}{sep}cb={int(time.time() * 1e6)}" + req = urllib.request.Request(url, headers={"accept": "application/json"}) + try: + with urllib.request.urlopen(req, timeout=timeout) as resp: + body = json.loads(resp.read().decode()) + h = {k.lower(): v for k, v in resp.headers.items()} + return {"path": path, "body": body, + "instance": h.get("x-guild-instance"), + "boot": h.get("x-guild-boot"), + "rev": int(h["x-guild-store-rev"]) + if h.get("x-guild-store-rev", "").isdigit() else None} + except (urllib.error.HTTPError, urllib.error.URLError, + TimeoutError, ValueError) as exc: + return {"path": path, "error": type(exc).__name__} + + +def _counter(body, path): + """One comparable scalar per endpoint — what actually flapped in the incident.""" + if not isinstance(body, dict): + return None + if path.startswith("/instrumentation"): + return body.get("total_events") + if path.startswith("/funnel/passports"): + stages = body.get("stages") or [] + row = next((s for s in stages if s.get("stage") == "offer_served"), None) + return (row or {}).get("total") + if path.startswith("/ledger/checkpoints"): + cps = body.get("checkpoints") or [] + return (cps[0].get("ledger_length") if cps else None) + if path.startswith("/release"): + return body.get("git_sha") + return None + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument("--url", default="https://agent-guild-5d5r.onrender.com") + ap.add_argument("--rounds", type=int, default=4) + ap.add_argument("--concurrency", type=int, default=8) + ap.add_argument("--json", action="store_true") + args = ap.parse_args() + base = args.url.rstrip("/") + + jobs = [(base, p) for _ in range(args.rounds) for p in READ_PATHS] + with concurrent.futures.ThreadPoolExecutor(args.concurrency) as ex: + results = list(ex.map(lambda j: _get(*j), jobs)) + + ok = [r for r in results if "error" not in r] + if not ok: + print("[divergence] could not read the service at all", file=sys.stderr) + return 1 + + instances = {r["instance"] for r in ok if r.get("instance")} + findings: list[str] = [] + + if not instances: + findings.append( + "no_view_identity: responses carry no X-Guild-Instance — this build " + "predates the divergence fix, so the cause CANNOT be decided") + + if len(instances) > 1: + findings.append( + f"split_origin: {len(instances)} distinct instance ids " + f"({sorted(instances)}) for one URL — more than one serving " + "process. SQLite lives on a single-mount disk; two writers is a " + "topology emergency") + + # per-instance revision monotonicity, in observation order + high: dict[str, int] = {} + for r in ok: + inst, rev = r.get("instance"), r.get("rev") + if inst is None or rev is None: + continue + if rev < high.get(inst, -1): + findings.append( + f"stale_in_process: instance {inst} served store_rev {rev} " + f"after already serving {high[inst]} — a frozen view of its " + "own state") + high[inst] = max(high.get(inst, -1), rev) + + # counter disagreement per endpoint + per_path: dict[str, set] = {} + for r in ok: + c = _counter(r.get("body"), r["path"]) + if c is not None: + per_path.setdefault(r["path"], set()).add(c) + flapping = {p: sorted(v) for p, v in per_path.items() if len(v) > 1} + + # /diagnostics/state + diag = next((r["body"] for r in ok + if r["path"].startswith("/diagnostics/state") + and isinstance(r.get("body"), dict)), None) + if diag and diag.get("divergence"): + findings.append( + f"memory_durable_split: {diag['divergence']} " + f"(in_memory={diag.get('in_memory')}, durable={diag.get('durable')})") + + if flapping and len(instances) <= 1 and not any( + f.startswith("stale_in_process") for f in findings): + findings.append( + f"intermediary: counters disagree {flapping} while the instance id " + "and store_rev are stable — the differing body was NOT produced by " + "this process's current state; suspect something in front of the app") + + out = { + "url": base, + "samples": len(ok), + "instances": sorted(instances), + "flapping_counters": flapping, + "diagnostics": diag, + "findings": findings, + "verdict": "consistent" if not findings else "divergent", + } + if args.json: + print(json.dumps(out, indent=2)) + else: + print(f"=== divergence check @ {base} ({len(ok)} samples) ===") + print(f" instances: {sorted(instances) or 'NONE (no view identity)'}") + print(f" flapping counters: {flapping or 'none'}") + if diag: + print(f" in_memory: {diag.get('in_memory')}") + print(f" durable: {diag.get('durable')}") + for f in findings: + print(f" ! {f}") + print(f" VERDICT: {out['verdict']}") + return 0 if not findings else 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/live/scripts/self_eval_tick.py b/live/scripts/self_eval_tick.py index a758bb7..911edca 100644 --- a/live/scripts/self_eval_tick.py +++ b/live/scripts/self_eval_tick.py @@ -38,13 +38,28 @@ # Fields we surface in the printed summary, in order: (key, label, formatter). DISPLAY = [ ("verdict", None, None), - ("measured_lift", "utility · measured_lift", lambda v: f"{v:+.3f}" if isinstance(v, (int, float)) else "n/a"), - ("agents_external", "growth · external agents", str), + # UTILITY — production only. The mixed/seeded bootstrap lift is NOT printed + # as a result; a lift with n_recommended == 0 prints "not measurable" + # rather than a number we would otherwise be quoting from seeded data. + ("production_measured_lift", "utility · production lift", + lambda v: f"{v:+.3f}" if isinstance(v, (int, float)) else "not measurable"), + ("production_n_recommended", "utility · production n_recommended", str), + # GROWTH — adoption-grade external activity, not "records lacking + # first_party" (which counts our own untagged tooling and crawlers). + ("adoption_grade_external_self_claims", + "growth · external agents holding their OWN credential", str), ("external_querying_agents", "growth · external actors querying", str), ("external_repeat_query_agents", "retention · repeat-query", str), ("external_repeat_paid_agents", "retention · repeat-paid", str), - ("external_paid_queries", "revenue · paid reads", str), - ("revenue_usd_external", "revenue · USD", lambda v: f"${v}"), + # ECONOMIC VALUE — independently confirmed external mainnet settlement is + # the ONLY revenue line. Sandbox credits print as credits, never as money. + ("verified_external_revenue_usd", "revenue · VERIFIED external (USD)", + lambda v: f"${float(v):.2f}"), + ("cryptographically_bound_machine_revenue_usd", + "revenue · bound-machine, ownership unproven (USD)", + lambda v: f"${float(v):.2f}"), + ("sandbox_credits_spent_external_NOT_MONEY", + "sandbox · credits spent (NOT money)", lambda v: f"{v} credits"), ("total_referrals", "referrals · total", str), ("activated_referrals", "referrals · activated", str), ] @@ -60,13 +75,18 @@ def _get(url: str, timeout: float = 25.0): def _fallback_verdict(v: dict) -> str: + """Same gate as the server verdict: a positive read requires BOTH an + adoption-grade external credential holder AND verified external mainnet + revenue. Sandbox credits never qualify (corrective pass 2026-07-31).""" + revenue = float(v.get("verified_external_revenue_usd") or 0.0) if v["external_querying_agents"] == 0: return "NO EXTERNAL DISCOVERY YET — no agent we don't operate has queried the trust layer." if v["external_repeat_query_agents"] == 0: return "REACH BUT NO RETENTION — queried once, none came back." - if v["external_paid_queries"] == 0: - return "RETENTION BUT NO WILLINGNESS-TO-PAY — agents return for free reads but none pay." - return "WILLINGNESS-TO-PAY PRESENT — external agents return and pay; watch the trend." + if revenue <= 0: + return ("NO VERIFIED EXTERNAL REVENUE — sandbox credits and first-party " + "canaries are not money; willingness-to-pay is unproven.") + return f"VERIFIED EXTERNAL REVENUE ${revenue:.2f} — watch the trend." def fallback_snapshot(base: str) -> dict: @@ -77,11 +97,16 @@ def fallback_snapshot(base: str) -> dict: ev = _get(f"{base}/evaluation") or {} agents = _get(f"{base}/agents") or [] refs = _get(f"{base}/referrals") or {} + rev = _get(f"{base}/billing/revenue") or {} paid = ext.get("paid_query", 0) v = { "at": datetime.now(timezone.utc).isoformat(), "source": "fallback", - "measured_lift": ev.get("lift"), + # production block only — never the mixed/bootstrap top-level lift + "production_measured_lift": (ev.get("production") or {}).get("lift"), + "production_n_recommended": int( + ((ev.get("production") or {}).get("n_recommended")) or 0), + "mixed_bootstrap_lift_NOT_PRODUCTION": ev.get("lift"), "recommended_success_rate": ev.get("recommended_success_rate"), "agents_total": len(agents) if isinstance(agents, list) else 0, "agents_external": ext.get("unique_agents", 0), @@ -89,8 +114,16 @@ def fallback_snapshot(base: str) -> dict: "external_repeat_query_agents": ext.get("repeat_query", 0), "external_repeat_paid_agents": ext.get("repeat_paid_query_agents", 0), "external_paid_queries": paid, - "credits_spent_external": None, - "revenue_usd_external": round(paid * 10 * CREDIT_USD, 4), + "sandbox_credits_spent_external_NOT_MONEY": None, + # Revenue can ONLY come from independently confirmed external mainnet + # settlement. The old line multiplied sandbox paid-read counts by a + # notional rate and printed it as USD — that was inventing money. + "verified_external_revenue_usd": float( + ((rev.get("real_settlement") or {}).get( + "independently_attested_external_revenue_usd")) or 0.0), + "cryptographically_bound_machine_revenue_usd": float( + ((rev.get("real_settlement") or {}).get( + "cryptographically_bound_machine_revenue_usd")) or 0.0), "total_referrals": refs.get("total_referrals", 0), "activated_referrals": refs.get("activated_referrals", 0), }