From 7e146676d957aad15e50b92d830032a2c83e2045 Mon Sep 17 00:00:00 2001 From: Joey Lau Date: Wed, 19 Aug 2026 14:22:03 +0800 Subject: [PATCH 1/4] fix: repair EVM examples, type-check them, document requirement + replay Three issues reported from the field, all reproduced against the code. 1. `provider` vs `evmProvider`. `CreateAcpClientInput` accepts `evmProvider` / `solanaProvider`; there is no `provider` key. All 8 EVM examples and 3 README snippets used `provider:`, so they failed to compile and threw "At least one provider ... must be provided" at runtime. Root cause: tsconfig.json excludes `src/examples*` (correctly -- they must not ship in dist/), so nothing ever type-checked them. Adds tsconfig.examples.json + `npm run typecheck:examples` so this can't regress, and makes the runtime error name the right key for plain-JS consumers who get no compile error at all. Also documents that `walletAddress` is viem's `Address`, so an env-sourced address needs an `as \`0x${string}\`` cast -- every example already did this, the README never mentioned it. 2. Raw job creation carries no requirement. Only createJobFromOffering / createJobByOfferingName post the "requirement" entry; createJob and the three hook variants send nothing but `description`. An evaluator on that path gets a deliverable with no stated ask. Adds TSDoc on all four creators, a README section, and a standalone evaluator quick start -- neither existing quick start covered the evaluator role. 3. Duplicate event delivery on restart. hydrateSessions() fires the handler with the latest entry of every active job on every start(), by design, so restarts replay whatever the job was waiting on. There is no dedup: the check in dispatch() is object identity, and JobRoomEntry has no stable id. Documents the replay contract and the persistent-dedup pattern it requires, and notes that the contract rejecting a redundant complete/reject is a failure path, not a deduplication mechanism. Also fixes the llm/ examples, which passed model: "gemini-3.1-flash-lite-preview" to a bare `new Anthropic()` client -- a Google model id on the Anthropic API. Verified: `npm run typecheck` and `npm run typecheck:examples` both clean (the latter reported 15 errors before this change); createAcpClients error paths and the evmProvider happy path exercised directly. Co-Authored-By: Claude Opus 5 (1M context) --- README.md | 240 ++++++++++++++++-- package-lock.json | 61 ++++- package.json | 3 + src/acpAgent.ts | 65 +++++ src/clientFactory.ts | 11 + src/examples/basic/buyer.ts | 2 +- src/examples/basic/seller.ts | 2 +- src/examples/fund-transfer/buyer.ts | 2 +- src/examples/fund-transfer/seller.ts | 2 +- src/examples/llm/README.md | 2 +- src/examples/llm/buyer.ts | 6 +- src/examples/llm/seller.ts | 4 +- .../subscription-fund-transfer/buyer.ts | 2 +- .../subscription-fund-transfer/seller.ts | 2 +- src/examples/subscription/buyer.ts | 2 +- src/examples/subscription/seller.ts | 2 +- tsconfig.examples.json | 16 ++ 17 files changed, 392 insertions(+), 32 deletions(-) create mode 100644 tsconfig.examples.json diff --git a/README.md b/README.md index bded569..0dac639 100644 --- a/README.md +++ b/README.md @@ -12,12 +12,15 @@ The Agent Commerce Protocol (ACP) Node SDK v2 is a ground-up rewrite of the ACP - [Quick Start](#quick-start) - [Buyer](#buyer) - [Seller](#seller) + - [Evaluator](#evaluator) - [Core Concepts](#core-concepts) - [AcpAgent](#acpagent) - [JobSession](#jobsession) - [Events](#events) + - [Restart & replay semantics](#restart--replay-semantics) - [AssetToken](#assettoken) - [Agent Discovery](#agent-discovery) + - [The requirement message](#the-requirement-message) - [LLM Integration](#llm-integration) - [Provider Adapters](#provider-adapters) - [Fund Transfer Jobs](#fund-transfer-jobs) @@ -68,7 +71,11 @@ import { base } from "@account-kit/infra"; async function main() { const buyer = await AcpAgent.create({ - provider: await PrivyAlchemyEvmProviderAdapter.create({ + // `evmProvider` for EVM chains, `solanaProvider` for Solana. There is no + // plain `provider` option -- see Provider Adapters below. + evmProvider: await PrivyAlchemyEvmProviderAdapter.create({ + // Typed `0x${string}`, so an env-sourced address needs a cast: + // process.env.BUYER_WALLET_ADDRESS as `0x${string}` walletAddress: "0xBuyerWalletAddress", walletId: "wallet-id", signerPrivateKey: "signer-private-key", @@ -128,7 +135,7 @@ import { base } from "@account-kit/infra"; async function main() { const seller = await AcpAgent.create({ - provider: await PrivyAlchemyEvmProviderAdapter.create({ + evmProvider: await PrivyAlchemyEvmProviderAdapter.create({ walletAddress: "0xSellerWalletAddress", walletId: "wallet-id", signerPrivateKey: "signer-private-key", @@ -175,6 +182,70 @@ async function main() { main().catch(console.error); ``` +### Evaluator + +A third-party evaluator is a separate process on its own wallet. The buyer opts +into it by passing that wallet as `evaluatorAddress` at job creation; the +evaluator then receives `job.submitted` and decides the job's outcome. Nothing +else in the lifecycle reaches it -- no `job.created`, no `budget.set`. + +```typescript +async function main() { + const evaluator = await AcpAgent.create({ + evmProvider: await PrivyAlchemyEvmProviderAdapter.create({ + walletAddress: process.env.EVALUATOR_WALLET_ADDRESS as `0x${string}`, + walletId: "wallet-id", + signerPrivateKey: "signer-private-key", + chains: [base], + }), + }); + + // start() replays the latest entry of every in-flight job, so a restart + // re-delivers a job.submitted you may already have ruled on. Persist this. + const ruled = await loadRuledJobKeys(); // your own store + + evaluator.on("entry", async (session: JobSession, entry: JobRoomEntry) => { + if (entry.kind !== "system" || entry.event.type !== "job.submitted") return; + + const key = `${session.chainId}-${session.jobId}-job.submitted`; + if (ruled.has(key)) return; + + // What was asked for, and what came back. + const requirement = session.entries.find( + (e) => e.kind === "message" && e.contentType === "requirement" + ); + const deliverable = entry.event.deliverable; + + const ok = await yourJudgement(requirement?.content, deliverable); + + // Record BEFORE the on-chain call -- a crash mid-transaction must not + // leave the job eligible for a second ruling on restart. + ruled.add(key); + await persistRuledJobKey(key); + + if (ok) { + await session.complete("Deliverable meets the requirement"); + } else { + await session.reject("Deliverable does not meet the requirement"); + } + }); + + await evaluator.start(() => console.log("Evaluator listening...")); +} +``` + +Two things to get right before an evaluator can do anything: + +- **The buyer must name it.** `createJobByOfferingName(..., { evaluatorAddress })` + -- omit it and the job runs in skip-evaluation mode, auto-completing on submit + so `job.submitted` never fires for anyone. +- **There must be a requirement to judge against.** Jobs created through the raw + `createJob` path carry no requirement message -- see + [The requirement message](#the-requirement-message). + +See [Restart & replay semantics](#restart--replay-semantics) for why the dedup +store above is not optional. + ## Core Concepts ### AcpAgent @@ -183,7 +254,8 @@ The main entry point. Creates an agent that listens for job events and manages s ```typescript const agent = await AcpAgent.create({ - provider: providerAdapter, // required -- EVM or Solana provider + evmProvider: evmProviderAdapter, // for EVM chains + // solanaProvider: solanaProviderAdapter, // for Solana -- at least one is required }); agent.on("entry", async (session, entry) => { @@ -200,12 +272,12 @@ await agent.stop(); | Method | Description | | ---------------------------------------------------------------------------------------------- | ------------------------------------------------- | -| `agent.start(onConnected?)` | Connect to event stream and hydrate existing jobs | +| `agent.start(onConnected?)` | Connect to event stream and hydrate existing jobs -- [replays the latest entry per active job](#restart--replay-semantics) | | `agent.stop()` | Disconnect and clean up | | `agent.on("entry", handler)` | Register handler for all job events and messages | | `agent.browseAgents(keyword, params?)` | Search for agents by keyword | -| `agent.createJob(chainId, params)` | Create an on-chain job | -| `agent.createFundTransferJob(chainId, params)` | Create a job with fund transfer intent | +| `agent.createJob(chainId, params)` | Create an on-chain job -- [sends no requirement message](#the-requirement-message) | +| `agent.createFundTransferJob(chainId, params)` | Create a job with fund transfer intent -- [sends no requirement message](#the-requirement-message) | | `agent.createJobByOfferingName(chainId, offeringName, providerAddress, requirementData, opts)` | Resolve offering by name → validated job creation | | `agent.createJobFromOffering(chainId, offering, providerAddress, requirementData, opts)` | Create job from full offering object | | `agent.getAgentByWalletAddress(walletAddress)` | Look up an agent by wallet address | @@ -264,6 +336,57 @@ agent.on("entry", async (session, entry) => { }); ``` +### Restart & replay semantics + +**`agent.start()` replays events, and your `entry` handler must be idempotent.** + +On startup the SDK calls `AcpJobApi.getActiveJobs()`, rebuilds a `JobSession` for +every in-flight job this wallet participates in, and fires your handler with the +**latest entry of each**. That replay is the feature that makes agents +restartable: kill a buyer sitting at `budget.set` and it resumes funding on the +next boot instead of stranding the job. + +The cost is that the same entry can reach your handler more than once across +restarts. An evaluator restarted while a job sits at `job.submitted` is called +for that submission again and will try to rule on a job it already ruled on. The +contract rejects the redundant `complete`/`reject`, so you'll see a revert rather +than a double payout -- **but do not treat that as your deduplication.** A revert +is a failure path, not a guard: it costs gas, it surfaces as an error you now +have to classify as benign, and any hook or fee transfer reached before the +revert still ran. + +The SDK does not dedupe for you, and it deliberately can't do it well: +`JobRoomEntry` carries no stable id, and the delivery you need to suppress +happens *across* process boundaries -- an in-memory `Set` is wiped by exactly the +restart that causes the replay. So dedup belongs in your own persistent store: + +```typescript +// Any durable store works -- SQLite, Redis, a JSON file. +const key = `${session.chainId}-${session.jobId}-${entry.event.type}`; +if (await store.has(key)) return; + +await store.put(key); // BEFORE the side effect, not after +await session.complete("..."); +``` + +Write the key **before** the on-chain call. Writing it after leaves a window +where a crash mid-transaction loses the record while the transaction lands, +which is the same duplicate you were trying to prevent. + +`(chainId, jobId, event.type)` is a good key for lifecycle events, which fire +once per job. Note that `budget.set` can legitimately repeat if a provider +re-proposes, so include `entry.timestamp` in the key if you act on it. + +Two related details worth knowing: + +- `agent.sessions` is populated by hydration, so you can detect in-flight work on + boot and avoid piling on a new job next to a resuming one -- see the `sessions` + TSDoc for the filter, and [`src/examples/basic/buyer.ts`](./src/examples/basic/buyer.ts) + for it in use. +- Where practical, make the decision itself idempotent by checking state instead + of history: `session.status` tells you whether a job is already terminal. The + SDK gates handler delivery by **role**, never by "has this agent already acted". + ### AssetToken Token abstraction that handles decimals and chain-specific addresses. @@ -318,6 +441,42 @@ const provider = await agent.getAgentByWalletAddress("0xProviderAddress"); If you already have the full offering object, you can use `createJobFromOffering` directly instead. +### The requirement message + +Step 4 above is the part that's easy to lose. **Only `createJobFromOffering` and +`createJobByOfferingName` send the requirement.** The lower-level creators -- +`createJob`, `createFundTransferJob`, `createSubscriptionJob`, +`createMultiHookJob` -- put a job on-chain and stop there. The job carries only +`params.description`, a free-text string. + +That matters most for evaluators. A job created through the raw path reaches +`job.submitted` with a deliverable and no stated ask, so an evaluator has nothing +to judge it against -- it can see what was delivered but not what was requested. +The provider is in the same position: no structured requirement ever arrives. + +If you create jobs outside the offering path, send the requirement yourself: + +```typescript +const jobId = await agent.createJob(base.id, { + providerAddress: SELLER_ADDRESS, + evaluatorAddress: EVALUATOR_ADDRESS, + expiredAt: Math.floor(Date.now() / 1000) + 3600, + description: "Meme Generation", // free text, not a requirement +}); + +// Send the structured ask -- contentType MUST be "requirement" +await agent.sendMessage( + base.id, + jobId.toString(), + JSON.stringify({ key: "I want a funny cat meme" }), + "requirement" +); +``` + +The offering path also gives you requirement validation against the offering's +JSON schema and an `expiredAt` derived from its SLA. Prefer it unless you need a +job that isn't backed by a registry offering. + **Browse parameters:** | Param | Description | @@ -379,21 +538,54 @@ See [`src/examples/llm/`](./src/examples/llm/) for complete LLM examples with Cl ## Provider Adapters -| Adapter | Use Case | -| -------------------------------- | ------------------------------------------------- | -| `PrivyAlchemyEvmProviderAdapter` | Privy-managed wallets with Alchemy infrastructure | -| `SolanaProviderAdapter` | Solana chain support | +| Adapter | Constructor key | Use Case | +| -------------------------------- | ---------------- | ------------------------------------------------- | +| `PrivyAlchemyEvmProviderAdapter` | `evmProvider` | Privy-managed wallets with Alchemy infrastructure | +| `ViemProviderAdapter` | `evmProvider` | A viem account you hold the key for | +| `PrivySolanaProviderAdapter` | `solanaProvider` | Privy-managed Solana wallets | +| `SolanaProviderAdapter` | `solanaProvider` | Solana with a signer you supply | + +`AcpAgent.create()` takes **`evmProvider`, `solanaProvider`, or both** — there is +no plain `provider` key. Passing one throws at runtime ("AcpAgent.create() has no +`provider` option"), and TypeScript only catches it when the adapter is an inline +object literal. ```typescript -// Privy + Alchemy -const provider = await PrivyAlchemyEvmProviderAdapter.create({ - walletAddress: "0x...", - walletId: "your-privy-wallet-id", - chains: [base], - signerPrivateKey: "your-privy-signer-private-key", +// EVM -- Privy + Alchemy +const agent = await AcpAgent.create({ + evmProvider: await PrivyAlchemyEvmProviderAdapter.create({ + walletAddress: process.env.WALLET_ADDRESS as `0x${string}`, // typed 0x${string} + walletId: "your-privy-wallet-id", + chains: [base], + signerPrivateKey: "your-privy-signer-private-key", + }), }); + +// Solana -- Privy +const solanaAgent = await AcpAgent.create({ + solanaProvider: await PrivySolanaProviderAdapter.create({ + walletAddress: process.env.SOLANA_WALLET_ADDRESS!, // plain string, no cast + walletId: "your-privy-wallet-id", + signerPrivateKey: "your-privy-signer-private-key", + chainId: 501, + }), +}); + +// Both -- one agent serving EVM and Solana jobs +const multiChain = await AcpAgent.create({ evmProvider, solanaProvider }); ``` +`PrivyAlchemyChainConfig.walletAddress` is viem's `Address`, a template literal +type. An inline literal starting with `0x` satisfies it, but anything read from +`process.env` is a plain `string` and needs a cast: + +```typescript +walletAddress: process.env.WALLET_ADDRESS as `0x${string}`, +``` + +The Solana adapters take a plain `string` and need no cast. See any file under +[`src/examples/`](./src/examples/) for the pattern. + All EVM provider adapters implement the `IEvmProviderAdapter` interface, which includes: - `sendCalls(chainId, calls)` — Submit transactions @@ -408,7 +600,10 @@ All EVM provider adapters implement the `IEvmProviderAdapter` interface, which i For jobs that involve transferring funds to the provider on submission: ```typescript -// Buyer: create a fund transfer job +// Buyer: create a fund transfer job. +// Like every raw creator, this sends no requirement message -- follow it with +// agent.sendMessage(..., "requirement"), or use createJobFromOffering when the +// offering has requiredFunds set. See "The requirement message". const jobId = await agent.createFundTransferJob(base.id, { providerAddress: SELLER_ADDRESS, evaluatorAddress: buyerAddress, @@ -467,6 +662,17 @@ See [migration.md](./migration.md) for a full migration guide with side-by-side We welcome contributions. Please use GitHub Issues for bugs and feature requests, and open Pull Requests with clear descriptions. +Before opening a PR, run both type-checks: + +```bash +npm run typecheck # src/, excluding examples (what ships in dist/) +npm run typecheck:examples # src/ including src/examples/ +``` + +The publish build excludes `src/examples/` so it never lands in `dist/`, which +also means `npm run build` will not catch a broken example. If you touch +anything under `src/examples/`, the second command is the one that matters. + **Community:** [Discord](https://discord.gg/virtualsio) | [Telegram](https://t.me/virtuals) | [X (Twitter)](https://x.com/virtuals_io) ## Useful Resources diff --git a/package-lock.json b/package-lock.json index 43995af..f797dae 100644 --- a/package-lock.json +++ b/package-lock.json @@ -21,6 +21,7 @@ "viem": "^2.47.0" }, "devDependencies": { + "@anthropic-ai/sdk": "^0.117.1", "@codama/nodes-from-anchor": "^1.4.0", "@codama/renderers-js": "^1.6.2", "@types/node": "^25.3.3", @@ -204,12 +205,34 @@ } } }, + "node_modules/@anthropic-ai/sdk": { + "version": "0.117.1", + "resolved": "https://registry.npmjs.org/@anthropic-ai/sdk/-/sdk-0.117.1.tgz", + "integrity": "sha512-Yn2QlXfyCiKJ5YGCOOay7ZE78ISvII2XY621WMCiflmG8IYgwx59IBwPExxki3Xk9jKUtnD/Sj6UvplWr0rZxg==", + "dev": true, + "license": "MIT", + "dependencies": { + "json-schema-to-ts": "^3.1.1", + "standardwebhooks": "^1.0.0" + }, + "bin": { + "anthropic-ai-sdk": "bin/cli" + }, + "peerDependencies": { + "zod": "^3.25.0 || ^4.0.0" + }, + "peerDependenciesMeta": { + "zod": { + "optional": true + } + } + }, "node_modules/@babel/runtime": { "version": "7.28.6", "resolved": "https://registry.npmjs.org/@babel/runtime/-/runtime-7.28.6.tgz", "integrity": "sha512-05WQkdpL9COIMz4LjTxGpPNCdlpyimKppYNoJ5Di5EUObifl8t4tuLuUBBZEpoLYOmfvIWrsp9fCl0HoPRVTdA==", + "devOptional": true, "license": "MIT", - "optional": true, "engines": { "node": ">=6.9.0" } @@ -5834,6 +5857,21 @@ "license": "MIT", "optional": true }, + "node_modules/jayson/node_modules/utf-8-validate": { + "version": "5.0.10", + "resolved": "https://registry.npmjs.org/utf-8-validate/-/utf-8-validate-5.0.10.tgz", + "integrity": "sha512-Z6czzLq4u8fPOyx7TU6X3dvUZVvoJmxSQ+IcrlmagKhilxlhZgxPK6C5Jqbkw1IDUmFTM+cz9QDnnLTwDz/2gQ==", + "hasInstallScript": true, + "license": "MIT", + "optional": true, + "peer": true, + "dependencies": { + "node-gyp-build": "^4.3.0" + }, + "engines": { + "node": ">=6.14.2" + } + }, "node_modules/jayson/node_modules/uuid": { "version": "8.3.2", "resolved": "https://registry.npmjs.org/uuid/-/uuid-8.3.2.tgz", @@ -5891,6 +5929,20 @@ "license": "MIT", "optional": true }, + "node_modules/json-schema-to-ts": { + "version": "3.1.1", + "resolved": "https://registry.npmjs.org/json-schema-to-ts/-/json-schema-to-ts-3.1.1.tgz", + "integrity": "sha512-+DWg8jCJG2TEnpy7kOm/7/AxaYoaRbjVB4LFZLySZlWn8exGs3A4OLJR966cVvU26N7X9TWxl+Jsw7dzAqKT6g==", + "dev": true, + "license": "MIT", + "dependencies": { + "@babel/runtime": "^7.18.3", + "ts-algebra": "^2.0.0" + }, + "engines": { + "node": ">=16" + } + }, "node_modules/json-schema-traverse": { "version": "1.0.0", "resolved": "https://registry.npmjs.org/json-schema-traverse/-/json-schema-traverse-1.0.0.tgz", @@ -6595,6 +6647,13 @@ "integrity": "sha512-N3WMsuqV66lT30CrXNbEjx4GEwlow3v6rr4mCcv6prnfwhS01rkgyFdjPNBYd9br7LpXV1+Emh01fHnq2Gdgrw==", "license": "MIT" }, + "node_modules/ts-algebra": { + "version": "2.0.0", + "resolved": "https://registry.npmjs.org/ts-algebra/-/ts-algebra-2.0.0.tgz", + "integrity": "sha512-FPAhNPFMrkwz76P7cdjdmiShwMynZYN6SgOujD1urY4oNm80Ou9oMdmbR45LotcKOXoy7wSmHkRFE6Mxbrhefw==", + "dev": true, + "license": "MIT" + }, "node_modules/tslib": { "version": "2.8.1", "resolved": "https://registry.npmjs.org/tslib/-/tslib-2.8.1.tgz", diff --git a/package.json b/package.json index e864d01..c3198d5 100644 --- a/package.json +++ b/package.json @@ -7,6 +7,8 @@ "prepare": "tsc", "test": "echo \"Error: no test specified\" && exit 1", "build": "tsc", + "typecheck": "tsc --noEmit", + "typecheck:examples": "tsc -p tsconfig.examples.json", "dev": "tsx watch src/index.ts", "start": "node dist/index.js" }, @@ -29,6 +31,7 @@ "viem": "^2.47.0" }, "devDependencies": { + "@anthropic-ai/sdk": "^0.117.1", "@codama/nodes-from-anchor": "^1.4.0", "@codama/renderers-js": "^1.6.2", "@types/node": "^25.3.3", diff --git a/src/acpAgent.ts b/src/acpAgent.ts index 2481a0d..7e8f4f0 100644 --- a/src/acpAgent.ts +++ b/src/acpAgent.ts @@ -360,6 +360,26 @@ export class AcpAgent { }; } + /** + * Connect to the event stream, then hydrate a session for every in-flight job + * this wallet participates in. + * + * **Your `entry` handler must be idempotent.** Hydration re-fires the handler + * with the *latest* entry of each active job, so every `start()` replays + * whatever the job was last waiting on. Restart an evaluator while a job sits + * at `job.submitted` and your handler is called for that submission again — it + * will try to `complete()` a job it already ruled on. The contract rejects the + * redundant call, but don't rely on that as your control: an on-chain revert + * is not a deduplication mechanism, and hooks or fee transfers reached before + * the revert are not free. + * + * The SDK deliberately does not dedupe this — replay is what lets an agent + * killed mid-flow pick the job back up. Since delivery survives process + * restarts, so must your record of what you've acted on: persist a key like + * `${chainId}-${jobId}-${event.type}` before the side effect and skip entries + * you've already handled. An in-memory `Set` is lost on exactly the restart + * that triggers the replay. + */ async start( onConnected?: () => void, streams: SupportedStreams[] = DEFAULT_STREAMS, @@ -612,6 +632,32 @@ export class AcpAgent { // Job creation (on-chain, room is created by the observer) // ------------------------------------------------------------------------- + /** + * Create a job on-chain and nothing else. + * + * **This does not send a requirement message.** The job carries only + * `params.description`, so the provider never receives the structured ask and + * an evaluator has a deliverable with nothing to judge it against. Only + * {@link createJobFromOffering} / {@link createJobByOfferingName} post the + * `"requirement"` entry. + * + * If you create jobs through this path, send the requirement yourself right + * after: + * + * ```ts + * const jobId = await agent.createJob(chainId, params); + * await agent.sendMessage( + * chainId, + * jobId.toString(), + * JSON.stringify({ key: "..." }), + * "requirement", + * ); + * ``` + * + * Prefer the offering-based creators unless you need a job that isn't backed + * by a registry offering — they validate the requirement against the + * offering's JSON schema and derive `expiredAt` from its SLA. + */ async createJob(chainId: number, params: CreateJobParams): Promise { const client = this.getClient(chainId); // On Solana, createJob precomputes the job PDA from acp_state.job_counter @@ -633,6 +679,12 @@ export class AcpAgent { return jobId; } + /** + * {@link createJob} with the FundTransferHook attached by default. + * + * Like `createJob`, this sends **no requirement message** — see that method + * for why that matters and how to send one yourself. + */ async createFundTransferJob( chainId: number, params: CreateJobParams, @@ -648,6 +700,12 @@ export class AcpAgent { }); } + /** + * {@link createJob} with the SubscriptionHook attached by default. + * + * Like `createJob`, this sends **no requirement message** — see that method + * for why that matters and how to send one yourself. + */ async createSubscriptionJob( chainId: number, params: CreateJobParams, @@ -663,6 +721,13 @@ export class AcpAgent { }); } + /** + * {@link createJob} routed through the MultiHookRouter, optionally + * configuring the per-selector hook layout in the same call. + * + * Like `createJob`, this sends **no requirement message** — see that method + * for why that matters and how to send one yourself. + */ async createMultiHookJob( chainId: number, params: CreateJobParams, diff --git a/src/clientFactory.ts b/src/clientFactory.ts index bd30212..634ce81 100644 --- a/src/clientFactory.ts +++ b/src/clientFactory.ts @@ -20,6 +20,17 @@ export async function createAcpClients( ): Promise> { const { evmProvider, solanaProvider } = input; if (!evmProvider && !solanaProvider) { + // `provider` is the most common miss: it reads naturally, TypeScript only + // catches it on an inline object literal (excess property check), and plain + // JS consumers get no signal at all. Name the right key instead of letting + // them re-read the type defs. + if ("provider" in input) { + throw new Error( + "AcpAgent.create() has no `provider` option. Use `evmProvider` for an " + + "EVM adapter (e.g. PrivyAlchemyEvmProviderAdapter) or `solanaProvider` " + + "for a Solana adapter (e.g. PrivySolanaProviderAdapter).", + ); + } throw new Error( "At least one provider (evmProvider or solanaProvider) must be provided.", ); diff --git a/src/examples/basic/buyer.ts b/src/examples/basic/buyer.ts index 9c1c4df..d95bcac 100644 --- a/src/examples/basic/buyer.ts +++ b/src/examples/basic/buyer.ts @@ -128,7 +128,7 @@ function promptYesNo(question: string, defaultYes: boolean): Promise { async function main(): Promise { const buyer = await AcpAgent.create({ - provider: await PrivyAlchemyEvmProviderAdapter.create({ + evmProvider: await PrivyAlchemyEvmProviderAdapter.create({ walletAddress: requireEnv("BUYER_WALLET_ADDRESS") as `0x${string}`, walletId: requireEnv("BUYER_WALLET_ID"), signerPrivateKey: requireEnv("BUYER_SIGNER_PRIVATE_KEY"), diff --git a/src/examples/basic/seller.ts b/src/examples/basic/seller.ts index 2b1f058..23927c6 100644 --- a/src/examples/basic/seller.ts +++ b/src/examples/basic/seller.ts @@ -80,7 +80,7 @@ function requireEnv(name: string): string { async function main(): Promise { const seller = await AcpAgent.create({ - provider: await PrivyAlchemyEvmProviderAdapter.create({ + evmProvider: await PrivyAlchemyEvmProviderAdapter.create({ walletAddress: requireEnv("SELLER_WALLET_ADDRESS") as `0x${string}`, walletId: requireEnv("SELLER_WALLET_ID"), signerPrivateKey: requireEnv("SELLER_SIGNER_PRIVATE_KEY"), diff --git a/src/examples/fund-transfer/buyer.ts b/src/examples/fund-transfer/buyer.ts index 3fd2b8c..3c2b5e4 100644 --- a/src/examples/fund-transfer/buyer.ts +++ b/src/examples/fund-transfer/buyer.ts @@ -98,7 +98,7 @@ const log = { async function main(): Promise { const buyer = await AcpAgent.create({ - provider: await PrivyAlchemyEvmProviderAdapter.create({ + evmProvider: await PrivyAlchemyEvmProviderAdapter.create({ walletAddress: requireEnv("BUYER_WALLET_ADDRESS") as `0x${string}`, walletId: requireEnv("BUYER_WALLET_ID"), signerPrivateKey: requireEnv("BUYER_SIGNER_PRIVATE_KEY"), diff --git a/src/examples/fund-transfer/seller.ts b/src/examples/fund-transfer/seller.ts index 4680fbb..7750b19 100644 --- a/src/examples/fund-transfer/seller.ts +++ b/src/examples/fund-transfer/seller.ts @@ -105,7 +105,7 @@ function readForwardFromRequirement( async function main(): Promise { const seller = await AcpAgent.create({ - provider: await PrivyAlchemyEvmProviderAdapter.create({ + evmProvider: await PrivyAlchemyEvmProviderAdapter.create({ walletAddress: requireEnv("SELLER_WALLET_ADDRESS") as `0x${string}`, walletId: requireEnv("SELLER_WALLET_ID"), signerPrivateKey: requireEnv("SELLER_SIGNER_PRIVATE_KEY"), diff --git a/src/examples/llm/README.md b/src/examples/llm/README.md index b3400b1..f338ba1 100644 --- a/src/examples/llm/README.md +++ b/src/examples/llm/README.md @@ -24,7 +24,7 @@ agent.on("entry", async (session, entry) => { const messages = await session.toMessages(); // history → chat format const response = await anthropic.messages.create({ - model: "claude-sonnet-4-…", + model: "claude-opus-5", system: "You are a … agent", messages: formatMessages(messages), tools: formatTools(tools), diff --git a/src/examples/llm/buyer.ts b/src/examples/llm/buyer.ts index 54f1191..e22bc5b 100644 --- a/src/examples/llm/buyer.ts +++ b/src/examples/llm/buyer.ts @@ -146,7 +146,7 @@ function toAnthropicMessages( async function main(): Promise { const buyer = await AcpAgent.create({ - provider: await PrivyAlchemyEvmProviderAdapter.create({ + evmProvider: await PrivyAlchemyEvmProviderAdapter.create({ walletAddress: requireEnv("BUYER_WALLET_ADDRESS") as `0x${string}`, walletId: requireEnv("BUYER_WALLET_ID"), signerPrivateKey: requireEnv("BUYER_SIGNER_PRIVATE_KEY"), @@ -248,7 +248,7 @@ async function main(): Promise { if (messages.length === 0) return; const response = await anthropic.messages.create({ - model: "gemini-3.1-flash-lite-preview", + model: "claude-opus-5", max_tokens: 1024, system: SYSTEM_PROMPT, messages, @@ -403,7 +403,7 @@ async function pickOfferingWithLlm( }; const response = await anthropic.messages.create({ - model: "gemini-3.1-flash-lite-preview", + model: "claude-opus-5", max_tokens: 1024, system: SYSTEM_PROMPT, messages: [ diff --git a/src/examples/llm/seller.ts b/src/examples/llm/seller.ts index f2b0378..b5b3d08 100644 --- a/src/examples/llm/seller.ts +++ b/src/examples/llm/seller.ts @@ -159,7 +159,7 @@ function offeringContextNote( async function main(): Promise { const seller = await AcpAgent.create({ - provider: await PrivyAlchemyEvmProviderAdapter.create({ + evmProvider: await PrivyAlchemyEvmProviderAdapter.create({ walletAddress: requireEnv("SELLER_WALLET_ADDRESS") as `0x${string}`, walletId: requireEnv("SELLER_WALLET_ID"), signerPrivateKey: requireEnv("SELLER_SIGNER_PRIVATE_KEY"), @@ -261,7 +261,7 @@ async function main(): Promise { if (messages.length === 0) return; const response = await anthropic.messages.create({ - model: "gemini-3.1-flash-lite-preview", + model: "claude-opus-5", max_tokens: 1024, system: systemPrompt, messages, diff --git a/src/examples/subscription-fund-transfer/buyer.ts b/src/examples/subscription-fund-transfer/buyer.ts index 0ea405b..e9d3406 100644 --- a/src/examples/subscription-fund-transfer/buyer.ts +++ b/src/examples/subscription-fund-transfer/buyer.ts @@ -96,7 +96,7 @@ function promptYesNo(question: string, defaultYes: boolean): Promise { async function main(): Promise { const buyer = await AcpAgent.create({ - provider: await PrivyAlchemyEvmProviderAdapter.create({ + evmProvider: await PrivyAlchemyEvmProviderAdapter.create({ walletAddress: requireEnv("BUYER_WALLET_ADDRESS") as `0x${string}`, walletId: requireEnv("BUYER_WALLET_ID"), signerPrivateKey: requireEnv("BUYER_SIGNER_PRIVATE_KEY"), diff --git a/src/examples/subscription-fund-transfer/seller.ts b/src/examples/subscription-fund-transfer/seller.ts index 6652368..7dea420 100644 --- a/src/examples/subscription-fund-transfer/seller.ts +++ b/src/examples/subscription-fund-transfer/seller.ts @@ -89,7 +89,7 @@ function requireEnv(name: string): string { async function main(): Promise { const seller = await AcpAgent.create({ - provider: await PrivyAlchemyEvmProviderAdapter.create({ + evmProvider: await PrivyAlchemyEvmProviderAdapter.create({ walletAddress: requireEnv("SELLER_WALLET_ADDRESS") as `0x${string}`, walletId: requireEnv("SELLER_WALLET_ID"), signerPrivateKey: requireEnv("SELLER_SIGNER_PRIVATE_KEY"), diff --git a/src/examples/subscription/buyer.ts b/src/examples/subscription/buyer.ts index 348d393..79b5e66 100644 --- a/src/examples/subscription/buyer.ts +++ b/src/examples/subscription/buyer.ts @@ -95,7 +95,7 @@ function promptYesNo(question: string, defaultYes: boolean): Promise { async function main(): Promise { const buyer = await AcpAgent.create({ - provider: await PrivyAlchemyEvmProviderAdapter.create({ + evmProvider: await PrivyAlchemyEvmProviderAdapter.create({ walletAddress: requireEnv("BUYER_WALLET_ADDRESS") as `0x${string}`, walletId: requireEnv("BUYER_WALLET_ID"), signerPrivateKey: requireEnv("BUYER_SIGNER_PRIVATE_KEY"), diff --git a/src/examples/subscription/seller.ts b/src/examples/subscription/seller.ts index 56c1190..4c83ef8 100644 --- a/src/examples/subscription/seller.ts +++ b/src/examples/subscription/seller.ts @@ -80,7 +80,7 @@ function requireEnv(name: string): string { async function main(): Promise { const seller = await AcpAgent.create({ - provider: await PrivyAlchemyEvmProviderAdapter.create({ + evmProvider: await PrivyAlchemyEvmProviderAdapter.create({ walletAddress: requireEnv("SELLER_WALLET_ADDRESS") as `0x${string}`, walletId: requireEnv("SELLER_WALLET_ID"), signerPrivateKey: requireEnv("SELLER_SIGNER_PRIVATE_KEY"), diff --git a/tsconfig.examples.json b/tsconfig.examples.json new file mode 100644 index 0000000..30c30ce --- /dev/null +++ b/tsconfig.examples.json @@ -0,0 +1,16 @@ +// Type-checks src/examples/, which the publish build (tsconfig.json) excludes. +// +// The examples are the code people copy first, so a compile error there is a +// docs bug with a runtime cost. tsconfig.json excludes them on purpose — they +// must not ship in dist/ — which meant nothing type-checked them and every EVM +// example drifted to a `provider:` key that `AcpAgent.create` never accepted. +// +// Run via `npm run typecheck:examples` (noEmit, so nothing lands in dist/). +{ + "extends": "./tsconfig.json", + "compilerOptions": { + "noEmit": true + }, + "include": ["src/**/*.ts"], + "exclude": ["scripts", "dist", "tests"] +} From 4fbec5be7e41586829c89c785f84de315a7c5d07 Mon Sep 17 00:00:00 2001 From: Joey Lau Date: Wed, 19 Aug 2026 15:13:25 +0800 Subject: [PATCH 2/4] fix: deliver each job room entry exactly once within a process `start()` registers the live entry handler and awaits `transport.connect()` before `hydrateSessions()`. The stream has to be live first -- otherwise entries occurring during catch-up are lost outright -- but dispatching them immediately is worse, because hydration has not built the session yet. Two deterministic failures came out of that window, both reproducible in a single process with no restart involved: - provider role: the handler ran *twice* for one `job.funded`. The live entry dispatched against a session built from no history, then hydration re-delivered the same entry as the job's latest. Two submits for one funding; for an evaluator, two rulings for one submission. - evaluator role: the handler ran *zero* times. A first sighting that isn't `job.created` has no role information, so `inferRoles([])` fell back to ["provider"] and `shouldRespond` dropped the `job.submitted`. A ruling silently never happened. The session also kept 1 of 2 history entries, so `toContext()` and `status` were computed on partial history. Entry identity was the underlying problem: `dispatch` tested membership with `session.entries.includes(entry)`, a reference compare, and the same logical entry arrives as a different object on every path that produces it -- an SSE frame and a `getHistory()` response each parse their own copy. Fixes: - `entryKey()` (new, exported): stable content key for a `JobRoomEntry`, since the type carries no server-assigned id. - `AcpAgent.start()`: queue live entries while `hydrating` is set, drain after hydration in a `finally` so a hydration failure can't strand them. - `JobSession.appendEntry()` is idempotent and returns whether the entry was new; `hasEntry()` and `mergeEntries()` added. `mergeEntries` sorts by timestamp because `status` scans backwards for the newest system event, so folding older history in after a newer live entry would walk status backwards. - `dispatch()`: a first sighting that isn't `job.created` fetches history so roles resolve; failure to fetch is logged rather than swallowed and the entry is still delivered. - `getOrCreateSession()` merges supplied history into an existing session instead of discarding it. Replay across process restarts is unchanged and still deliberate: it is what lets an agent killed mid-flow pick the job back up. Integrators still need persistent dedup for that, and the key set here dies with the process by design. Adds tests/entryDelivery.test.ts -- 7 cases over fake transports, no framework, no network. Run via `npm test` (previously a stub that exited 1), type-checked via `npm run typecheck:tests` with tsconfig.tests.json. Against the previous commit the suite fails 5/7; the 2 that pass before and after are the ones asserting preserved behavior (restart replay, the `job.created` path), so the change is provably narrow. tests/ stays out of dist/. Co-Authored-By: Claude Opus 5 (1M context) --- package.json | 3 +- src/acpAgent.ts | 110 +++++++++++-- src/events/entryKey.ts | 22 +++ src/index.ts | 1 + src/jobSession.ts | 44 ++++- tests/entryDelivery.test.ts | 311 ++++++++++++++++++++++++++++++++++++ tsconfig.tests.json | 17 ++ 7 files changed, 491 insertions(+), 17 deletions(-) create mode 100644 src/events/entryKey.ts create mode 100644 tests/entryDelivery.test.ts create mode 100644 tsconfig.tests.json diff --git a/package.json b/package.json index c3198d5..3bf79f5 100644 --- a/package.json +++ b/package.json @@ -5,10 +5,11 @@ "main": "dist/index.js", "scripts": { "prepare": "tsc", - "test": "echo \"Error: no test specified\" && exit 1", + "test": "for f in tests/*.test.ts; do tsx \"$f\" || exit 1; done", "build": "tsc", "typecheck": "tsc --noEmit", "typecheck:examples": "tsc -p tsconfig.examples.json", + "typecheck:tests": "tsc -p tsconfig.tests.json", "dev": "tsx watch src/index.ts", "start": "node dist/index.js" }, diff --git a/src/acpAgent.ts b/src/acpAgent.ts index 7e8f4f0..04ab910 100644 --- a/src/acpAgent.ts +++ b/src/acpAgent.ts @@ -180,6 +180,12 @@ export class AcpAgent { private entryHandler: EntryHandler | null = null; private sessionMap = new Map(); private addresses = new Map(); + /** + * True from `start()` until hydration finishes. Live entries are queued rather + * than dispatched during that window — see `start()`. + */ + private hydrating = false; + private pendingEntries: JobRoomEntry[] = []; constructor( clients: Map, @@ -373,11 +379,13 @@ export class AcpAgent { * is not a deduplication mechanism, and hooks or fee transfers reached before * the revert are not free. * - * The SDK deliberately does not dedupe this — replay is what lets an agent - * killed mid-flow pick the job back up. Since delivery survives process - * restarts, so must your record of what you've acted on: persist a key like - * `${chainId}-${jobId}-${event.type}` before the side effect and skip entries - * you've already handled. An in-memory `Set` is lost on exactly the restart + * Within one process each entry is delivered once — sessions track entries by + * content key (`entryKey`), so a stream reconnect or an entry landing + * mid-hydration can't fire the handler twice. Across restarts it can, and + * deliberately so: that replay is what lets an agent killed mid-flow pick the + * job back up. Since delivery survives restarts, so must your record of what + * you've acted on — persist a key like `${chainId}-${jobId}-${event.type}` + * before the side effect. An in-memory `Set` is lost on exactly the restart * that triggers the replay. */ async start( @@ -389,13 +397,50 @@ export class AcpAgent { } this.started = true; - - this.transport.onEntry((entry) => - this.dispatch(entry).catch(console.error), - ); + this.hydrating = true; + + // The stream has to be live before hydration, or entries occurring mid-catch-up + // are lost outright. But dispatching them right away is worse: hydration hasn't + // built the session yet, so the entry would create one from no history — + // `inferRoles([])` silently defaults to ["provider"], and hydration then + // re-delivers the same entry as the job's latest. Queue instead, and drain + // once every session has its history. + this.transport.onEntry((entry) => { + if (this.hydrating) { + this.pendingEntries.push(entry); + return; + } + this.dispatch(entry).catch(console.error); + }); await this.transport.connect(onConnected, streams); - await this.hydrateSessions(); + try { + await this.hydrateSessions(); + } finally { + // Drain even if hydration threw: the queue is the only path those entries + // have, and `dispatch` can rebuild a session on its own. + this.hydrating = false; + await this.drainPendingEntries(); + } + } + + /** + * Dispatch entries that arrived while hydration was running. + * + * Anything hydration already replayed is dropped by `dispatch` itself, which + * ignores an entry the session has seen — so an entry that landed in the + * window reaches the handler exactly once, not twice and not never. + */ + private async drainPendingEntries(): Promise { + const queued = this.pendingEntries; + this.pendingEntries = []; + for (const entry of queued) { + try { + await this.dispatch(entry); + } catch (err) { + console.error(err); + } + } } async stop(): Promise { @@ -403,6 +448,8 @@ export class AcpAgent { await this.transport.disconnect(); this.started = false; } + this.hydrating = false; + this.pendingEntries = []; this.sessionMap.clear(); } @@ -476,7 +523,12 @@ export class AcpAgent { initialEntries: JobRoomEntry[] = [], ): JobSession { let session = this.sessionMap.get(this.getSessionKey(chainId, jobId)); - if (session) return session; + if (session) { + // A session built from a single live entry knows almost nothing about the + // job; fold in whatever history the caller has rather than discarding it. + if (initialEntries.length > 0) session.mergeEntries(initialEntries); + return session; + } const roles = this.inferRoles(initialEntries); session = new JobSession( @@ -520,12 +572,42 @@ export class AcpAgent { private async dispatch(entry: JobRoomEntry): Promise { const jobId = entry.onChainJobId; const chainId = entry.chainId; - const session = this.getOrCreateSession(jobId, chainId, []); - if (session.entries.length === 0 || !session.entries.includes(entry)) { - session.appendEntry(entry); + let session = this.getSession(chainId, jobId); + let sessionIsNew = false; + if (!session) { + // `job.created` carries the client/provider/evaluator addresses, so it can + // stand up a session on its own. Any other first-sighting cannot: roles + // come from the creation event, and without it `inferRoles` falls back to + // ["provider"] — an evaluator would quietly stop responding to the job. + const isCreation = + entry.kind === "system" && entry.event.type === "job.created"; + let history: JobRoomEntry[] = []; + if (!isCreation) { + try { + history = await this.transport.getHistory(chainId, jobId); + } catch (err) { + // Carry on with the single entry rather than dropping it, but say so: + // without history the session falls back to the "provider" role and + // may ignore entries it should have handled. + console.error( + `Failed to fetch history for job ${jobId} on chain ${chainId}; ` + + `session roles may be incomplete`, + err, + ); + } + } + session = this.getOrCreateSession(jobId, chainId, history); + sessionIsNew = true; } + // A known session already holding this entry means another path (hydration, + // or a reconnect replay) delivered it — don't fire twice. On a session we + // just built, the entry being present only means our own history fetch + // included it, and nothing has fired for it yet. + const entryIsNew = session.appendEntry(entry); + if (!entryIsNew && !sessionIsNew) return; + if (entry.kind === "system" && entry.event.type === "job.created") { const roles = this.inferRoles([entry]); const rolesChanged = diff --git a/src/events/entryKey.ts b/src/events/entryKey.ts new file mode 100644 index 0000000..e2dd8c0 --- /dev/null +++ b/src/events/entryKey.ts @@ -0,0 +1,22 @@ +import type { JobRoomEntry } from "./types.js"; + +/** + * Stable content key for a room entry. + * + * `JobRoomEntry` carries no server-assigned id, and the same logical entry + * reaches the SDK as a different object on every path that produces it — an SSE + * frame and a `getHistory()` response each `JSON.parse` their own copy. Compare + * by reference and one entry looks like several, which is what lets a handler + * run twice on a single event. + * + * This answers "is this the entry I already have?" for one process. It is not a + * substitute for the persistent dedup an agent needs across restarts — see + * "Restart & replay semantics" in the README. That store records what you + * *acted on*; this records what you've *seen*, and dies with the process. + */ +export function entryKey(entry: JobRoomEntry): string { + const base = `${entry.chainId}:${entry.onChainJobId}:${entry.timestamp}:${entry.kind}`; + return entry.kind === "message" + ? `${base}:${entry.from.toLowerCase()}:${entry.contentType}:${entry.content}` + : `${base}:${entry.event.type}`; +} diff --git a/src/index.ts b/src/index.ts index 0da6b6b..cb9aef0 100644 --- a/src/index.ts +++ b/src/index.ts @@ -40,6 +40,7 @@ export type { JobStateDiagnosis } from "./core/solana/jobStateRetryGuard.js"; export { AcpHttpClient } from "./events/acpHttpClient.js"; export { AcpApiClient } from "./events/acpApiClient.js"; export { SseTransport, STREAMS } from "./events/sseTransport.js"; +export { entryKey } from "./events/entryKey.js"; // Public enums export { AcpJobStatus } from "./events/types.js"; diff --git a/src/jobSession.ts b/src/jobSession.ts index e7c32b4..c31cb7f 100644 --- a/src/jobSession.ts +++ b/src/jobSession.ts @@ -6,6 +6,7 @@ import type { AgentRole, AcpJobEventType, } from "./events/types.js"; +import { entryKey } from "./events/entryKey.js"; import type { AcpAgent } from "./acpAgent.js"; import { AcpJob } from "./acpJob.js"; import { AssetToken } from "./core/assetToken.js"; @@ -176,6 +177,8 @@ export class JobSession { private _job: AcpJob | null = null; private readonly agent: AcpAgent; private readonly agentAddresses: Set; + /** Keys of everything in `entries`, so membership is content- not reference-based. */ + private readonly entryKeys = new Set(); constructor( agent: AcpAgent, @@ -190,7 +193,7 @@ export class JobSession { this.jobId = jobId; this.chainId = chainId; this.roles = roles; - this.entries.push(...initialEntries); + for (const entry of initialEntries) this.appendEntry(entry); } get job(): AcpJob | null { @@ -216,8 +219,45 @@ export class JobSession { // Entry management // ------------------------------------------------------------------------- - appendEntry(entry: JobRoomEntry): void { + /** True if this exact entry is already in `entries` (compared by content). */ + hasEntry(entry: JobRoomEntry): boolean { + return this.entryKeys.has(entryKey(entry)); + } + + /** + * Append an entry, ignoring one already present. + * + * Idempotent on purpose: hydration and the live stream can both produce the + * same entry as separate objects, and appending it twice would duplicate the + * transcript. Returns whether the entry was new — callers use that to decide + * whether the handler still owes a delivery for it. + */ + appendEntry(entry: JobRoomEntry): boolean { + const key = entryKey(entry); + if (this.entryKeys.has(key)) return false; + this.entryKeys.add(key); this.entries.push(entry); + return true; + } + + /** + * Fold fetched history into this session, keeping `entries` ordered by + * timestamp. + * + * Order matters beyond tidiness: `status` reads backwards for the newest + * system event, so appending older history after a newer live entry would + * walk the job's status backwards. + */ + mergeEntries(entries: JobRoomEntry[]): number { + const fresh = entries.filter((e) => !this.hasEntry(e)); + if (fresh.length === 0) return 0; + for (const entry of fresh) { + this.entryKeys.add(entryKey(entry)); + this.entries.push(entry); + } + // Array#sort is stable, so entries sharing a timestamp keep arrival order. + this.entries.sort((a, b) => a.timestamp - b.timestamp); + return fresh.length; } // ------------------------------------------------------------------------- diff --git a/tests/entryDelivery.test.ts b/tests/entryDelivery.test.ts new file mode 100644 index 0000000..2cb5014 --- /dev/null +++ b/tests/entryDelivery.test.ts @@ -0,0 +1,311 @@ +/** + * Delivery guarantees for the `entry` handler. + * + * Covers the ordering inside `AcpAgent.start()`: the stream goes live before + * hydration finishes, so entries can arrive while sessions are still being + * rebuilt. Every case here is about a handler running exactly once — never + * twice (a second on-chain ruling) and never zero times (a job left hanging). + * + * Run with `npm test`. No framework: plain assertions over fake transports, so + * nothing here touches the network or a chain. + */ +import assert from "node:assert/strict"; +import { AcpAgent } from "../src/acpAgent.js"; +import { AcpJobStatus } from "../src/events/types.js"; +import type { + AcpAgentDetail, + AcpChatTransport, + AcpJobApi, + JobRoomEntry, + OffChainJob, + SupportedStreams, +} from "../src/events/types.js"; + +const CHAIN = 8453; +const JOB = "4242"; +const CLIENT = "0x1111111111111111111111111111111111111111"; +const PROVIDER = "0x2222222222222222222222222222222222222222"; +const EVALUATOR = "0x3333333333333333333333333333333333333333"; + +const tick = (ms = 25): Promise => + new Promise((resolve) => setTimeout(resolve, ms)); + +// Fresh object per call throughout: the SSE frame and the getHistory response +// are parsed separately, so the same logical entry is never the same reference. +const createdEntry = (): JobRoomEntry => ({ + kind: "system", + chainId: CHAIN, + onChainJobId: JOB, + timestamp: 1_000, + event: { + type: "job.created", + onChainJobId: JOB, + client: CLIENT, + provider: PROVIDER, + evaluator: EVALUATOR, + }, +} as unknown as JobRoomEntry); + +const fundedEntry = (): JobRoomEntry => ({ + kind: "system", + chainId: CHAIN, + onChainJobId: JOB, + timestamp: 2_000, + event: { type: "job.funded", onChainJobId: JOB, client: CLIENT, amount: 0.01 }, +} as unknown as JobRoomEntry); + +const submittedEntry = (): JobRoomEntry => ({ + kind: "system", + chainId: CHAIN, + onChainJobId: JOB, + timestamp: 3_000, + event: { + type: "job.submitted", + onChainJobId: JOB, + provider: PROVIDER, + deliverableHash: "0xabc", + deliverable: "the deliverable", + }, +} as unknown as JobRoomEntry); + +type FakeOpts = { + /** Entries the server pushes while `hydrateSessions()` is still running. */ + duringHydration?: () => JobRoomEntry[]; + /** What `getHistory` returns, after `historyDelayMs`. */ + history?: () => JobRoomEntry[]; + historyDelayMs?: number; + /** Jobs `getActiveJobs()` reports; empty means nothing to hydrate. */ + activeJobs?: { chainId: number; onChainJobId: string }[]; + jobStatus?: AcpJobStatus; +}; + +class FakeTransport implements AcpChatTransport { + handler: ((entry: JobRoomEntry) => void) | null = null; + historyCalls = 0; + constructor(private readonly opts: FakeOpts) {} + + async connect(onConnected?: () => void, _s?: SupportedStreams[]): Promise { + onConnected?.(); + const pushed = this.opts.duringHydration?.() ?? []; + // Land after connect() resolves, i.e. inside the hydration window. + for (const entry of pushed) setTimeout(() => this.emit(entry), 0); + } + async disconnect(): Promise {} + onEntry(handler: (entry: JobRoomEntry) => void): void { + this.handler = handler; + } + emit(entry: JobRoomEntry): void { + this.handler?.(entry); + } + sendMessage(): void {} + async postMessage(): Promise {} + async getHistory(): Promise { + this.historyCalls++; + await tick(this.opts.historyDelayMs ?? 20); + return this.opts.history?.() ?? []; + } +} + +class FakeApi implements AcpJobApi { + constructor(private readonly opts: FakeOpts) {} + async getActiveJobs() { + return this.opts.activeJobs ?? [{ chainId: CHAIN, onChainJobId: JOB }]; + } + async getJob(): Promise { + return { + chainId: CHAIN, + onChainJobId: JOB, + jobStatus: this.opts.jobStatus ?? AcpJobStatus.FUNDED, + clientAddress: CLIENT, + providerAddress: PROVIDER, + evaluatorAddress: EVALUATOR, + description: "test job", + budget: "10000", + expiredAt: new Date(Date.now() + 600_000).toISOString(), + hookAddress: null, + deliverable: "the deliverable", + hookConfigs: null, + clientSubscription: null, + }; + } + async postDeliverable(): Promise {} + async browseAgents(): Promise { + return []; + } + async getAgentByWalletAddress(): Promise { + return null; + } +} + +type Harness = { + agent: AcpAgent; + transport: FakeTransport; + /** One entry per handler invocation, in order. */ + fires: JobRoomEntry[]; + firesOf: (eventType: string) => number; +}; + +async function harness(myAddress: string, opts: FakeOpts = {}): Promise { + const transport = new FakeTransport(opts); + const agent = new AcpAgent(new Map(), transport, new FakeApi(opts)); + // buildTransportContext() normally fills this from the provider adapters. + (agent as unknown as { addresses: Map }).addresses.set( + "evm", + myAddress, + ); + const fires: JobRoomEntry[] = []; + agent.on("entry", (_session, entry) => { + fires.push(entry); + }); + await agent.start(); + await tick(60); // let queued/live dispatches settle + return { + agent, + transport, + fires, + firesOf: (eventType) => + fires.filter((e) => e.kind === "system" && e.event.type === eventType) + .length, + }; +} + +// --------------------------------------------------------------------------- + +const tests: Array<[string, () => Promise]> = [ + [ + "entry arriving mid-hydration is delivered once, with the evaluator's real role", + async () => { + const h = await harness(EVALUATOR, { + duringHydration: () => [submittedEntry()], + history: () => [createdEntry(), fundedEntry(), submittedEntry()], + jobStatus: AcpJobStatus.SUBMITTED, + }); + const session = h.agent.getSession(CHAIN, JOB)!; + // Roles come from job.created; a session built from the live entry alone + // would default to ["provider"] and drop the ruling entirely. + assert.deepEqual(session.roles, ["evaluator"]); + assert.equal(h.firesOf("job.submitted"), 1); + assert.equal(session.entries.length, 3); + assert.equal(session.status, "submitted"); + }, + ], + [ + "entry arriving mid-hydration is not delivered twice to a provider", + async () => { + const h = await harness(PROVIDER, { + duringHydration: () => [fundedEntry()], + history: () => [createdEntry(), fundedEntry()], + }); + // The pre-fix failure: live dispatch fired, then hydration replayed the + // same entry as the job's latest — two submits for one funding. + assert.equal(h.firesOf("job.funded"), 1); + assert.equal(h.agent.getSession(CHAIN, JOB)!.entries.length, 2); + }, + ], + [ + "hydration still replays the latest entry on a cold start (restart resumption)", + async () => { + const h = await harness(PROVIDER, { + history: () => [createdEntry(), fundedEntry()], + }); + // The replay is the feature: a provider killed at job.funded must be + // asked to deliver again on the next boot. + assert.equal(h.firesOf("job.funded"), 1); + assert.equal(h.firesOf("job.created"), 0, "only the latest entry replays"); + }, + ], + [ + "the same live entry delivered twice reaches the handler once", + async () => { + const h = await harness(PROVIDER, { + activeJobs: [], + history: () => [createdEntry(), fundedEntry()], + }); + h.transport.emit(createdEntry()); + await tick(); + h.transport.emit(fundedEntry()); + await tick(); + h.transport.emit(fundedEntry()); // reconnect replay, distinct object + await tick(); + assert.equal(h.firesOf("job.funded"), 1); + const session = h.agent.getSession(CHAIN, JOB)!; + assert.equal( + session.entries.filter( + (e) => e.kind === "system" && e.event.type === "job.funded", + ).length, + 1, + "transcript must not double up", + ); + }, + ], + [ + "a live job.created stands up a session without fetching history", + async () => { + const h = await harness(PROVIDER, { activeJobs: [] }); + const before = h.transport.historyCalls; + h.transport.emit(createdEntry()); + await tick(); + assert.equal(h.firesOf("job.created"), 1); + assert.deepEqual(h.agent.getSession(CHAIN, JOB)!.roles, ["provider"]); + assert.equal( + h.transport.historyCalls, + before, + "job.created carries the role addresses; no round-trip needed", + ); + }, + ], + [ + "a first sighting that isn't job.created pulls history to resolve roles", + async () => { + const h = await harness(EVALUATOR, { + activeJobs: [], + history: () => [createdEntry(), fundedEntry(), submittedEntry()], + jobStatus: AcpJobStatus.SUBMITTED, + }); + h.transport.emit(submittedEntry()); + await tick(60); + const session = h.agent.getSession(CHAIN, JOB)!; + assert.deepEqual(session.roles, ["evaluator"]); + assert.equal(h.firesOf("job.submitted"), 1, "must still fire exactly once"); + assert.equal(session.status, "submitted"); + }, + ], + [ + "merged history keeps status on the newest event, not the last one appended", + async () => { + const h = await harness(PROVIDER, { + activeJobs: [], + // History is older than the live entry that created the session. + history: () => [createdEntry(), fundedEntry()], + }); + h.transport.emit(submittedEntry()); + await tick(60); + const session = h.agent.getSession(CHAIN, JOB)!; + assert.equal( + session.status, + "submitted", + "appending older history must not walk status backwards", + ); + assert.deepEqual( + session.entries.map((e) => e.timestamp), + [1_000, 2_000, 3_000], + ); + }, + ], +]; + +let failed = 0; +for (const [name, fn] of tests) { + try { + await fn(); + console.log(` ok ${name}`); + } catch (err) { + failed++; + console.error(` FAIL ${name}`); + console.error(` ${(err as Error).message.split("\n").join("\n ")}`); + } +} +console.log( + `\n${tests.length - failed}/${tests.length} passed${failed ? ` — ${failed} FAILED` : ""}`, +); +process.exit(failed ? 1 : 0); diff --git a/tsconfig.tests.json b/tsconfig.tests.json new file mode 100644 index 0000000..5664cf6 --- /dev/null +++ b/tsconfig.tests.json @@ -0,0 +1,17 @@ +// Type-checks tests/, which the publish build (tsconfig.json) excludes so the +// suite never lands in dist/. Same reason tsconfig.examples.json exists: code +// outside the build's `include` gets no compile checking at all unless something +// else checks it, and it drifts. +// +// Run via `npm run typecheck:tests` (noEmit). +{ + "extends": "./tsconfig.json", + "compilerOptions": { + "noEmit": true, + // tests/ sits outside the build's rootDir (src). Widen it here; noEmit means + // nothing is written either way. + "rootDir": "." + }, + "include": ["tests/**/*.ts"], + "exclude": ["scripts", "dist"] +} From 82f34ade57763fa6531728e7698c8094b89698ee Mon Sep 17 00:00:00 2001 From: Joey Lau Date: Wed, 19 Aug 2026 15:13:40 +0800 Subject: [PATCH 3/4] docs: fix README snippets that don't compile, state real dedup boundary Every README code block was extracted and compiled against the packed tarball's `.d.ts`, the way a reader would use it. Three problems: - The evaluator quick start didn't compile. `session.entries.find((e) => e.kind === "message" && ...)` returns `JobRoomEntry`, so `requirement ?.content` is a type error. Narrowed with an `e is AgentMessage` predicate and added the import block so the snippet stands alone. - The Agent Discovery snippet indexed `agents[0].offerings[0]` unguarded. That's a compile error under `noUncheckedIndexedAccess` (which this repo's own tsconfig sets and `typecheck:examples` enforces), and an empty `browseAgents` result is a real runtime case -- src/examples/basic/buyer.ts already guards it. Guarded both index reads; the local is now `seller` so it no longer collides with the `provider` declared later in the same block. - `reason` on `job.completed` / `job.rejected` is typed `string`, but the value delivered is the on-chain `bytes32`: the string hex-encoded and right-padded, e.g. `"rejected"` arrives as `0x72656a6563746564...0000`. Nothing decoded it and nothing said so. Documented the actual behavior, the 32-byte truncation limit, and a `hexToString(..., { size: 32 })` decode (viem is already a dependency). Behavior itself is untouched -- decoding it in the SDK would change event payloads for existing consumers. Also corrects the restart & replay section, which claimed the SDK "does not dedupe for you" and "deliberately can't". Since the previous commit it does, within a process, by content key. The cross-restart replay it can't dedupe is still the integrator's job, so the section now says exactly where that line falls and why an in-memory set was never the answer for it. 13 of 14 typescript blocks now compile standalone; the exception is a deliberate one-line property fragment. Co-Authored-By: Claude Opus 5 (1M context) --- README.md | 52 +++++++++++++++++++++++++++++++++++++++++++++------- 1 file changed, 45 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index 0dac639..224a6f6 100644 --- a/README.md +++ b/README.md @@ -190,6 +190,14 @@ evaluator then receives `job.submitted` and decides the job's outcome. Nothing else in the lifecycle reaches it -- no `job.created`, no `budget.set`. ```typescript +import { AcpAgent, PrivyAlchemyEvmProviderAdapter } from "@virtuals-protocol/acp-node-v2"; +import type { + AgentMessage, + JobRoomEntry, + JobSession, +} from "@virtuals-protocol/acp-node-v2"; +import { base } from "@account-kit/infra"; + async function main() { const evaluator = await AcpAgent.create({ evmProvider: await PrivyAlchemyEvmProviderAdapter.create({ @@ -212,7 +220,8 @@ async function main() { // What was asked for, and what came back. const requirement = session.entries.find( - (e) => e.kind === "message" && e.contentType === "requirement" + (e): e is AgentMessage => + e.kind === "message" && e.contentType === "requirement" ); const deliverable = entry.event.deliverable; @@ -336,6 +345,21 @@ agent.on("entry", async (session, entry) => { }); ``` +`reason` on `job.completed` / `job.rejected` is typed `string`, but the value you +receive is the on-chain `bytes32` -- the string you passed to +`complete()`/`reject()`, hex-encoded and right-padded (`"rejected"` arrives as +`0x72656a6563746564...0000`). The SDK does not decode it. Decode it yourself if +you display it, and note the 32-byte limit truncates longer reasons: + +```typescript +import { hexToString } from "viem"; + +if (entry.kind === "system" && entry.event.type === "job.rejected") { + const reason = hexToString(entry.event.reason as `0x${string}`, { size: 32 }); + console.log(`rejected: ${reason}`); +} +``` + ### Restart & replay semantics **`agent.start()` replays events, and your `entry` handler must be idempotent.** @@ -355,13 +379,20 @@ is a failure path, not a guard: it costs gas, it surfaces as an error you now have to classify as benign, and any hook or fee transfer reached before the revert still ran. -The SDK does not dedupe for you, and it deliberately can't do it well: -`JobRoomEntry` carries no stable id, and the delivery you need to suppress -happens *across* process boundaries -- an in-memory `Set` is wiped by exactly the -restart that causes the replay. So dedup belongs in your own persistent store: +Within a single process the SDK does dedupe: entries are tracked by content key +(`entryKey`, exported if you want it), so the same entry never reaches your +handler twice in one run -- not on a stream reconnect, and not when a live entry +lands while `start()` is still hydrating. + +What it cannot do is dedupe *across* process boundaries. The key set lives in +memory and dies with the process, which is exactly the restart that triggers the +replay -- and "have I seen this entry" is a different question from "have I +already ruled on this job" anyway. So cross-restart dedup belongs in your own +persistent store: ```typescript // Any durable store works -- SQLite, Redis, a JSON file. +if (entry.kind !== "system") return; // narrows `entry.event` const key = `${session.chainId}-${session.jobId}-${entry.event.type}`; if (await store.has(key)) return; @@ -416,14 +447,21 @@ const agents = await agent.browseAgents("meme seller", { showHidden: true, }); +// browseAgents can come back empty, and a registered agent can have no +// offerings -- guard before indexing (required under `noUncheckedIndexedAccess`, +// and a real runtime case either way). +const seller = agents[0]; +if (!seller) throw new Error("no agent matched the query"); + // Each agent has offerings with typed requirements -const offering = agents[0].offerings[0]; +const offering = seller.offerings[0]; +if (!offering) throw new Error("agent has no offerings"); // Create job by offering name (simplest approach) const jobId = await agent.createJobByOfferingName( base.id, offering.name, - agents[0].walletAddress, + seller.walletAddress, { ticker: "PEPE", amount: 100 }, // requirement data validated against offering schema { evaluatorAddress: await agent.getAddress() } ); From 975145f438beaca593dd141dfb6afff5725a53d0 Mon Sep 17 00:00:00 2001 From: Joey Lau Date: Wed, 19 Aug 2026 15:38:52 +0800 Subject: [PATCH 4/4] fix: don't treat transcript membership as handler delivery `dispatch` gated the handler on `appendEntry`'s return value: const entryIsNew = session.appendEntry(entry); if (!entryIsNew && !sessionIsNew) return; ... await session.fetchJob(); this.fireHandler(session, entry); The entry was recorded in the transcript before `fetchJob()` and `fireHandler()` ran, so a transient `getJob()` failure -- routine while the observer lags behind chain state -- left it permanently marked "seen". Every later arrival of the same logical entry hit `entryIsNew === false` and returned: the live entry drained after a failed hydrate, a reconnect replay, any of it. The in-process exactly-once path delivered zero times, silently. `hydrateSessions()` compounded it. Its `await session.fetchJob()` had no try/catch, so one lagging job threw out of the whole loop and every job after it in `getActiveJobs()` order lost its hydration replay too. Being in `entries` is a fact about history, not about the handler. The two are now tracked separately: - `JobSession` gains `deliveredEntryKeys`, `tryClaimDelivery()` and `unclaimDelivery()`. The claim is synchronous, so two paths racing the same entry can't both win it. - `dispatch()` and `hydrateSessions()` claim before the fetch and release on failure, leaving a later dispatch or replay free to retry. Hydration's failure path `continue`s to the next job instead of aborting the loop. - The `job.created` role-swap branch hands the claim from the superseded session to its replacement. - `appendEntry()` stays idempotent but no longer carries delivery meaning. Verified against the parent commit by running the two new cases on its source in a detached worktree: 7/9, with the live-dispatch case asserting 0 !== 1 (handler fired zero times) and the hydration case throwing out of `hydrateSessions`. The 7 pre-existing cases pass before and after, so the change is narrow. 9/9 here, `typecheck` and `typecheck:tests` clean. One narrower hole is left open deliberately. Live dispatch is fire-and-forget, so if two copies of an entry arrive concurrently *and* the first one's `fetchJob()` fails, the second returns on the still-held claim before the first releases it -- zero deliveries, no pending replay. Closing it needs an in-flight-promise map keyed by entry to serialize duplicate dispatches, which is a separate change; both triggers fixed here are sequential. Co-Authored-By: Claude Opus 5 (1M context) --- src/acpAgent.ts | 101 ++++++++++++++++++++++-------------- src/jobSession.ts | 22 +++++++- tests/entryDelivery.test.ts | 43 ++++++++++++++- 3 files changed, 123 insertions(+), 43 deletions(-) diff --git a/src/acpAgent.ts b/src/acpAgent.ts index 04ab910..027efe0 100644 --- a/src/acpAgent.ts +++ b/src/acpAgent.ts @@ -379,9 +379,10 @@ export class AcpAgent { * is not a deduplication mechanism, and hooks or fee transfers reached before * the revert are not free. * - * Within one process each entry is delivered once — sessions track entries by - * content key (`entryKey`), so a stream reconnect or an entry landing - * mid-hydration can't fire the handler twice. Across restarts it can, and + * Within one process each entry is delivered once — sessions track handler + * delivery separately from the transcript (`tryClaimDelivery`), so a + * transient `getJob` failure during hydration or dispatch can retry without + * being mistaken for an already-handled entry. Across restarts it can, and * deliberately so: that replay is what lets an agent killed mid-flow pick the * job back up. Since delivery survives restarts, so must your record of what * you've acted on — persist a key like `${chainId}-${jobId}-${event.type}` @@ -427,9 +428,9 @@ export class AcpAgent { /** * Dispatch entries that arrived while hydration was running. * - * Anything hydration already replayed is dropped by `dispatch` itself, which - * ignores an entry the session has seen — so an entry that landed in the - * window reaches the handler exactly once, not twice and not never. + * Anything hydration already delivered is dropped by `dispatch` itself via + * `tryClaimDelivery` — so an entry that landed in the window reaches the + * handler exactly once, not twice and not never. */ private async drainPendingEntries(): Promise { const queued = this.pendingEntries; @@ -473,8 +474,20 @@ export class AcpAgent { job.chainId, entries, ); - await session.fetchJob(); - this.fireHandler(session, entries[entries.length - 1]!); + const latest = entries[entries.length - 1]!; + if (!session.tryClaimDelivery(latest)) continue; + try { + await session.fetchJob(); + } catch (err) { + session.unclaimDelivery(latest); + console.error( + `Failed to fetch job ${job.onChainJobId} on chain ${job.chainId} during hydration; ` + + `will retry on the next dispatch`, + err, + ); + continue; + } + this.fireHandler(session, latest); } } @@ -573,9 +586,11 @@ export class AcpAgent { const jobId = entry.onChainJobId; const chainId = entry.chainId; - let session = this.getSession(chainId, jobId); - let sessionIsNew = false; - if (!session) { + const existingSession = this.getSession(chainId, jobId); + let activeSession: JobSession; + if (existingSession) { + activeSession = existingSession; + } else { // `job.created` carries the client/provider/evaluator addresses, so it can // stand up a session on its own. Any other first-sighting cannot: roles // come from the creation event, and without it `inferRoles` falls back to @@ -597,40 +612,46 @@ export class AcpAgent { ); } } - session = this.getOrCreateSession(jobId, chainId, history); - sessionIsNew = true; + activeSession = this.getOrCreateSession(jobId, chainId, history); } - // A known session already holding this entry means another path (hydration, - // or a reconnect replay) delivered it — don't fire twice. On a session we - // just built, the entry being present only means our own history fetch - // included it, and nothing has fired for it yet. - const entryIsNew = session.appendEntry(entry); - if (!entryIsNew && !sessionIsNew) return; - - if (entry.kind === "system" && entry.event.type === "job.created") { - const roles = this.inferRoles([entry]); - const rolesChanged = - roles.length !== session.roles.length || - roles.some((r, i) => r !== session.roles[i]); - if (rolesChanged) { - const newSession = new JobSession( - this, - [...this.addresses.values()], - jobId, - chainId, - roles, - session.entries, - ); - this.sessionMap.set(this.getSessionKey(chainId, jobId), newSession); - await newSession.fetchJob(); - this.fireHandler(newSession, entry); - return; + activeSession.appendEntry(entry); + if (!activeSession.tryClaimDelivery(entry)) return; + + try { + if (entry.kind === "system" && entry.event.type === "job.created") { + const roles = this.inferRoles([entry]); + const rolesChanged = + roles.length !== activeSession.roles.length || + roles.some((r, i) => r !== activeSession.roles[i]); + if (rolesChanged) { + const prevSession = activeSession; + const newSession = new JobSession( + this, + [...this.addresses.values()], + jobId, + chainId, + roles, + activeSession.entries, + ); + this.sessionMap.set(this.getSessionKey(chainId, jobId), newSession); + prevSession.unclaimDelivery(entry); + if (!newSession.tryClaimDelivery(entry)) return; + activeSession = newSession; + } } + + await activeSession.fetchJob(); + } catch (err) { + activeSession.unclaimDelivery(entry); + console.error( + `Failed to fetch job ${jobId} on chain ${chainId}; will retry on the next dispatch`, + err, + ); + return; } - await session.fetchJob(); - this.fireHandler(session, entry); + this.fireHandler(activeSession, entry); } private fireHandler(session: JobSession, entry: JobRoomEntry): void { diff --git a/src/jobSession.ts b/src/jobSession.ts index c31cb7f..68c32d5 100644 --- a/src/jobSession.ts +++ b/src/jobSession.ts @@ -179,6 +179,8 @@ export class JobSession { private readonly agentAddresses: Set; /** Keys of everything in `entries`, so membership is content- not reference-based. */ private readonly entryKeys = new Set(); + /** Keys whose handler delivery completed (distinct from merely being in the transcript). */ + private readonly deliveredEntryKeys = new Set(); constructor( agent: AcpAgent, @@ -224,13 +226,29 @@ export class JobSession { return this.entryKeys.has(entryKey(entry)); } + /** + * Reserve delivery for `entry`. Returns false if it was already delivered or + * claimed by a concurrent path. Call {@link unclaimDelivery} when + * `fetchJob()` fails so a later dispatch or hydration replay can retry. + */ + tryClaimDelivery(entry: JobRoomEntry): boolean { + const key = entryKey(entry); + if (this.deliveredEntryKeys.has(key)) return false; + this.deliveredEntryKeys.add(key); + return true; + } + + /** Release a delivery claim after a failed pre-handler step. */ + unclaimDelivery(entry: JobRoomEntry): void { + this.deliveredEntryKeys.delete(entryKey(entry)); + } + /** * Append an entry, ignoring one already present. * * Idempotent on purpose: hydration and the live stream can both produce the * same entry as separate objects, and appending it twice would duplicate the - * transcript. Returns whether the entry was new — callers use that to decide - * whether the handler still owes a delivery for it. + * transcript. Handler delivery is tracked separately via {@link tryClaimDelivery}. */ appendEntry(entry: JobRoomEntry): boolean { const key = entryKey(entry); diff --git a/tests/entryDelivery.test.ts b/tests/entryDelivery.test.ts index 2cb5014..36a5dd7 100644 --- a/tests/entryDelivery.test.ts +++ b/tests/entryDelivery.test.ts @@ -77,6 +77,8 @@ type FakeOpts = { /** Jobs `getActiveJobs()` reports; empty means nothing to hydrate. */ activeJobs?: { chainId: number; onChainJobId: string }[]; jobStatus?: AcpJobStatus; + /** Fail the first N `getJob()` calls (simulates observer lag). */ + getJobFailCount?: number; }; class FakeTransport implements AcpChatTransport { @@ -107,11 +109,19 @@ class FakeTransport implements AcpChatTransport { } class FakeApi implements AcpJobApi { + getJobCalls = 0; constructor(private readonly opts: FakeOpts) {} async getActiveJobs() { return this.opts.activeJobs ?? [{ chainId: CHAIN, onChainJobId: JOB }]; } async getJob(): Promise { + this.getJobCalls++; + if ( + this.opts.getJobFailCount !== undefined && + this.getJobCalls <= this.opts.getJobFailCount + ) { + throw new Error("observer lag"); + } return { chainId: CHAIN, onChainJobId: JOB, @@ -147,7 +157,8 @@ type Harness = { async function harness(myAddress: string, opts: FakeOpts = {}): Promise { const transport = new FakeTransport(opts); - const agent = new AcpAgent(new Map(), transport, new FakeApi(opts)); + const api = new FakeApi(opts); + const agent = new AcpAgent(new Map(), transport, api); // buildTransportContext() normally fills this from the provider adapters. (agent as unknown as { addresses: Map }).addresses.set( "evm", @@ -292,6 +303,36 @@ const tests: Array<[string, () => Promise]> = [ ); }, ], + [ + "a transient getJob failure during hydration retries on drain, not zero times", + async () => { + const h = await harness(PROVIDER, { + duringHydration: () => [fundedEntry()], + history: () => [createdEntry(), fundedEntry()], + getJobFailCount: 1, + }); + assert.equal( + h.firesOf("job.funded"), + 1, + "hydration failure must not permanently skip the handler", + ); + }, + ], + [ + "a transient getJob failure on live dispatch retries on reconnect replay", + async () => { + const h = await harness(PROVIDER, { + activeJobs: [], + history: () => [createdEntry(), fundedEntry()], + getJobFailCount: 1, + }); + h.transport.emit(fundedEntry()); + await tick(); + h.transport.emit(fundedEntry()); // reconnect replay after fetchJob recovers + await tick(60); + assert.equal(h.firesOf("job.funded"), 1); + }, + ], ]; let failed = 0;