diff --git a/README.md b/README.md index 1a9bdcca..1280764e 100644 --- a/README.md +++ b/README.md @@ -68,6 +68,33 @@ Delivery order, from [PRD §14](docs/PRD.md#14-delivery-phases): Documentation lived in a separate `mergit-docs` repository until October 2026 and now lives here under `docs/`. +## ETHGlobal Mumbai — Arkiv track + +Continuity Track pick: Arkiv's ["Build on Arkiv (existing projects)"](https://ethglobal.com/events/mumbai/prizes/arkiv). +Mergit's router already picks agents for a task and, on an exploring call, records the draw in its +own Postgres (`routing_decisions`, pre-existing); what's new at this event is a second, public copy +of each exploring decision written to Arkiv's Rabdi testnet and read back through a new endpoint. +Full writeup: [`arkiv/schema.md`](arkiv/schema.md), [`arkiv/why-arkiv.md`](arkiv/why-arkiv.md), +[`arkiv/how-you-use-arkiv.md`](arkiv/how-you-use-arkiv.md), [`arkiv/reach.md`](arkiv/reach.md), +[`arkiv/friction.md`](arkiv/friction.md). + +- **Service wallet (writes every mirrored entity):** `` +- **One query anyone can run**, against `https://rpc.rabdi.db-chain.testnet.arkiv.network` + (`arkiv_query`) or through the hosted app's `GET /v0/plans/{uuid}/arkiv-suggestions`: + ``` + arkiv.schema = str('mergit.routing_decision') AND planId = str('') AND $owner = addr() + ``` +- **In the app:** the live run screen's "Suggested agents" panel, after asking for an exploring + suggestion, shows a "View on Arkiv" button that reads the mirrored entities back from Arkiv live + (`frontend/src/screens/live/run/SuggestAgents.tsx`). +- **Writes go through a small sidecar** (`chain/arkiv/`, Node, the official JS SDK), not the Python + control plane directly — the only Python Arkiv SDK targets a different, incompatible network, + confirmed live (`arkiv/friction.md`). Reads stay plain Python (httpx against `arkiv_query`). +- **AI disclosure:** this track's implementation (`mergit_cp/arkiv_mirror.py`, the + `/arkiv-suggestions` route, the frontend panel, the `arkiv/*.md` files, this section) was written + with AI assistance (Claude Code), reviewed and tested by the team before merging. + ## Repository layout | Path | What | Language | @@ -76,7 +103,7 @@ under `docs/`. | `runtime/` | Receipt builder (`receipt/`) and its Python binding (`py/`, module `mergit_receipt`) | Rust | | `verifier/` | Receipt verifier library and `mergit-verify` CLI; `wasm/` builds it for the browser. Never imports `runtime/` | Rust | | `evidence/` | Receipt store (`store/`) and validator (`validator/`) | Rust | -| `chain/` | Contract client (`client/`), chain gateway with payment path and free credits (`gateway/`), generated ABIs (`abi/`) | Rust | +| `chain/` | Contract client (`client/`), chain gateway with payment path and free credits (`gateway/`), generated ABIs (`abi/`); the Arkiv write sidecar (`arkiv/`) is Node/JS, not Rust | Rust | | `control-plane/` | Planner, task graph, approvals, router, agent workers | Python | | `frontend/` | Web app | TypeScript, React | | `deploy/` | Docker images, local compose, hosting | | diff --git a/arkiv/friction.md b/arkiv/friction.md new file mode 100644 index 00000000..9a6a8733 --- /dev/null +++ b/arkiv/friction.md @@ -0,0 +1,69 @@ +# Arkiv feedback + +What got in the way building this, and how to reproduce each point. Everything below was checked +live against Rabdi (`https://rpc.rabdi.db-chain.testnet.arkiv.network`), not assumed from docs — +an earlier version of this file made claims from reading the Python SDK's source alone, which +turned out to be wrong for this network; this version corrects that. + +## The official Python SDK targets a different, incompatible network + +`arkiv-sdk` on PyPI (`Arkiv-Network/arkiv-sdk-python`) exists and looks like exactly what a Python +backend needs. Live, it doesn't work against Rabdi at all, in two separate ways: + +- Its query options use `includeData` (`{"select": ..., "includeData": {...}}` style params). Rabdi + rejects it outright: `"unknown field includeData, expected one of atBlock, select, limit, offset, + cursor"`. +- Its write target (contract/processor address) is `0x…61726B6976`. Rabdi's own write address, + confirmed by checking its code via `eth_getCode`, is `0x4400000000000000000000000000000000000044` + — the one address `eth_getCode` on Rabdi actually returns bytecode for; the SDK's address returns + `"0x"` (nothing deployed there on this chain). The newest prerelease (`1.0.0b3.dev0`) has migrated + the write address but still sends the old, rejected query options for reads. + +**Reproduce:** `pip install --pre arkiv-sdk`, point `ProviderBuilder().custom(...)` at Rabdi's RPC, +and run any `query_entities` call; the node returns the `includeData` error above. + +## Rabdi's own JSON-RPC has no payload/content-type field at all + +The docs' `select` field list (`key`, `owner`, `creator`, `createdAt`, `updatedAt`, `expiresAt`, +`creationFlags`, `attributes`, `attributeSchema`, `count`) looks complete until you look for where +the actual binary content goes. Checked live on dozens of real entities already on Rabdi +(`arkiv_getEntity`): none ever returns a top-level `payload` or `contentType` field, including ones +whose author clearly meant to store a binary blob (one real entity uses a custom attribute literally +*named* `payload`, since the protocol has none). Everything lives in `attributes`, including +anything you'd think of as "the payload" — hence `body` in our own schema is a `bytes`-typed +*attribute*, not the payload/content-type pair the JS quickstart's prose implies and the (wrong-network) +Python SDK's `create_entity(payload=..., content_type=...)` signature assumes. + +**Reproduce:** `arkiv_getEntity` any real key from a live `arkiv_query` result; the response's keys +are exactly the `select` list above, never `payload`. + +## The query language's literal syntax isn't what the JS quickstart's prose suggests + +The attribute-value syntax the raw JSON-RPC wants is stricter than a first read of the docs +suggests: string literals need `str('...')` (bare `"..."` is a parse error, `-32001`), and address +literals need **unquoted** hex — `addr(0xAbC…)`, not `addr('0xAbC…')` (quoted gives `-32003`, "addr +takes a 0x-prefixed address"). Neither is spelled out next to the `select`-field docs; both were +found by reading the actual error messages from a live node. + +**Reproduce:** `arkiv_query` with `'kind = "x"'` → `-32001`; with `$owner = addr('0x...')` → `-32003`. + +## A JSON-RPC reachability probe can silently ignore its own filter + +One specific sequence produced 60 real, unrelated entities from a query meant to match zero: an +`$owner`-only filter against a throwaway burn address (`0x…dEaD`) returned every entity anyone has +ever sent that address, which is correct — but it looked, for a few minutes, like the `kind`/schema +filter in a *different* query wasn't being applied, until re-running each query in isolation with a +`count`-only `select` showed the schema filter alone genuinely matches zero. Not an Arkiv bug: two +different test queries were run back to back and their results were briefly misread as one. Worth +recording anyway, since it's an easy trap when scripting several `arkiv_query` calls in a row and +reading output after the fact rather than per-call. + +## No live write exercised yet — but a real, verified attempt, not a guess + +`chain/arkiv/`'s sidecar, using the official JS SDK (the one confirmed above to target the right +network), reaches real on-chain gas estimation with an unfunded throwaway wallet (confirmed `0` +balance via `eth_getBalance`) and fails there — `"Execution error without revert data"` — not at +encoding, not at signing, not at the RPC connection. That is the furthest this can be verified +without a funded wallet, which needs a human with a browser wallet (`hub.arkiv.network/faucet` is +wallet-connect gated, not a plain API). No successful create yet, so no real entity key exists; the +placeholders in `schema.md` and the README stay placeholders until someone funds a wallet. diff --git a/arkiv/how-you-use-arkiv.md b/arkiv/how-you-use-arkiv.md new file mode 100644 index 00000000..5a7a4ba3 --- /dev/null +++ b/arkiv/how-you-use-arkiv.md @@ -0,0 +1,37 @@ +# How we use Arkiv + +**Do the things we filter on, who can edit each entity, and when it expires fit what the product +does?** + +- **Filters** (the built-in `arkiv.schema`, plus `planId`, `decisionId`, `taskIndex`, `category`, + `seed`, `budgetMicros`, `decidedAt` — all indexed, `schema.md`) are exactly the fields the product + itself already groups suggestions by: one query per plan (`ArkivMirror.describe_query` builds + `arkiv.schema = str('mergit.routing_decision') AND planId = str('...') AND $owner = addr('...')`), + which is the same shape as the existing Postgres query `decisions()` in the test suite uses + (`SELECT ... FROM routing_decisions WHERE plan_uuid = %s`). We didn't invent new query axes for + Arkiv; we reused the ones the feature already needed, plus `$owner`, which Arkiv being a public, + permissionless database makes necessary and Postgres's own access control never required. No + `requester`: review caught that it was never filtered on and would outlive a discarded draft, so + it's a Postgres-only column now. +- **Who can edit:** nobody, by design — we never call `update_entity`, `extend_entity`, or + `delete_entity`. A routing decision is a historical fact about one draw at one point in time + (paper §9.3: "any draw can be replayed from its seed and the posteriors it saw"); the matching + Postgres row is enforced append-only by a database trigger for the same reason + (`routing_decisions_append_only()`, migration `0016`). Arkiv's entity having a fixed owner and no + update path we exercise is a match for that invariant, not an extra feature we're using because + it's there — and `seed` plus `body`'s posterior snapshot are both on the entity now, so "any draw + can be replayed" is something the public copy can actually back, not just the Postgres original. +- **Expiry** (`ARKIV_ENTITY_TTL_DAYS`, default 30) fits the data's own lifetime: a suggestion is + advice for a plan that is still a *draft*. Once a requester approves, funds, and runs a plan, the + suggestion is behind them — Postgres keeps it forever as an audit trail because storage there is + cheap and ours; Arkiv keeps it only long enough to be checkable, then lets it go, which matches + "ephemeral coordination state... that [agents] let expire," the exact framing Arkiv's own + Build-on-Arkiv prize uses for this pattern. + +**Does each Arkiv feature we use change something users see?** +Yes, directly: `GET /v0/plans/{uuid}/arkiv-suggestions` is a new endpoint the frontend's +`SuggestAgents` panel calls through a "View on Arkiv" button, shown only after an exploring +suggestion (greedy writes nothing, so there's nothing to show it for). It renders the entity keys +read back live from Arkiv's query API — not a value cached in our own database — so what the user +sees is proof the mirror round-trips, not a write nobody reads back (the one thing this track's +qualification rules call out as not counted). diff --git a/arkiv/reach.md b/arkiv/reach.md new file mode 100644 index 00000000..76f5dc33 --- /dev/null +++ b/arkiv/reach.md @@ -0,0 +1,32 @@ +# Reach + +**Who uses this today, and will they use the part on Arkiv?** +Mergit's `/v0/plans/{uuid}/suggestions?explore=true` is already a live, used feature: a requester +drafting a plan can ask for an exploring (rather than greedy) ranking of which agent to run each +task, and that call already writes a replay row to our own Postgres today, independent of this +event. Every exploring call a requester makes now also gets mirrored to Arkiv — the same users, +the same action, no new UI flow for them to opt into. + +**Will it ship to them after Mumbai?** +Yes, assuming it holds up: nothing about the mirror is event-only scaffolding. `ArkivMirror` is a +best-effort side effect (`routes_routing._mirror_decisions` catches and logs any failure rather +than raising it), so leaving it on costs us nothing on the request path even if Arkiv's testnet has +an outage, and removing it later is a one-line settings change (`ARKIV_RPC_URL` unset), not a +migration. The honest caveat: Arkiv itself is currently testnet-only (see its own Terms of Use), so +"after Mumbai" for us currently means "until Arkiv has a production network," at which point moving +the service wallet over is the only change needed. + +**Who else would use a copy of this data on Arkiv?** +Two concrete audiences, neither hypothetical: + +- **Our own router's future work (paper §9.3, §7.5).** The reputation/routing paper already wants + an auditable, replayable record of every exploring draw — that's *why* `routing_decisions` is + append-only in Postgres today. A public copy is strictly more auditable than a private table only + we can read: anyone can verify a draw was recorded honestly and wasn't edited after the fact, + which a database we control cannot prove to an outside party. +- **Other agent-coordination projects on Arkiv.** This is a direct instance of the "ephemeral + coordination state that agents share, query and let expire" pattern Arkiv's own Build-on-Arkiv + prize names as a target use case. A team building on Arkiv to compare how different projects + model agent-assignment data would have a real, running example to query + (`arkiv.schema = str('mergit.routing_decision')`), not a synthetic demo entity nobody else + depends on. diff --git a/arkiv/schema.md b/arkiv/schema.md new file mode 100644 index 00000000..d2a1b1aa --- /dev/null +++ b/arkiv/schema.md @@ -0,0 +1,97 @@ +# Arkiv schema + +What we write to Arkiv's Rabdi testnet, for ETHGlobal Mumbai's "Build on Arkiv (existing projects)" +track. Code: `control-plane/mergit_cp/arkiv_mirror.py` (reads, plain httpx), `chain/arkiv/` (the +sidecar that writes, holding the service wallet's key), called from +`control-plane/mergit_cp/routes_routing.py` (`_mirror_decisions`) and read back in +`GET /v0/plans/{uuid}/arkiv-suggestions` (`arkiv_suggestions`). + +## What's new at this event, and what existed before + +Mergit's router already picked agents for a task before Mumbai: `GET /v0/plans/{uuid}/suggestions` +ranks the roster by validated success per unit cost (`routing.py`, paper §9.3), and — only when +asked to explore — records the draw in our own Postgres table `routing_decisions` (migration +`0016_routing_decisions.sql`) so it can be replayed from its seed. **That Postgres write, the +ranking logic, and the whole suggestions endpoint are pre-existing.** What's new at this event is +everything that follows: a second, public copy of each exploring decision written to Arkiv, and a +new endpoint that reads it back from there. + +## Entity + +One `arkiv.schema = "mergit.routing_decision"` entity per task per exploring +`GET /v0/plans/{uuid}/suggestions?explore=true` call, all of one call's tasks in a single batched +transaction (`chain/arkiv/src/server.js`'s `executeBatch`), not one transaction per task. + +`defineSchema("mergit.routing_decision", {...})` (`chain/arkiv/src/schema.js`) makes the schema's own +name the built-in `arkiv.schema` attribute — confirmed live against Rabdi: `arkiv.schema = +str('my-app.posts')` matches the docs' own example entity on the real network, so filtering on it is +a real query, not SDK sugar. + +| Attribute | Type | Indexed | What | +| --- | --- | --- | --- | +| `planId` | `str` | yes | The plan's `0x`-prefixed id (`mergit_receipt.plan_id`), so a query can ask for one plan's decisions | +| `decisionId` | `u64` | yes | The exact `routing_decisions.decision_id` this entity mirrors — links the public copy back to the real row, not just a record of the outcome | +| `taskIndex` | `i32` | yes | Which task in the plan this decision is for | +| `category` | `str` | yes | The task's category, same value the router ranked within | +| `seed` | `u64` | yes | The Thompson draw's seed (`routing_decisions.seed`) — with it, and `body`'s posterior snapshot, anyone can replay the draw and check it wasn't rigged (migration 0016's own reason the Postgres table is append-only) | +| `budgetMicros` | `u64` | yes | The task's budget in USDC micros, the same value the draw ranked candidates against | +| `decidedAt` | `str` | yes | ISO-8601 UTC timestamp, taken from Postgres's own `created_at` on the same row (not a second, separately-computed one) | +| `body` | `bytes` | no | The candidates/picks JSON, as raw bytes — see below | + +No `requester` attribute: an earlier version of this schema had one, dropped after review. Arkiv +entities are append-only — only expiry removes them — so attaching a wallet address to data that +outlives a requester's own "discard this draft" action was wrong, and nothing here ever filtered +reads by it anyway (`$owner`, the one service wallet, is what every read already filters on). + +There is no separate "payload" field at the protocol level on Rabdi: checked live, `arkiv_getEntity` +never returns one, on any entity, ours or anyone else's (`friction.md`). Everything is a named +attribute, including the blob, which is why `body` is itself an attribute (`bytes`-typed) rather than +a payload/content-type pair — the Python SDK's `create_entity(payload=..., content_type=...)` API, +which assumes that pair exists, targets a different, incompatible network (`friction.md`). + +`body`'s JSON: + +```json +{ + "candidates": [{"agent_id": "...", "alpha": 0, "beta": 0, "kappa": 0, "tau": 0, "n": 0}], + "picks": [{"agent_id": "...", "score": 0, "success": 0, "cost": 0}] +} +``` + +`candidates` is every roster agent's posterior at the moment of the draw (what `_snapshot` in +`routes_routing.py` already recorded to Postgres); `picks` is the ordered result. Same shapes +Postgres's `routing_decisions.candidates`/`.picks` columns hold — this is a mirror of that row, not +a redesign of it. + +## Who owns it, and why + +One service wallet signs every write. Its *private key* lives only in `chain/arkiv/`'s own env +(`ARKIV_PRIVATE_KEY`), never in the control plane's process or env — the control plane only knows +the wallet's *public* address (`ARKIV_OWNER_ADDRESS`), which it needs to filter reads (below). It is +not a per-requester wallet: Mergit requesters sign in with Ethereum but never hold funds on Arkiv's +testnet, so making them the Arkiv-level owner would mean asking them to fund a second chain to see +their own suggestion history. + +Service wallet address: ``. + +## Expiry + +`ARKIV_ENTITY_TTL_DAYS` (default 30) from first write. A routing suggestion is advice for one draft +plan; once that plan is approved, funded, run and done, the suggestion that preceded it has no +further use, so letting it expire rather than grow an unbounded public log is correct, not merely +cheap. + +## One query anyone can run + +``` +arkiv.schema = str('mergit.routing_decision') AND planId = str('') AND $owner = addr() ORDER BY $key ASC +``` + +The `$owner` clause matters: without it, anyone could write an entity with our exact schema tag and +a matching `planId` and have it show up as if it were ours (Arkiv is public and permissionless — +checked live, nothing stops a third party from using any tag). Run it against +`https://rpc.rabdi.db-chain.testnet.arkiv.network` via `arkiv_query` +(`{"select": {"key": true, "attributes": true}}`), or through +`GET /v0/plans/{uuid}/arkiv-suggestions` on the hosted app, which runs exactly this query +(`ArkivMirror.describe_query`) and returns the decoded result. diff --git a/arkiv/why-arkiv.md b/arkiv/why-arkiv.md new file mode 100644 index 00000000..ffc2fdb8 --- /dev/null +++ b/arkiv/why-arkiv.md @@ -0,0 +1,24 @@ +# Why Arkiv + +**If this data sat in Postgres or any other database instead of Arkiv, what would we lose?** +We wouldn't lose anything functionally — Postgres is still the copy the product actually runs on, +and stays that way (`arkiv_mirror.py`'s own docstring: "Postgres stays the source of truth: nothing +here is read back to decide anything the product depends on"). What we'd lose is *provability to +someone outside Mergit*. Our `routing_decisions` table is already append-only with a trigger that +refuses `UPDATE`/`DELETE` (migration `0016_routing_decisions.sql`) — but that guarantee only holds +inside our own database, enforced by us, for anyone who trusts us to have set it up honestly and not +touched it since. On Arkiv, the same guarantee (an entity with a fixed owner and a set expiry) is +enforced by a network we don't operate, so "this decision wasn't quietly edited after the fact" is +something a requester, a researcher, or a competing agent-coordination project can check themselves +against a public RPC, not something they have to take Mergit's word for. + +**What did we keep off Arkiv, and why?** +Everything except the exploring-suggestion mirror: plans, tasks, budgets, approvals, accounts, +money events, the reputation posteriors themselves. Those stay in Postgres because they're either +load-bearing for a real financial decision (escrow, budget checks, what an agent actually gets +paid) or carry information a requester has no reason to make public (their goal text, their repo +URL, their account). Arkiv is a public database — every entity's attributes are readable by anyone, +and `$owner` only controls who can *change* an entity, not who can *see* it. A routing suggestion is +advice that was already shown to the requester in the API response; mirroring exactly that is a +natural fit. Mirroring a plan's goal text or budget would be publishing data nobody asked to make +public, for no reach benefit this track's own judging criteria reward. diff --git a/chain/arkiv/.gitignore b/chain/arkiv/.gitignore new file mode 100644 index 00000000..3c3629e6 --- /dev/null +++ b/chain/arkiv/.gitignore @@ -0,0 +1 @@ +node_modules diff --git a/chain/arkiv/Dockerfile b/chain/arkiv/Dockerfile new file mode 100644 index 00000000..3ad77594 --- /dev/null +++ b/chain/arkiv/Dockerfile @@ -0,0 +1,17 @@ +# The Arkiv write sidecar (ETHGlobal Mumbai "Build on Arkiv" track): holds the service wallet's +# private key, never the control plane's own process. Build from this directory: +# docker build -t mergit-arkiv-sidecar:local chain/arkiv +FROM node:22-slim AS builder +WORKDIR /src +COPY package.json package-lock.json . +RUN npm ci --omit=dev --no-audit --no-fund + +FROM node:22-slim +RUN useradd --system --uid 10001 arkiv +WORKDIR /app +COPY --from=builder /src/node_modules node_modules +COPY package.json . +COPY src src +USER arkiv +EXPOSE 8088 +CMD ["node", "src/server.js"] diff --git a/chain/arkiv/README.md b/chain/arkiv/README.md new file mode 100644 index 00000000..f44f3782 --- /dev/null +++ b/chain/arkiv/README.md @@ -0,0 +1,55 @@ +# Arkiv write sidecar + +ETHGlobal Mumbai "Build on Arkiv" track. The one thing this exists for: signing and sending the write +Rabdi's current protocol needs — a batched `executeBatch` create of routing-decision entities — which +only the official JS SDK (`@arkiv-network/sdk`) demonstrably does correctly against Rabdi. The control +plane's own `arkiv_mirror.py` reads directly (plain `arkiv_query` over httpx, no SDK); only the write +needs this, and only this process ever holds the service wallet's private key. + +## Why a sidecar, not Python + +The only Python SDK (`arkiv-sdk` on PyPI) targets a different, incompatible Arkiv network — verified +live against Rabdi: its query options (`includeData`) and its write target (processor address) are both +rejected by Rabdi's node. See `../../arkiv/friction.md` for the exact RPC errors. The JS SDK's own write +target and query language both matched Rabdi live when checked the same way. + +## Running it + +``` +npm install +ARKIV_PRIVATE_KEY=0x... ARKIV_SIDECAR_TOKEN=... node src/server.js +``` + +Env vars: + +| Name | Required | What | +| --- | --- | --- | +| `ARKIV_PRIVATE_KEY` | yes | The service wallet's key. Fund it from `hub.arkiv.network/faucet`; get an access key from `hub.arkiv.network/access-keys`. Never the control plane's own env. | +| `ARKIV_SIDECAR_TOKEN` | yes | Shared secret the control plane sends as `Authorization: Bearer`; must match `ARKIV_SIDECAR_TOKEN` in `deploy/secrets/api.env`. | +| `ARKIV_ACCESS_KEY` | no | Raises the public RPC's heavy rate limit. | +| `PORT` | no | Default `8088`. | + +`GET /healthz` — liveness. `POST /internal/mirror` — the one real endpoint (see `src/server.js`'s own +comments for the body shape); everything else is `404`, and a wrong or missing bearer token is `401`. + +## Deploying it + +Gated by the `arkiv` compose profile (`deploy/compose.controlplane.yaml`), same opt-in pattern as +`keyrelay`: nothing starts until `deploy/secrets/arkiv.env` exists (create it with the two vars above) +and the profile is passed — + +``` +docker compose -f deploy/compose.testnet.yaml -f deploy/compose.controlplane.yaml --profile arkiv up -d --build +``` + +Until then, `ARKIV_SIDECAR_URL` in `api.env` points nowhere running, and every mirror write just fails +and gets logged — never a crash, same as any other Arkiv misconfiguration (`arkiv_mirror.py`'s own +docstring). + +## What's verified, and what isn't + +Checked live against Rabdi, from this sidecar's own code path: the schema (`src/schema.js`) builds and +sends; an unfunded test wallet's write reaches real on-chain gas estimation and fails there +("Execution error without revert data", on a wallet confirmed to hold 0 tGLM) — not at encoding, not at +signing, not at the RPC connection. That is as far as this can be verified without a funded wallet. No +successful create, and so no real entity key, exists yet. diff --git a/chain/arkiv/package-lock.json b/chain/arkiv/package-lock.json new file mode 100644 index 00000000..f1dbcd04 --- /dev/null +++ b/chain/arkiv/package-lock.json @@ -0,0 +1,247 @@ +{ + "name": "mergit-arkiv-sidecar", + "version": "0.1.0", + "lockfileVersion": 3, + "requires": true, + "packages": { + "": { + "name": "mergit-arkiv-sidecar", + "version": "0.1.0", + "dependencies": { + "@arkiv-network/sdk": "1.0.1", + "viem": "^2.21.0" + }, + "engines": { + "node": ">=22.12" + } + }, + "node_modules/@adraffy/ens-normalize": { + "version": "1.11.1", + "resolved": "https://registry.npmjs.org/@adraffy/ens-normalize/-/ens-normalize-1.11.1.tgz", + "integrity": "sha512-nhCBV3quEgesuf7c7KYfperqSS14T8bYuvJ8PcLJp6znkZpFc0AuW4qBtr8eKVyPPe/8RSr7sglCWPU5eaxwKQ==", + "license": "MIT" + }, + "node_modules/@arkiv-network/sdk": { + "version": "1.0.1", + "resolved": "https://registry.npmjs.org/@arkiv-network/sdk/-/sdk-1.0.1.tgz", + "integrity": "sha512-DCOFdGunn4I5nYrvIV/YIkMCkLZM7EQNSyNPtqpmQkP2JzTEfRLBJleBSB2rokWMa833/9P8DMiwsEXACr3CUQ==", + "engines": { + "node": ">=18.0.0" + }, + "peerDependencies": { + "typescript": ">=5.0.0", + "viem": "^2.0.0" + }, + "peerDependenciesMeta": { + "typescript": { + "optional": true + }, + "viem": { + "optional": false + } + } + }, + "node_modules/@noble/ciphers": { + "version": "1.3.0", + "resolved": "https://registry.npmjs.org/@noble/ciphers/-/ciphers-1.3.0.tgz", + "integrity": "sha512-2I0gnIVPtfnMw9ee9h1dJG7tp81+8Ob3OJb3Mv37rx5L40/b0i7djjCVvGOVqc9AEIQyvyu1i6ypKdFw8R8gQw==", + "license": "MIT", + "engines": { + "node": "^14.21.3 || >=16" + }, + "funding": { + "url": "https://paulmillr.com/funding/" + } + }, + "node_modules/@noble/curves": { + "version": "1.9.1", + "resolved": "https://registry.npmjs.org/@noble/curves/-/curves-1.9.1.tgz", + "integrity": "sha512-k11yZxZg+t+gWvBbIswW0yoJlu8cHOC7dhunwOzoWH/mXGBiYyR4YY6hAEK/3EUs4UpB8la1RfdRpeGsFHkWsA==", + "license": "MIT", + "dependencies": { + "@noble/hashes": "1.8.0" + }, + "engines": { + "node": "^14.21.3 || >=16" + }, + "funding": { + "url": "https://paulmillr.com/funding/" + } + }, + "node_modules/@noble/hashes": { + "version": "1.8.0", + "resolved": "https://registry.npmjs.org/@noble/hashes/-/hashes-1.8.0.tgz", + "integrity": "sha512-jCs9ldd7NwzpgXDIf6P3+NrHh9/sD6CQdxHyjQI+h/6rDNo88ypBxxz45UDuZHz9r3tNz7N/VInSVoVdtXEI4A==", + "license": "MIT", + "engines": { + "node": "^14.21.3 || >=16" + }, + "funding": { + "url": "https://paulmillr.com/funding/" + } + }, + "node_modules/@scure/base": { + "version": "1.2.6", + "resolved": "https://registry.npmjs.org/@scure/base/-/base-1.2.6.tgz", + "integrity": "sha512-g/nm5FgUa//MCj1gV09zTJTaM6KBAHqLN907YVQqf7zC49+DcO4B1so4ZX07Ef10Twr6nuqYEH9GEggFXA4Fmg==", + "license": "MIT", + "funding": { + "url": "https://paulmillr.com/funding/" + } + }, + "node_modules/@scure/bip32": { + "version": "1.7.0", + "resolved": "https://registry.npmjs.org/@scure/bip32/-/bip32-1.7.0.tgz", + "integrity": "sha512-E4FFX/N3f4B80AKWp5dP6ow+flD1LQZo/w8UnLGYZO674jS6YnYeepycOOksv+vLPSpgN35wgKgy+ybfTb2SMw==", + "license": "MIT", + "dependencies": { + "@noble/curves": "~1.9.0", + "@noble/hashes": "~1.8.0", + "@scure/base": "~1.2.5" + }, + "funding": { + "url": "https://paulmillr.com/funding/" + } + }, + "node_modules/@scure/bip39": { + "version": "1.6.0", + "resolved": "https://registry.npmjs.org/@scure/bip39/-/bip39-1.6.0.tgz", + "integrity": "sha512-+lF0BbLiJNwVlev4eKelw1WWLaiKXw7sSl8T6FvBlWkdX+94aGJ4o8XjUdlyhTCjd8c+B3KT3JfS8P0bLRNU6A==", + "license": "MIT", + "dependencies": { + "@noble/hashes": "~1.8.0", + "@scure/base": "~1.2.5" + }, + "funding": { + "url": "https://paulmillr.com/funding/" + } + }, + "node_modules/abitype": { + "version": "1.2.3", + "resolved": "https://registry.npmjs.org/abitype/-/abitype-1.2.3.tgz", + "integrity": "sha512-Ofer5QUnuUdTFsBRwARMoWKOH1ND5ehwYhJ3OJ/BQO+StkwQjHw0XyVh4vDttzHB7QOFhPHa/o413PJ82gU/Tg==", + "license": "MIT", + "funding": { + "url": "https://github.com/sponsors/wevm" + }, + "peerDependencies": { + "typescript": ">=5.0.4", + "zod": "^3.22.0 || ^4.0.0" + }, + "peerDependenciesMeta": { + "typescript": { + "optional": true + }, + "zod": { + "optional": true + } + } + }, + "node_modules/eventemitter3": { + "version": "5.0.1", + "resolved": "https://registry.npmjs.org/eventemitter3/-/eventemitter3-5.0.1.tgz", + "integrity": "sha512-GWkBvjiSZK87ELrYOSESUYeVIc9mvLLf/nXalMOS5dYrgZq9o5OVkbZAVM06CVxYsCwH9BDZFPlQTlPA1j4ahA==", + "license": "MIT" + }, + "node_modules/isows": { + "version": "1.0.7", + "resolved": "https://registry.npmjs.org/isows/-/isows-1.0.7.tgz", + "integrity": "sha512-I1fSfDCZL5P0v33sVqeTDSpcstAg/N+wF5HS033mogOVIp4B+oHC7oOCsA3axAbBSGTJ8QubbNmnIRN/h8U7hg==", + "funding": [ + { + "type": "github", + "url": "https://github.com/sponsors/wevm" + } + ], + "license": "MIT", + "peerDependencies": { + "ws": "*" + } + }, + "node_modules/ox": { + "version": "0.14.59", + "resolved": "https://registry.npmjs.org/ox/-/ox-0.14.59.tgz", + "integrity": "sha512-Y+DPEQScTEGKteQATjtytE1JRQ5cg1R2+DvYQ/lbhUCm/xS9n0Jnkkf2AqQyIntefQxs8xY6dZ+kyNZ7zf2FzA==", + "funding": [ + { + "type": "github", + "url": "https://github.com/sponsors/wevm" + } + ], + "license": "MIT", + "dependencies": { + "@adraffy/ens-normalize": "^1.11.0", + "@noble/ciphers": "^1.3.0", + "@noble/curves": "1.9.1", + "@noble/hashes": "^1.8.0", + "@scure/bip32": "^1.7.0", + "@scure/bip39": "^1.6.0", + "abitype": "^1.2.3", + "eventemitter3": "5.0.1" + }, + "peerDependencies": { + "typescript": ">=5.4.0" + }, + "peerDependenciesMeta": { + "typescript": { + "optional": true + } + } + }, + "node_modules/viem": { + "version": "2.57.16", + "resolved": "https://registry.npmjs.org/viem/-/viem-2.57.16.tgz", + "integrity": "sha512-+4uaUw3NfkqjaqsZp8Zci94K33oc9rv7mIUgvPduJxsY/tI0ASi2YkaXoUVSBGfdIw/QPWuVnIf9LTFf35IoMg==", + "funding": [ + { + "type": "github", + "url": "https://github.com/sponsors/wevm" + } + ], + "license": "MIT", + "dependencies": { + "@noble/curves": "1.9.1", + "@noble/hashes": "1.8.0", + "@scure/bip32": "1.7.0", + "@scure/bip39": "1.6.0", + "abitype": "1.2.3", + "isows": "1.0.7", + "ox": "0.14.59", + "ws": "8.21.0" + }, + "peerDependencies": { + "mppx": ">=0.17.0", + "typescript": ">=5.0.4" + }, + "peerDependenciesMeta": { + "mppx": { + "optional": true + }, + "typescript": { + "optional": true + } + } + }, + "node_modules/ws": { + "version": "8.21.0", + "resolved": "https://registry.npmjs.org/ws/-/ws-8.21.0.tgz", + "integrity": "sha512-Vsp28b7DRcimFQvrqu2Wek3z1iYxDCWqHYB8Qsnk/S4RfaCQzPGPyBNuVjJV3cd6UiKtUtp6sNM77gWvzcCH+g==", + "license": "MIT", + "engines": { + "node": ">=10.0.0" + }, + "peerDependencies": { + "bufferutil": "^4.0.1", + "utf-8-validate": ">=5.0.2" + }, + "peerDependenciesMeta": { + "bufferutil": { + "optional": true + }, + "utf-8-validate": { + "optional": true + } + } + } + } +} diff --git a/chain/arkiv/package.json b/chain/arkiv/package.json new file mode 100644 index 00000000..cd8b3651 --- /dev/null +++ b/chain/arkiv/package.json @@ -0,0 +1,15 @@ +{ + "name": "mergit-arkiv-sidecar", + "version": "0.1.0", + "private": true, + "description": "Signs and sends the one write Rabdi needs: a batched create of routing-decision entities. Holds the service wallet's private key, never the control plane.", + "type": "module", + "engines": { "node": ">=22.12" }, + "scripts": { + "start": "node src/server.js" + }, + "dependencies": { + "@arkiv-network/sdk": "1.0.1", + "viem": "^2.21.0" + } +} diff --git a/chain/arkiv/src/schema.js b/chain/arkiv/src/schema.js new file mode 100644 index 00000000..3602ead3 --- /dev/null +++ b/chain/arkiv/src/schema.js @@ -0,0 +1,28 @@ +// The entity shape every mirrored routing decision takes on Arkiv. `defineSchema`'s own name +// ("mergit.routing_decision") becomes the built-in, always-queryable `arkiv.schema` attribute — +// verified live against Rabdi: `arkiv.schema = str('my-app.posts')` matches the docs' own example +// entity on the real network, so this is the real filter, not SDK sugar. +// +// `decisionId`, `seed` and `budgetMicros` link this entity back to the exact Postgres +// `routing_decisions` row and let anyone replay the draw — the thing migration 0016 says that +// table exists for ("any draw can be replayed from its seed and the posteriors it saw"). No +// `requester`: a discarded draft plan can't be deleted off Arkiv (entities are append-only, only +// expiry removes them), so attaching a wallet address to data that outlives a "discard" is wrong, +// and nothing ever queried by it anyway (`$owner`, the service wallet, is what reads filter on). +// +// `body` carries the JSON payload (candidates/picks) as raw bytes: Arkiv attributes are +// `str | i32 | u64 | u256 | dec | addr | bytes` only, no nested structures (confirmed by reading +// the SDK's own `Attributes` type), and a plain JSON string is smaller round-tripped as UTF-8 +// bytes than re-encoded through any of the scalar types. +import { defineSchema, i32, str, bytes, u64 } from "@arkiv-network/sdk/schema"; + +export const RoutingDecision = defineSchema("mergit.routing_decision", { + planId: str().indexed(), + decisionId: u64().indexed(), + taskIndex: i32().indexed(), + category: str().indexed(), + seed: u64().indexed(), + budgetMicros: u64().indexed(), + decidedAt: str().indexed(), + body: bytes(), +}); diff --git a/chain/arkiv/src/server.js b/chain/arkiv/src/server.js new file mode 100644 index 00000000..a7506239 --- /dev/null +++ b/chain/arkiv/src/server.js @@ -0,0 +1,114 @@ +// The one thing this sidecar exists for: a single `executeBatch` write per call, so every task's +// routing decision for one `explore=true` suggestions call lands in ONE transaction, not one per +// task. Reads never come through here — they're plain JSON-RPC `arkiv_query` calls the control +// plane makes itself (see control-plane/mergit_cp/arkiv_mirror.py); only a write needs real +// transaction encoding, which only the JS SDK demonstrably supports against Rabdi (verified live: +// the Python `arkiv-sdk` targets a different, incompatible processor address/query shape). +// +// The private key lives only here, never in the control plane's own process or env. +import { createServer } from "node:http"; +import { createWalletClient, ExpirationTime } from "@arkiv-network/sdk"; +import { u64 } from "@arkiv-network/sdk/schema"; +import { rabdi } from "@arkiv-network/sdk/chains"; +import { http as viemHttp } from "viem"; +import { privateKeyToAccount } from "viem/accounts"; +import { RoutingDecision } from "./schema.js"; + +const PORT = Number(process.env.PORT ?? 8088); +const ACCESS_KEY = process.env.ARKIV_ACCESS_KEY ?? ""; +const PRIVATE_KEY = process.env.ARKIV_PRIVATE_KEY; +const TOKEN = process.env.ARKIV_SIDECAR_TOKEN; + +if (!PRIVATE_KEY) throw new Error("ARKIV_PRIVATE_KEY is required"); +if (!TOKEN) throw new Error("ARKIV_SIDECAR_TOKEN is required"); + +const account = privateKeyToAccount(PRIVATE_KEY); +const walletClient = createWalletClient({ + account, + chain: rabdi({ accessKey: ACCESS_KEY }), + transport: viemHttp(), +}); + +// Serializes every write through this one account: two concurrent explore calls (different plans, +// different HTTP requests) must not send two transactions with the same nonce. A single-process +// promise chain is enough for one service wallet; it is not meant to survive a second replica. +let queue = Promise.resolve(); +function enqueue(fn) { + const result = queue.then(fn, fn); + queue = result.then( + () => undefined, + () => undefined, + ); + return result; +} + +async function readBody(req) { + const chunks = []; + for await (const chunk of req) chunks.push(chunk); + return Buffer.concat(chunks).toString("utf8"); +} + +function send(res, status, body) { + res.writeHead(status, { "content-type": "application/json" }).end(JSON.stringify(body)); +} + +const server = createServer(async (req, res) => { + if (req.method === "GET" && req.url === "/healthz") { + res.writeHead(200).end("ok"); + return; + } + if (req.method !== "POST" || req.url !== "/internal/mirror") { + res.writeHead(404).end(); + return; + } + if (req.headers.authorization !== `Bearer ${TOKEN}`) { + res.writeHead(401).end(); + return; + } + + let payload; + try { + payload = JSON.parse(await readBody(req)); + } catch { + send(res, 400, { error: "invalid json" }); + return; + } + + const { decisions, ttl_days: ttlDays } = payload; + if (!Array.isArray(decisions) || decisions.length === 0) { + send(res, 400, { error: "decisions must be a non-empty array" }); + return; + } + + try { + const { txHash, createdEntities } = await enqueue(() => + walletClient.executeBatch({ + creates: decisions.map((d) => ({ + schema: RoutingDecision, + values: { + planId: d.plan_id, + decisionId: u64(BigInt(d.decision_id)), + taskIndex: d.task_index, + category: d.category, + seed: u64(BigInt(d.seed)), + budgetMicros: u64(BigInt(d.budget_micros)), + decidedAt: d.decided_at, + body: new TextEncoder().encode( + JSON.stringify({ candidates: d.candidates, picks: d.picks }), + ), + }, + expiration: ExpirationTime.fromDays(ttlDays ?? 30), + })), + }), + ); + // `createdEntities` is already `Hex[]` (the SDK's own type), not objects with a `.key` — an + // earlier version mapped `.key` off each string here, which silently produced `[null, ...]`. + send(res, 200, { tx_hash: txHash, entity_keys: createdEntities }); + } catch (err) { + send(res, 502, { error: String(err?.message ?? err) }); + } +}); + +server.listen(PORT, () => { + console.log(`arkiv sidecar listening on :${PORT}, wallet ${account.address}`); +}); diff --git a/control-plane/README.md b/control-plane/README.md index 0970b3e0..8701d7c9 100644 --- a/control-plane/README.md +++ b/control-plane/README.md @@ -204,6 +204,31 @@ without it, `GET /v0/plans/{plan_uuid}/events` sees no session and answers `401` **Database.** The control plane uses its own database, not the gateway's: both have tables named `plans`, `tasks` and `outbox`, and spec/events-v0.md says neither reads the other's. +## Arkiv mirror (ETHGlobal Mumbai, "Build on Arkiv" track) + +An exploring `GET /v0/plans/{uuid}/suggestions?explore=true` call already records its draw in this +service's own `routing_decisions` table (pre-existing, migration `0016`); `mergit_cp/arkiv_mirror.py` +additionally mirrors it as a public entity on Arkiv's Rabdi testnet, scheduled with FastAPI's +`BackgroundTasks` so it runs after the response is already on its way — a down testnet or a slow +write (the sidecar waits for the transaction to land) never holds up the request +(`routes_routing._mirror_decisions` catches and logs any failure). `GET +/v0/plans/{uuid}/arkiv-suggestions` reads the mirror back for the plan's owner. See +`../arkiv/schema.md` for the entity shape and `../chain/arkiv/README.md` for the write side. + +**Reads** are plain httpx against Rabdi's `arkiv_query` JSON-RPC — no SDK; the only Python Arkiv SDK +on PyPI targets a different, incompatible network (confirmed live, `../arkiv/friction.md`). **Writes** +go to `../chain/arkiv/`, a small Node sidecar holding the service wallet's private key and using the +official JS SDK (the one confirmed correct for Rabdi), so this service's own process and env never +hold that key. + +All optional, all unset by default (`Settings.arkiv_enabled`): `ARKIV_RPC_URL` (Rabdi's RPC, +`https://rpc.rabdi.db-chain.testnet.arkiv.network`), `ARKIV_OWNER_ADDRESS` (the service wallet's +*public* address, to filter reads by `$owner` — not its key, which this service never holds), +`ARKIV_ACCESS_KEY` (from `hub.arkiv.network/access-keys`, raises the public RPC's rate limit), +`ARKIV_SIDECAR_URL` (the sidecar's internal address) and `ARKIV_SIDECAR_TOKEN` (the shared secret +sent as `Authorization: Bearer`, same pattern as `WORKER_TOKEN`), `ARKIV_ENTITY_TTL_DAYS` (default +30). No extra Python dependency: reads are plain `httpx`, already a dependency. + ## The worker `python -m mergit_worker` claims one `queued` job at a time (`SELECT … FOR UPDATE SKIP LOCKED`), runs the diff --git a/control-plane/mergit_cp/app.py b/control-plane/mergit_cp/app.py index a27864c5..ebd84b15 100644 --- a/control-plane/mergit_cp/app.py +++ b/control-plane/mergit_cp/app.py @@ -24,6 +24,7 @@ routes_runs, wallet_interest, ) +from .arkiv_mirror import ArkivMirror from .config import Settings from .db import migrate, open_pool from .events import EventDeliverer @@ -46,6 +47,7 @@ def create_app( process_outcomes: bool = True, update_reputation: bool = True, mailer: otp.Mailer | None = None, + arkiv: ArkivMirror | None = None, ) -> FastAPI: """``deliver_events=False`` leaves events in the outbox, and ``process_outcomes=False`` leaves finished jobs unprocessed, for tests that want to inspect or drive that state directly.""" @@ -69,6 +71,18 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None]: app.state.costs = PgCostSink(pool, prices) app.state.broadcaster = PlanBroadcaster() app.state.mailer = mailer or (otp.SmtpMailer(settings) if settings.email_enabled else None) + app.state.arkiv = arkiv or ( + ArkivMirror( + settings.arkiv_rpc_url, # type: ignore[arg-type] # non-None: arkiv_enabled checked it + settings.arkiv_access_key.get_secret_value() if settings.arkiv_access_key else None, + settings.arkiv_owner_address, # type: ignore[arg-type] # non-None: arkiv_enabled checked it + settings.arkiv_sidecar_url, + settings.arkiv_sidecar_token.get_secret_value() if settings.arkiv_sidecar_token else None, + settings.arkiv_entity_ttl_days, + ) + if settings.arkiv_enabled + else None + ) stop = asyncio.Event() deliverer = EventDeliverer( @@ -95,6 +109,8 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None]: if task is not None: with suppress(asyncio.CancelledError): await task + if app.state.arkiv is not None: + await app.state.arkiv.aclose() await deliverer.aclose() await costs.aclose() app.state.pr_store.close() diff --git a/control-plane/mergit_cp/arkiv_mirror.py b/control-plane/mergit_cp/arkiv_mirror.py new file mode 100644 index 00000000..9294c6cc --- /dev/null +++ b/control-plane/mergit_cp/arkiv_mirror.py @@ -0,0 +1,191 @@ +"""Mirrors routing decisions (``routing_decisions``, migration 0016) onto Arkiv's Rabdi testnet, for +ETHGlobal Mumbai's "Build on Arkiv (existing projects)" track. + +Postgres stays the source of truth: nothing here is read back to decide anything the product depends on. +This is a best-effort copy, scheduled after the request that recorded the decision has already responded +(`routes_routing.suggestions` uses `BackgroundTasks`), so a down testnet or a slow write never holds up a +real request. + +Two separate paths, deliberately: + +* **Reads** (`read_decisions`) are a plain JSON-RPC ``arkiv_query`` call over httpx, no SDK. Verified live + against Rabdi: it returns ``key``/``owner``/``attributes`` but never a top-level payload field — Rabdi + has no such field at the protocol level (``arkiv/friction.md``), so every value this module writes is a + named *attribute*, including the JSON blob (``body``, a ``bytes``-typed attribute), not the "payload" + concept the old (wrong-network) Python SDK exposes. +* **Writes** (`write_decisions`) go to a small Node sidecar (`chain/arkiv/`) that holds the one service + wallet's key and uses the official JS SDK, the only one confirmed to encode a write Rabdi accepts. One + HTTP call carries every task's decision for one plan, so the sidecar sends ONE batched transaction + (``executeBatch``), not one per task — fewer round trips, and no nonce race between a plan's own tasks. +""" + +from __future__ import annotations + +import json +import logging +from dataclasses import dataclass +from typing import Any, cast + +import httpx + +log = logging.getLogger(__name__) + +SCHEMA = "mergit.routing_decision" +"""`defineSchema`'s own name (`chain/arkiv/src/schema.js`), which Arkiv turns into the built-in, +always-queryable `arkiv.schema` attribute — confirmed live: `arkiv.schema = str('my-app.posts')` +matches the docs' own example entity on the real Rabdi network, so this is a real filter, not sugar.""" + + +@dataclass(frozen=True) +class Decision: + """One routing decision to mirror (`write_decisions`) or as read back (`read_decisions`'s + `MirroredDecision`, which adds `entity_key`). `decision_id` and `seed` link this entity back to + the exact `routing_decisions` row and let anyone replay the draw — the thing migration 0016 says + that table exists for; a mirror without them is only a record of the outcome, not of how it was + reached.""" + + decision_id: int + task_index: int + category: str + seed: int + budget_micros: int + decided_at: str + candidates: list[dict[str, Any]] + picks: list[dict[str, Any]] + + +@dataclass(frozen=True) +class MirroredDecision(Decision): + entity_key: str = "" + + +class ArkivMirror: + def __init__( + self, + rpc_url: str, + access_key: str | None, + owner_address: str, + sidecar_url: str | None, + sidecar_token: str | None, + ttl_days: int, + *, + http_client: httpx.AsyncClient | None = None, + ) -> None: + self._rpc_url = f"{rpc_url}/{access_key}" if access_key else rpc_url + self._owner = owner_address.lower() + self._sidecar_url = sidecar_url + self._sidecar_token = sidecar_token + self._ttl_days = ttl_days + self._http = http_client or httpx.AsyncClient() + + async def aclose(self) -> None: + await self._http.aclose() + + async def write_decisions(self, plan_id: str, decisions: list[Decision]) -> None: + """One HTTP call to the sidecar, one on-chain transaction for every task in `decisions`. Raises + on failure (no sidecar configured, unreachable, out of funds, a bad key); the caller decides + whether that may fail the request (it must not — see this module's docstring). + + No requester address on the entity: a discarded draft plan can't be deleted off Arkiv (entities + are append-only; only expiry removes them), so attaching the requester's wallet to public, + permanent-until-expiry data would outlive the user's own "discard" action. Nothing here ever + filtered reads by it either — `describe_query` only needs `planId` and `$owner`.""" + if not self._sidecar_url: + raise RuntimeError("ARKIV_SIDECAR_URL is not set; nothing to mirror to") + body = { + "ttl_days": self._ttl_days, + "decisions": [ + { + "plan_id": plan_id, + "decision_id": d.decision_id, + "task_index": d.task_index, + "category": d.category, + "seed": d.seed, + "budget_micros": d.budget_micros, + "decided_at": d.decided_at, + "candidates": d.candidates, + "picks": d.picks, + } + for d in decisions + ], + } + headers = {"Authorization": f"Bearer {self._sidecar_token}"} if self._sidecar_token else {} + response = await self._http.post( + f"{self._sidecar_url}/internal/mirror", json=body, headers=headers, timeout=120.0 + ) + response.raise_for_status() + + def describe_query(self, plan_id: str) -> str: + """The exact query `read_decisions` runs, for display (README's/the panel's "one query anyone + can run") — real and copy-pasteable, not illustrative: schema tag, plan id, and the owner filter + that keeps a spoofed entity with Mergit's own tag out of someone else's results. Ordered by + key so a second page (`read_decisions`' own `limit`) is stable across calls, per the docs' + own pagination note.""" + return ( + f"arkiv.schema = str('{SCHEMA}') AND planId = str('{plan_id}')" + f" AND $owner = addr({self._owner}) ORDER BY $key ASC" + ) + + async def read_decisions(self, plan_id: str, *, limit: int = 50) -> list[MirroredDecision]: + """Every live mirrored decision for one plan, filtered by schema, plan id, *and* `$owner` (the + service wallet) — without the owner filter, anyone could write a spoofed entity with Mergit's + own schema tag and a matching `planId` and have it show up in another plan's panel.""" + query = self.describe_query(plan_id) + response = await self._http.post( + self._rpc_url, + json={ + "jsonrpc": "2.0", + "id": 1, + "method": "arkiv_query", + "params": [query, {"select": {"key": True, "attributes": True}, "limit": limit}], + }, + timeout=20.0, + ) + response.raise_for_status() + body = response.json() + if "error" in body: + raise RuntimeError(f"arkiv_query failed: {body['error']}") + out: list[MirroredDecision] = [] + for entity in body.get("result", {}).get("data", []): + decoded = _decode(entity) + if decoded is not None: + out.append(decoded) + return out + + +def _as_int(value: Any) -> int: + """`i32` attributes (`taskIndex`) come back as a plain JSON number; `u64` ones (`decisionId`, + `seed`, `budgetMicros`) come back as a hex quantity (`"0x2a"`) — confirmed live against Rabdi's + own `select` output. Both go through `int()` once this normalizes the hex case.""" + if isinstance(value, str) and value.startswith("0x"): + return int(value, 16) + return int(value) + + +def _decode(entity: dict[str, Any]) -> MirroredDecision | None: + """Arkiv is a public, permissionless database: anyone can write an entity matching our schema, + plan id and owner filter (if they also control the service wallet — they don't, but a malformed + or future-schema entity from our own future code is still a real possibility), so a decode failure + is skipped and logged, never raised.""" + try: + attrs = {a["name"]: a["value"] for a in entity.get("attributes", [])} + body: dict[str, Any] = {} + raw_body = attrs.get("body") + if isinstance(raw_body, str) and raw_body.startswith("0x"): + parsed: Any = json.loads(bytes.fromhex(raw_body[2:])) + if isinstance(parsed, dict): + body = cast("dict[str, Any]", parsed) + return MirroredDecision( + entity_key=str(entity.get("key", "")), + decision_id=_as_int(attrs.get("decisionId", -1)), + task_index=_as_int(attrs.get("taskIndex", -1)), + category=str(attrs.get("category", "")), + seed=_as_int(attrs.get("seed", 0)), + budget_micros=_as_int(attrs.get("budgetMicros", 0)), + decided_at=str(attrs.get("decidedAt", "")), + candidates=body.get("candidates", []), + picks=body.get("picks", []), + ) + except (TypeError, ValueError, KeyError, json.JSONDecodeError): + log.warning("skipping an unreadable entity tagged %r: %r", SCHEMA, entity.get("key")) + return None diff --git a/control-plane/mergit_cp/config.py b/control-plane/mergit_cp/config.py index 7bc1054b..ee636a89 100644 --- a/control-plane/mergit_cp/config.py +++ b/control-plane/mergit_cp/config.py @@ -135,6 +135,25 @@ class Settings(BaseSettings): otp_max_attempts: int = Field(default=5, ge=1) otp_sends_per_hour: int = Field(default=5, ge=1) + arkiv_rpc_url: str | None = None + """Arkiv's Rabdi testnet RPC (``https://rpc.rabdi.db-chain.testnet.arkiv.network``), used directly + (``arkiv_query`` over httpx) for reads. Unset: the mirror and its read endpoint are both no-ops + (``arkiv_enabled``).""" + arkiv_access_key: SecretStr | None = None + """Raises the public RPC's heavy rate limit; free, from ``hub.arkiv.network/access-keys``.""" + arkiv_owner_address: str | None = None + """The service wallet's public address (not its key — this service never holds that; see + ``chain/arkiv/``). Read-only, used to filter `read_decisions` by `$owner` so a spoofed entity + with Mergit's own schema tag cannot show up in someone else's results.""" + arkiv_sidecar_url: str | None = None + """The Arkiv write sidecar (`chain/arkiv/`), the only thing that holds the service wallet's private + key and can sign a write Rabdi accepts (`arkiv_mirror.py`'s own docstring explains why the control + plane cannot do this itself).""" + arkiv_sidecar_token: SecretStr | None = None + """Shared secret sent as ``Authorization: Bearer`` to the sidecar, same pattern as `worker_token`.""" + arkiv_entity_ttl_days: int = Field(default=30, ge=1) + """How long a mirrored routing decision lives on Arkiv before it expires.""" + worker_token: SecretStr """Shared secret the worker sends as ``Authorization: Bearer`` on POST /v0/internal/progress (plan task O7). Not named in spec/events-v0.md, which only covers the orchestrator-gateway leg; @@ -180,6 +199,13 @@ def google_enabled(self) -> bool: def email_enabled(self) -> bool: return bool(self.smtp_host and self.smtp_from) + @property + def arkiv_enabled(self) -> bool: + """Reads need only the RPC and the wallet's public address. A write additionally needs the + sidecar reachable (`arkiv_mirror.write_decisions` raises its own clear error when it isn't, + caught the same best-effort way as any other mirror failure).""" + return bool(self.arkiv_rpc_url and self.arkiv_owner_address) + @field_validator("cors_origins", "repo_hosts", "extra_domains", "planner_models", mode="before") @classmethod def _split_origins(cls, value: object) -> object: diff --git a/control-plane/mergit_cp/deps.py b/control-plane/mergit_cp/deps.py index 0308238a..02036d5c 100644 --- a/control-plane/mergit_cp/deps.py +++ b/control-plane/mergit_cp/deps.py @@ -1,9 +1,10 @@ -"""Request-scoped access to the process-wide settings and database pool.""" +"""Request-scoped access to the process-wide settings, database pool, and Arkiv mirror.""" -from typing import cast +from typing import Annotated, cast -from fastapi import Request +from fastapi import Depends, Request +from .arkiv_mirror import ArkivMirror from .config import Settings from .db import Pool @@ -14,3 +15,11 @@ def get_settings(request: Request) -> Settings: def get_pool(request: Request) -> Pool: return cast(Pool, request.app.state.pool) + + +def get_arkiv(request: Request) -> ArkivMirror | None: + """``None`` when `Settings.arkiv_enabled` is false (no RPC/key configured).""" + return cast("ArkivMirror | None", request.app.state.arkiv) + + +ArkivDep = Annotated[ArkivMirror | None, Depends(get_arkiv)] diff --git a/control-plane/mergit_cp/routes_routing.py b/control-plane/mergit_cp/routes_routing.py index 45c255ec..90398431 100644 --- a/control-plane/mergit_cp/routes_routing.py +++ b/control-plane/mergit_cp/routes_routing.py @@ -13,26 +13,45 @@ """ import json +import logging import random import secrets from datetime import UTC, datetime -from typing import Annotated, Any +from typing import Annotated, Any, NamedTuple from uuid import UUID -from fastapi import APIRouter, Depends, HTTPException, Query +from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query from psycopg import AsyncConnection from psycopg.rows import TupleRow from psycopg.types.json import Jsonb from . import eligibility, plans from .agents import Agent +from .arkiv_mirror import ArkivMirror, Decision from .auth import PoolDep, PrincipalDep, SettingsDep +from .deps import ArkivDep from .reputation_model import Params, Posterior from .routes_plans import get_agents from .routing import BELOW_MAX_ATTEMPTS, OVER_BUDGET_IN_THIS_DRAW, Pick, rank router = APIRouter(prefix="/v0/plans", tags=["routing"]) +log = logging.getLogger(__name__) + + +class MirrorRow(NamedTuple): + """One just-inserted `routing_decisions` row, exactly as Postgres has it — `decision_id` and + `created_at` come from the `INSERT ... RETURNING`, not recomputed, so the Arkiv mirror + (`_mirror_decisions`) links back to the real row and carries its real timestamp, not a second, + slightly later one taken when the background task happens to run.""" + + decision_id: int + task_index: int + budget_micros: int + created_at: datetime + picks: list[Pick] + + NOTE = ( "A suggestion, not an assignment: you still choose the agents when you approve. It ranks agents by " "validated success per unit cost in this category, from outcomes the platform recorded; free-credit work " @@ -89,6 +108,8 @@ async def suggestions( settings: SettingsDep, pool: PoolDep, agents: Annotated[list[Agent], Depends(get_agents)], + arkiv: ArkivDep, + background_tasks: BackgroundTasks, explore: Annotated[bool, Query()] = False, ) -> dict[str, Any]: """Per task: the agents expected to fit its budget, best first (at most ``max_attempts``), and every @@ -166,10 +187,12 @@ async def suggestions( status_code=429, detail="too many exploring suggestions this hour; try again later" ) await conn.execute("INSERT INTO suggestion_attempts (requester) VALUES (%s)", (address,)) + mirror_rows: list[MirrorRow] = [] for index, budget, picks in decisions: - await conn.execute( + cur = await conn.execute( "INSERT INTO routing_decisions (plan_uuid, task_index, category, seed, budget_micros," - " candidates, picks) VALUES (%s, %s, %s, %s, %s, %s, %s)", + " candidates, picks) VALUES (%s, %s, %s, %s, %s, %s, %s)" + " RETURNING decision_id, created_at", ( plan.plan_uuid, index, @@ -180,6 +203,14 @@ async def suggestions( Jsonb(json.loads(json.dumps([_pick_record(p) for p in picks]))), ), ) + row = await cur.fetchone() + assert row is not None # INSERT ... RETURNING always returns the row it just inserted + decision_id, created_at = row + mirror_rows.append(MirrorRow(decision_id, index, budget, created_at, picks)) + if arkiv is not None: + background_tasks.add_task( + _mirror_decisions, arkiv, plan.plan_id, plan.category, seed, mirror_rows, candidates + ) return { "plan_id": plan.plan_id, "mode": "explore" if explore else "greedy", @@ -189,6 +220,53 @@ async def suggestions( } +@router.get("/{plan_uuid}/arkiv-suggestions") +async def arkiv_suggestions( + plan_uuid: UUID, principal: PrincipalDep, pool: PoolDep, arkiv: ArkivDep +) -> dict[str, Any]: + """The owner's own exploring-suggestion mirror, as stored on Arkiv's Rabdi testnet rather than in + `routing_decisions` — proof that the write side (`_mirror_decisions`) is actually read somewhere a + user sees, not merely written and ignored (the Arkiv prize's own distinction). ``enabled: false`` when + `Settings.arkiv_enabled` is false; the frontend then hides the panel instead of calling this at all. + A reachability failure (bad key, unreachable RPC) is reported in ``error``, not a 500: this read is + advisory, same as the write it reads back, and a misconfigured or down testnet must not break the + page for every other, unrelated feature on it.""" + if arkiv is None: + return {"enabled": False, "decisions": []} + async with pool.connection() as conn: + plan = await plans.load(conn, plan_uuid) + if plan is None: + raise HTTPException(status_code=404, detail="no such plan") + if not principal.owns(plan.requester): + raise HTTPException( + status_code=403, detail="only the requester may see suggestions for this plan" + ) + query = arkiv.describe_query(plan.plan_id) + try: + decisions = await arkiv.read_decisions(plan.plan_id) + except Exception: # any SDK/RPC failure; see the docstring + log.warning("arkiv read failed for plan %s", plan.plan_id, exc_info=True) + return {"enabled": True, "query": query, "decisions": [], "error": "could not reach Arkiv"} + return { + "enabled": True, + "query": query, + "decisions": [ + { + "entity_key": d.entity_key, + "decision_id": d.decision_id, + "task_index": d.task_index, + "category": d.category, + "seed": d.seed, + "budget_micros": d.budget_micros, + "decided_at": d.decided_at, + "candidates": d.candidates, + "picks": d.picks, + } + for d in decisions + ], + } + + def _empty_reason( left_out: dict[str, str], excluded: dict[str, str], candidates: dict[str, Posterior], prior: Posterior ) -> str: @@ -242,3 +320,40 @@ def _snapshot(candidates: dict[str, Posterior]) -> list[dict[str, Any]]: def _pick_record(pick: Pick) -> dict[str, Any]: return {"agent_id": pick.agent_id, "score": pick.score, "success": pick.success, "cost": pick.cost} + + +async def _mirror_decisions( + arkiv: ArkivMirror, + plan_id: str, + category: str, + seed: int, + rows: list[MirrorRow], + candidates: dict[str, Posterior], +) -> None: + """Runs as a `BackgroundTasks` entry, after the response is already on its way: a down testnet or + a slow write (the sidecar waits for the transaction to land) must not hold up a suggestions call + that Postgres already recorded, so every failure is caught and logged, never raised. One call to + `write_decisions` for every task in this plan's draw, so the sidecar sends one transaction, not + one per task (`chain/arkiv/src/server.js`'s `executeBatch`). + + Mirrors `routing_decisions` for real, not just a record of which agents got picked: `decision_id` + and `seed` are what migration 0016 means by "any draw can be replayed from its seed and the + posteriors it saw" — without them the Arkiv copy can't back that claim (`arkiv/why-arkiv.md`).""" + snapshot = _snapshot(candidates) + to_write = [ + Decision( + decision_id=row.decision_id, + task_index=row.task_index, + category=category, + seed=seed, + budget_micros=row.budget_micros, + decided_at=row.created_at.isoformat(), + candidates=snapshot, + picks=[_pick_record(p) for p in row.picks], + ) + for row in rows + ] + try: + await arkiv.write_decisions(plan_id, to_write) + except Exception: # any sidecar/RPC failure; Arkiv is a mirror, not the record of truth + log.warning("arkiv mirror failed for plan %s", plan_id, exc_info=True) diff --git a/control-plane/pyproject.toml b/control-plane/pyproject.toml index 54d2c39c..d77ec941 100644 --- a/control-plane/pyproject.toml +++ b/control-plane/pyproject.toml @@ -8,6 +8,10 @@ dependencies = [ # siwe 4.4 (last release Oct 2024) defines ALPHA in its grammar; abnf 2.9 refuses that at import. "abnf>=2.2,<2.9", "anthropic>=2.3,<3", + # NOT arkiv-sdk (Python): verified live against Rabdi that its query options and write target + # (processor address) are both wrong for this network — see arkiv/friction.md. Reads here are + # plain httpx against arkiv_query; writes go to chain/arkiv/ (the JS SDK, which is correct for + # Rabdi) over HTTP. "docker>=7.1", "fastapi>=0.115", "httpx>=0.28", diff --git a/control-plane/tests/test_arkiv_mirror.py b/control-plane/tests/test_arkiv_mirror.py new file mode 100644 index 00000000..bc552cab --- /dev/null +++ b/control-plane/tests/test_arkiv_mirror.py @@ -0,0 +1,200 @@ +"""`ArkivMirror`: writes (one HTTP call to the sidecar) and reads (plain `arkiv_query` over httpx), both +against a mocked transport (`respx`), never a real Rabdi RPC or a real sidecar process. + +No `pytest-asyncio` in this repo (see other test files): each test drives its coroutine with +``asyncio.run``, same as the rest of the suite drives Postgres helpers. +""" + +from __future__ import annotations + +import asyncio +import json + +import httpx +import pytest +import respx + +from mergit_cp.arkiv_mirror import SCHEMA, ArkivMirror, Decision, _as_int + +RPC = "https://rpc.example" +SIDECAR = "https://sidecar.example" +OWNER = "0x1111111111111111111111111111111111111111" + + +@pytest.fixture +def mirror() -> ArkivMirror: + return ArkivMirror(RPC, None, OWNER, SIDECAR, "sidecar-secret", 30) + + +def _decision(task_index: int, **over: object) -> Decision: + base = { + "decision_id": 100 + task_index, + "task_index": task_index, + "category": "code", + "seed": 123456789, + "budget_micros": 500_000, + "decided_at": "2026-11-06T00:00:00+00:00", + "candidates": [{"agent_id": "a1"}], + "picks": [{"agent_id": "a1"}], + } + return Decision(**(base | over)) # type: ignore[arg-type] + + +def _entity( + key: str, + *, + decision_id: int = 100, + task_index: int, + category: str = "code", + seed: int = 123456789, + budget_micros: int = 500_000, + decided_at: str, + body: dict[str, object], +) -> dict[str, object]: + encoded = "0x" + json.dumps(body).encode().hex() + return { + "key": key, + "attributes": [ + {"name": "decisionId", "type": "u64", "value": hex(decision_id)}, + {"name": "taskIndex", "type": "i32", "value": task_index}, + {"name": "category", "type": "str", "value": category}, + {"name": "seed", "type": "u64", "value": hex(seed)}, + {"name": "budgetMicros", "type": "u64", "value": hex(budget_micros)}, + {"name": "decidedAt", "type": "str", "value": decided_at}, + {"name": "body", "type": "bytes", "value": encoded}, + ], + } + + +def test_write_decisions_sends_one_batched_call(mirror: ArkivMirror): + with respx.mock(assert_all_called=True) as mock: + route = mock.post(f"{SIDECAR}/internal/mirror").respond( + json={"tx_hash": "0xdead", "entity_keys": ["0xe1", "0xe2"]} + ) + asyncio.run(mirror.write_decisions("plan-1", [_decision(0), _decision(1)])) + assert route.call_count == 1 + sent = json.loads(route.calls[0].request.content) + assert sent["ttl_days"] == 30 + assert [d["task_index"] for d in sent["decisions"]] == [0, 1] + assert [d["decision_id"] for d in sent["decisions"]] == [100, 101] + assert all(d["plan_id"] == "plan-1" for d in sent["decisions"]) + assert all(d["seed"] == 123456789 for d in sent["decisions"]) + assert all(d["budget_micros"] == 500_000 for d in sent["decisions"]) + assert all("requester" not in d for d in sent["decisions"]) + assert route.calls[0].request.headers["authorization"] == "Bearer sidecar-secret" + + +def test_write_decisions_raises_when_no_sidecar_is_configured(): + arkiv = ArkivMirror(RPC, None, OWNER, None, None, 30) + with pytest.raises(RuntimeError): + asyncio.run(arkiv.write_decisions("plan-1", [_decision(0)])) + + +def test_write_decisions_raises_on_a_sidecar_error(mirror: ArkivMirror): + with respx.mock(assert_all_called=True) as mock: + mock.post(f"{SIDECAR}/internal/mirror").respond(502, json={"error": "insufficient funds"}) + with pytest.raises(httpx.HTTPStatusError): + asyncio.run(mirror.write_decisions("plan-1", [_decision(0)])) + + +def test_describe_query_matches_what_read_decisions_sends(mirror: ArkivMirror): + query = mirror.describe_query("plan-1") + assert query == ( + f"arkiv.schema = str('{SCHEMA}') AND planId = str('plan-1')" + f" AND $owner = addr({OWNER}) ORDER BY $key ASC" + ) + with respx.mock(assert_all_called=True) as mock: + route = mock.post(RPC).respond(json={"jsonrpc": "2.0", "id": 1, "result": {"data": []}}) + asyncio.run(mirror.read_decisions("plan-1")) + sent = json.loads(route.calls[0].request.content) + assert sent["method"] == "arkiv_query" + assert sent["params"][0] == query + + +def test_read_decisions_decodes_the_bytes_body_attribute(mirror: ArkivMirror): + entity = _entity( + "0xe1", + decision_id=42, + task_index=0, + seed=999, + budget_micros=500_000, + decided_at="2026-11-06T00:00:00+00:00", + body={"candidates": [{"agent_id": "a1"}], "picks": [{"agent_id": "a1"}]}, + ) + with respx.mock(assert_all_called=True) as mock: + mock.post(RPC).respond(json={"jsonrpc": "2.0", "id": 1, "result": {"data": [entity]}}) + decisions = asyncio.run(mirror.read_decisions("plan-1")) + assert len(decisions) == 1 + d = decisions[0] + assert d.entity_key == "0xe1" + assert d.decision_id == 42 + assert d.task_index == 0 + assert d.category == "code" + assert d.seed == 999 + assert d.budget_micros == 500_000 + assert d.decided_at == "2026-11-06T00:00:00+00:00" + assert d.candidates == [{"agent_id": "a1"}] + assert d.picks == [{"agent_id": "a1"}] + + +def test_read_decisions_raises_on_an_rpc_error(mirror: ArkivMirror): + with respx.mock(assert_all_called=True) as mock: + mock.post(RPC).respond(json={"jsonrpc": "2.0", "id": 1, "error": {"code": -32001, "message": "bad"}}) + with pytest.raises(RuntimeError): + asyncio.run(mirror.read_decisions("plan-1")) + + +def test_read_decisions_raises_on_an_unreachable_rpc(mirror: ArkivMirror): + with respx.mock(assert_all_called=True) as mock: + mock.post(RPC).mock(side_effect=httpx.ConnectError("refused")) + with pytest.raises(httpx.ConnectError): + asyncio.run(mirror.read_decisions("plan-1")) + + +def test_read_decisions_skips_an_entity_it_cannot_decode(mirror: ArkivMirror): + """Arkiv is a public, permissionless database: anyone can write an entity matching our schema tag, + plan id and owner filter, so a genuinely malformed match (unparseable body, a non-numeric + ``taskIndex``) is skipped, not raised.""" + bad_body = { + "key": "0xbad1", + "attributes": [ + {"name": "taskIndex", "type": "i32", "value": 0}, + {"name": "body", "type": "bytes", "value": "0xnotvalidhex"}, + ], + } + bad_index = { + "key": "0xbad2", + "attributes": [{"name": "taskIndex", "type": "str", "value": "not a number"}], + } + good = _entity("0xgood", task_index=1, decided_at="t", body={}) + with respx.mock(assert_all_called=True) as mock: + mock.post(RPC).respond( + json={"jsonrpc": "2.0", "id": 1, "result": {"data": [bad_body, bad_index, good]}} + ) + decisions = asyncio.run(mirror.read_decisions("plan-1")) + assert [d.entity_key for d in decisions] == ["0xgood"] + + +def test_read_decisions_tolerates_a_body_that_is_valid_json_but_not_an_object(mirror: ArkivMirror): + """A `body` like a bare JSON array parses fine but has no `candidates`/`picks` to read; the entity + still decodes, just with both empty.""" + entity = { + "key": "0xe1", + "attributes": [ + {"name": "taskIndex", "type": "i32", "value": 0}, + {"name": "body", "type": "bytes", "value": "0x" + json.dumps([1, 2, 3]).encode().hex()}, + ], + } + with respx.mock(assert_all_called=True) as mock: + mock.post(RPC).respond(json={"jsonrpc": "2.0", "id": 1, "result": {"data": [entity]}}) + decisions = asyncio.run(mirror.read_decisions("plan-1")) + assert len(decisions) == 1 + assert decisions[0].candidates == [] + assert decisions[0].picks == [] + + +def test_as_int_reads_both_plain_numbers_and_hex_quantities(): + """`i32` attributes come back as a plain JSON number; `u64` ones come back as a hex quantity — + confirmed live against Rabdi's own `select` output (`arkiv/friction.md`).""" + assert _as_int(5) == 5 + assert _as_int("0x2a") == 42 diff --git a/control-plane/tests/test_routes_routing.py b/control-plane/tests/test_routes_routing.py index 0495f2b7..b9aa4d9c 100644 --- a/control-plane/tests/test_routes_routing.py +++ b/control-plane/tests/test_routes_routing.py @@ -16,13 +16,16 @@ from types import SimpleNamespace from typing import Any +import httpx import psycopg import pytest +import respx from fastapi.testclient import TestClient from mergit_cp import routes_routing from mergit_cp.agents import Agent from mergit_cp.app import create_app +from mergit_cp.arkiv_mirror import ArkivMirror from mergit_cp.config import Settings from mergit_cp.db import Pool from mergit_cp.eligibility import AT_CAP, IN_PROBATION @@ -578,3 +581,108 @@ def test_an_empty_answer_with_agents_out_for_probation_does_not_blame_the_budget assert why(task) == {"3": IN_PROBATION} | dict.fromkeys("1245", OVER_BUDGET) assert task["reason"] == PARTLY_LEFT_OUT + NO_FIT assert task["reason"].startswith("some agents are left out for probation or the operator cap") + + +# -- Arkiv mirror (ETHGlobal Mumbai "Build on Arkiv" track): routing_decisions also written to Arkiv +# (one batched call to the sidecar, chain/arkiv/), and readable back through its own endpoint. Both the +# sidecar and the Rabdi RPC are mocked at the HTTP layer (`respx`), never a real network or process. -- + +ARKIV_RPC = "https://rpc.example" +ARKIV_SIDECAR = "https://sidecar.example" +ARKIV_OWNER = "0x1111111111111111111111111111111111111111" + + +def _fake_entity(decision: dict[str, Any]) -> dict[str, Any]: + """An entity shaped like a real `arkiv_query` match, encoding one decision the sidecar "wrote" + (`_entity` in `test_arkiv_mirror.py` is the unit-level version of this).""" + encoded_body = ( + "0x" + json.dumps({"candidates": decision["candidates"], "picks": decision["picks"]}).encode().hex() + ) + return { + "key": f"0xe{decision['task_index']}", + "attributes": [ + {"name": "taskIndex", "type": "i32", "value": decision["task_index"]}, + {"name": "category", "type": "str", "value": decision["category"]}, + {"name": "decidedAt", "type": "str", "value": decision["decided_at"]}, + {"name": "body", "type": "bytes", "value": encoded_body}, + ], + } + + +@contextmanager +def arkiv_client(settings: Settings, llm: FakeLLM) -> Iterator[tuple[TestClient, respx.MockRouter]]: + arkiv = ArkivMirror(ARKIV_RPC, None, ARKIV_OWNER, ARKIV_SIDECAR, "sidecar-secret", 30) + with ( + respx.mock(assert_all_called=False) as mock, + TestClient( + create_app(settings, llm=llm, deliver_events=False, arkiv=arkiv), + base_url="https://testserver", + ) as c, + ): + mock.post(f"{ARKIV_SIDECAR}/internal/mirror").respond(json={"tx_hash": "0xdead", "entity_keys": []}) + yield c, mock + + +def test_an_exploring_suggestion_is_mirrored_to_arkiv_and_readable_back(settings: Settings, llm: FakeLLM): + with arkiv_client(settings, llm) as (c, mock): + sign_in(c) + plan = create(c) + write_route = mock.routes[-1] # the default write mock registered by arkiv_client + got = c.get(f"/v0/plans/{plan['plan_uuid']}/suggestions", params={"explore": "true"}).json() + + # One write call, carrying every task's decision — never one call per task. + assert write_route.call_count == 1 + sent = json.loads(write_route.calls[0].request.content) + assert len(sent["decisions"]) == len(got["tasks"]) == 2 + assert all(d["plan_id"] == plan["plan_id"] for d in sent["decisions"]) + # Replayable, not just a record of the outcome: links back to the real Postgres row and + # carries what a replay needs (migration 0016's own reason routing_decisions exists). + assert all(isinstance(d["decision_id"], int) for d in sent["decisions"]) + assert all(d["seed"] == got["seed"] for d in sent["decisions"]) + assert all(isinstance(d["budget_micros"], int) for d in sent["decisions"]) + assert all("requester" not in d for d in sent["decisions"]) + + # The read endpoint decodes what was (supposedly) written on Arkiv, not a copy kept in Postgres. + entities = [_fake_entity(d) for d in sent["decisions"]] + mock.post(ARKIV_RPC).respond(json={"jsonrpc": "2.0", "id": 1, "result": {"data": entities}}) + mirrored = c.get(f"/v0/plans/{plan['plan_uuid']}/arkiv-suggestions").json() + assert mirrored["enabled"] is True + assert {d["task_index"] for d in mirrored["decisions"]} == {0, 1} + + +def test_arkiv_suggestions_degrades_instead_of_500_when_arkiv_is_unreachable( + settings: Settings, llm: FakeLLM +): + """A bad key or a down RPC must not 500 the page for an unrelated feature; the owner still sees + their plan, just told Arkiv could not be reached.""" + with arkiv_client(settings, llm) as (c, mock): + sign_in(c) + plan = create(c) + c.get(f"/v0/plans/{plan['plan_uuid']}/suggestions", params={"explore": "true"}) + + mock.post(ARKIV_RPC).mock(side_effect=httpx.ConnectError("rpc unreachable")) + resp = c.get(f"/v0/plans/{plan['plan_uuid']}/arkiv-suggestions") + + assert resp.status_code == 200 + body = resp.json() + assert body["enabled"] is True + assert body["decisions"] == [] + assert body["error"] == "could not reach Arkiv" + + +def test_arkiv_suggestions_is_owner_only(settings: Settings, llm: FakeLLM): + with arkiv_client(settings, llm) as (c, _mock): + sign_in(c) + plan = create(c) + c.get(f"/v0/plans/{plan['plan_uuid']}/suggestions", params={"explore": "true"}) + c.post("/v0/auth/logout") + sign_in(c) # a different, freshly generated account + resp = c.get(f"/v0/plans/{plan['plan_uuid']}/arkiv-suggestions") + assert resp.status_code == 403 + + +def test_arkiv_suggestions_is_disabled_when_arkiv_is_not_configured(app_client: TestClient): + sign_in(app_client) + plan = create(app_client) + got = app_client.get(f"/v0/plans/{plan['plan_uuid']}/arkiv-suggestions").json() + assert got == {"enabled": False, "decisions": []} diff --git a/control-plane/uv.lock b/control-plane/uv.lock index 50cefa42..0bae7900 100644 --- a/control-plane/uv.lock +++ b/control-plane/uv.lock @@ -2503,7 +2503,7 @@ wheels = [ [[package]] name = "web3" -version = "7.16.0" +version = "7.13.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "aiohttp" }, @@ -2521,9 +2521,9 @@ dependencies = [ { name = "typing-extensions" }, { name = "websockets" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/f1/d9/bdfa9e715804020c3f3676346065c18adbc207c9343a3458246d7430f45c/web3-7.16.0.tar.gz", hash = "sha256:b4a75a3fa94fef4d23d502eb3c2244146ef9a1ee0082cf1cb0a91586ba0510c3", size = 2211469, upload-time = "2026-05-01T21:22:20.666Z" } +sdist = { url = "https://files.pythonhosted.org/packages/8a/b4/1f026b025c51ce092a7a573b9308a2526b92a4044f94d9af78e7bb820a1c/web3-7.13.0.tar.gz", hash = "sha256:281795e0f5d404c1374e1771f6710bb43e0c975f3141366eb1680763edfb4806", size = 2194483, upload-time = "2025-08-04T16:57:19.442Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/2c/f9/5345c13f8469f3ce344b4d9934c0387b83a49420192482111e9c1fa95ec2/web3-7.16.0-py3-none-any.whl", hash = "sha256:760b2718c473980d70708c3593d9d28395db4b482f45e38a63a36fa028178f51", size = 1372181, upload-time = "2026-05-01T21:22:17.015Z" }, + { url = "https://files.pythonhosted.org/packages/95/a2/fc390ac63c8008318769eba2de071e95b7ba59b5fdc099529db73e5f1356/web3-7.13.0-py3-none-any.whl", hash = "sha256:4da7e953300577b7dadbaf430e5fd4479b127b1ad4910234b230fdcb8a49f735", size = 1369941, upload-time = "2025-08-04T16:57:17.587Z" }, ] [[package]] diff --git a/deploy/compose.controlplane.yaml b/deploy/compose.controlplane.yaml index 9eb2ac65..be1f80b6 100644 --- a/deploy/compose.controlplane.yaml +++ b/deploy/compose.controlplane.yaml @@ -79,6 +79,26 @@ services: GATEWAY_URL: http://gateway:8082 STORE_URL: http://store:8081 AGENTS_FILE: /agents.json + # Unset: the Arkiv mirror (ETHGlobal Mumbai "Build on Arkiv" track) is a no-op, same as any other + # ARKIV_* left blank (mergit_cp/config.py's arkiv_enabled). The sidecar below holds the one secret + # this needs beyond its public RPC/owner; ARKIV_SIDECAR_TOKEN is shared with it in secrets/api.env. + ARKIV_SIDECAR_URL: http://arkiv-sidecar:8088 + + # The one write Rabdi needs: holds the service wallet's private key, never the control plane's own + # process (chain/arkiv/README's own reasoning). Started only once deploy/secrets/arkiv.env exists — + # `docker compose --profile arkiv ... up -d`, same opt-in pattern as `keyrelay` above. + arkiv-sidecar: + profiles: ["arkiv"] + build: { context: ../chain/arkiv } + image: mergit-arkiv-sidecar:local + restart: unless-stopped + env_file: secrets/arkiv.env + healthcheck: + test: ["CMD", "node", "-e", "fetch('http://127.0.0.1:8088/healthz').then(r => process.exit(r.ok ? 0 : 1))"] + interval: 15s + timeout: 5s + retries: 3 + start_period: 10s worker: build: { context: .., dockerfile: control-plane/worker.Dockerfile } diff --git a/deploy/server/update.sh b/deploy/server/update.sh index 6ae74a87..b3e1f6d7 100755 --- a/deploy/server/update.sh +++ b/deploy/server/update.sh @@ -5,8 +5,11 @@ set -euo pipefail cd "$(dirname "$0")/../.." what=${1:-all} -# The key rotator runs only where it has been configured (see tools/keyrelay/README.md). -profiles=(); [ -f deploy/secrets/keyrelay/keyrelay.toml ] && profiles=(--profile keyrelay) +# The key rotator and the Arkiv write sidecar each run only where they've been configured (see +# tools/keyrelay/README.md and chain/arkiv/README.md). +profiles=() +[ -f deploy/secrets/keyrelay/keyrelay.toml ] && profiles+=(--profile keyrelay) +[ -f deploy/secrets/arkiv.env ] && profiles+=(--profile arkiv) compose() { docker compose "${profiles[@]}" -f deploy/compose.testnet.yaml -f deploy/compose.controlplane.yaml -f deploy/compose.server.yaml "$@"; } case "$what" in all|services|web|app) ;; *) echo "usage: $0 [all|services|web|app]" >&2; exit 2 ;; esac diff --git a/frontend/src/api/client.ts b/frontend/src/api/client.ts index 941e6a9f..a20a9fca 100644 --- a/frontend/src/api/client.ts +++ b/frontend/src/api/client.ts @@ -5,6 +5,7 @@ import type { AgentInfo, ApproveRequest, ApproveResponse, + ArkivSuggestions, NewPlanRequest, Plan, PlanDetail, @@ -123,6 +124,11 @@ export const getReputation = () => request("GET", "/v0/agent export const getSuggestions = (planUuid: string, explore: boolean) => request("GET", `${planPath(planUuid)}/suggestions${explore ? "?explore=true" : ""}`); +/** The same exploring suggestions, read back from Arkiv's Rabdi testnet rather than from Postgres + * (ETHGlobal Mumbai's "Build on Arkiv" track). `enabled: false` when the server has no Arkiv configured. */ +export const getArkivSuggestions = (planUuid: string) => + request("GET", `${planPath(planUuid)}/arkiv-suggestions`); + /** Server-Sent Events for one plan. Open with `withCredentials: true`, or the session cookie is not sent. */ export const planEventsUrl = (planUuid: string) => `${base()}${planPath(planUuid)}/events`; diff --git a/frontend/src/api/types.ts b/frontend/src/api/types.ts index 563c7385..1c7321f1 100644 --- a/frontend/src/api/types.ts +++ b/frontend/src/api/types.ts @@ -115,6 +115,32 @@ export type Suggestions = { tasks: TaskSuggestion[]; }; +export type ArkivPick = { agent_id: string; score: number; success: number; cost: number }; + +/** One exploring suggestion mirrored onto Arkiv's Rabdi testnet (ETHGlobal Mumbai's "Build on Arkiv" track), + * as read back through `arkiv_query` rather than from this plan's own Postgres row. */ +export type ArkivDecision = { + entity_key: string; + decision_id: number; + task_index: number; + category: string; + seed: number; + budget_micros: number; + decided_at: string; + candidates: unknown[]; + picks: ArkivPick[]; +}; + +export type ArkivSuggestions = { + /** False when the server has no `ARKIV_*` configured; the panel that renders this should then hide. */ + enabled: boolean; + query?: string; + /** Set when the server reached Arkiv's RPC and the query itself failed; distinct from `decisions: []`, + * which also means the query ran fine but matched nothing yet. */ + error?: string; + decisions: ArkivDecision[]; +}; + export type StepCost = { model: string | null; input_tokens: number; diff --git a/frontend/src/screens/live/run/SuggestAgents.tsx b/frontend/src/screens/live/run/SuggestAgents.tsx index a9735aa5..193aeaab 100644 --- a/frontend/src/screens/live/run/SuggestAgents.tsx +++ b/frontend/src/screens/live/run/SuggestAgents.tsx @@ -1,7 +1,16 @@ import { useState } from "react"; -import { errorMessage, getSuggestions } from "../../../api/client"; +import { errorMessage, getArkivSuggestions, getSuggestions } from "../../../api/client"; import { percent, usdc } from "../../../api/reputation"; -import type { Suggestions } from "../../../api/types"; +import type { ArkivDecision, ArkivSuggestions, Suggestions, TaskSuggestion } from "../../../api/types"; + +/** Does what Arkiv has for this task match the picks Postgres just answered with — proof the mirror + * carries what the product actually ranked, not just a write nobody checks against anything. */ +function picksMatch(task: TaskSuggestion | undefined, decision: ArkivDecision | undefined): boolean | undefined { + if (!task || !decision) return undefined; + const postgres = task.picks.map((p) => p.agent_id); + const arkiv = decision.picks.map((p) => p.agent_id); + return postgres.length === arkiv.length && postgres.every((id, i) => id === arkiv[i]); +} /** * Advice on which agents to pick, from what the platform has validated about each (never from free-credit work). @@ -14,10 +23,15 @@ export function SuggestAgents({ planUuid, onUse }: { planUuid: string; onUse: (s const [busy, setBusy] = useState(false); const [error, setError] = useState(); const [got, setGot] = useState(); + const [arkivBusy, setArkivBusy] = useState(false); + const [arkivError, setArkivError] = useState(); + const [arkiv, setArkiv] = useState(); const ask = () => { setBusy(true); setError(undefined); + setArkiv(undefined); + setArkivError(undefined); getSuggestions(planUuid, explore).then( (s) => { setGot(s); @@ -30,6 +44,24 @@ export function SuggestAgents({ planUuid, onUse }: { planUuid: string; onUse: (s ); }; + /** An exploring answer (and only that: greedy writes nothing) is mirrored onto Arkiv's Rabdi testnet + * (ETHGlobal Mumbai's "Build on Arkiv" track); this reads it back from there, not from this plan's own + * database, to show the mirror is actually read somewhere a user sees. */ + const viewOnArkiv = () => { + setArkivBusy(true); + setArkivError(undefined); + getArkivSuggestions(planUuid).then( + (a) => { + setArkiv(a); + setArkivBusy(false); + }, + (e: unknown) => { + setArkivError(errorMessage(e)); + setArkivBusy(false); + }, + ); + }; + return (
Suggested agents @@ -90,6 +122,53 @@ export function SuggestAgents({ planUuid, onUse }: { planUuid: string; onUse: (s > Use these picks + {got.mode === "explore" && ( +
+ + {arkivError && ( + + {arkivError} + + )} + {arkiv && !arkiv.enabled && Arkiv mirroring is off on this deployment.} + {arkiv?.enabled && arkiv.error && ( + + Arkiv's testnet could not be reached just now; try again shortly. + + )} + {arkiv?.enabled && !arkiv.error && ( +
+ + {arkiv.decisions.length === 0 + ? "Not found on Arkiv yet; the mirror write may still be in flight or failed silently." + : `${String(arkiv.decisions.length)} decision${arkiv.decisions.length === 1 ? "" : "s"} read back from Arkiv's Rabdi testnet, not this app's own database:`} + + {arkiv.decisions.map((d) => { + const matches = picksMatch(got.tasks[d.task_index], d); + return ( +
+ + Task {String(d.task_index + 1)} · {d.entity_key} + + {matches !== undefined && ( + + {matches ? "matches ✓" : "does not match the picks above"} + + )} +
+ ); + })} +
+ )} +
+ )}
)}