diff --git a/README.md b/README.md index bded569..224a6f6 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,79 @@ 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 +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({ + 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 is AgentMessage => + 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 +263,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 +281,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 +345,79 @@ 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.** + +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. + +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; + +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. @@ -293,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() } ); @@ -318,6 +479,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 +576,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 +638,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 +700,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..3bf79f5 100644 --- a/package.json +++ b/package.json @@ -5,8 +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" }, @@ -29,6 +32,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..027efe0 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, @@ -360,6 +366,29 @@ 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. + * + * 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}` + * before the side effect. An in-memory `Set` is lost on exactly the restart + * that triggers the replay. + */ async start( onConnected?: () => void, streams: SupportedStreams[] = DEFAULT_STREAMS, @@ -369,13 +398,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 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; + this.pendingEntries = []; + for (const entry of queued) { + try { + await this.dispatch(entry); + } catch (err) { + console.error(err); + } + } } async stop(): Promise { @@ -383,6 +449,8 @@ export class AcpAgent { await this.transport.disconnect(); this.started = false; } + this.hydrating = false; + this.pendingEntries = []; this.sessionMap.clear(); } @@ -406,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); } } @@ -456,7 +536,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( @@ -500,35 +585,73 @@ 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); + 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 + // ["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, + ); + } + } + activeSession = this.getOrCreateSession(jobId, chainId, history); } - 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 { @@ -612,6 +735,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 +782,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 +803,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 +824,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/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/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/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..68c32d5 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,10 @@ 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(); + /** Keys whose handler delivery completed (distinct from merely being in the transcript). */ + private readonly deliveredEntryKeys = new Set(); constructor( agent: AcpAgent, @@ -190,7 +195,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 +221,61 @@ 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)); + } + + /** + * 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. Handler delivery is tracked separately via {@link tryClaimDelivery}. + */ + 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..36a5dd7 --- /dev/null +++ b/tests/entryDelivery.test.ts @@ -0,0 +1,352 @@ +/** + * 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; + /** Fail the first N `getJob()` calls (simulates observer lag). */ + getJobFailCount?: number; +}; + +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 { + 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, + 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 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", + 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], + ); + }, + ], + [ + "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; +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.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"] +} 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"] +}