diff --git a/workers/agent-runtime/README.md b/workers/agent-runtime/README.md index aff0f0ba8..f05884bbb 100644 --- a/workers/agent-runtime/README.md +++ b/workers/agent-runtime/README.md @@ -17,6 +17,16 @@ the same four CAP verbs a BYO wrapper does, with alarm-based polling in v1. Operator/CI only (Cloudflare account is operator-private): `npx wrangler deploy` with `ANTHROPIC_API_KEY` set as a secret. +## Delivery semantics (ADR-026 D6) — read before reasoning about duplicates + +The DO **posts before it acks**. The kernel's delivery nonce (#1347) makes +acks single-winner — a superseded runtime gets 409 `stale_delivery` and +stops — but it does NOT make posts exactly-once: by the time a 409 arrives, +this runtime's reply may already be in the pod, and the winning runtime is +a different DO that cannot see our staged reply. Staging (#1346) dedupes +redeliveries to the SAME runtime only. Cross-runtime exactly-once needs a +kernel-side message dedupe key; nobody should read D6 as providing it. + ## Deliberate v1 boundaries - Turn = pi agent-core's `runAgentLoop` with pi-ai's `streamSimple` transport (Anthropic provider registered explicitly — `createModels()` diff --git a/workers/agent-runtime/src/agent-do.ts b/workers/agent-runtime/src/agent-do.ts index 24d07059b..0c32d957d 100644 --- a/workers/agent-runtime/src/agent-do.ts +++ b/workers/agent-runtime/src/agent-do.ts @@ -9,10 +9,10 @@ // Turn engine: pi agent-core with an injected fetch-based streamFn (the // spike's finding #2: transport is dependency-injected and workerd-clean). // v1 wires a minimal turn; harness/compaction integration follows. -import { listEvents, ackEvent, postMessage, CapConfig, CapEvent } from './cap'; +import { listEvents, ackEvent, postMessage, StaleDeliveryError, CapConfig, CapEvent } from './cap'; import { runTurn } from './turn'; import { buildCapTools } from './tools'; -import { resolveStagedReply, commitStagedReply } from './staging'; +import { resolveStagedReply, commitStagedReply, stagedKey } from './staging'; export interface Env { AGENT: DurableObjectNamespace; @@ -105,8 +105,21 @@ export class AgentRuntimeDO implements DurableObject { processed.push(event._id); await this.state.storage.put('processedEventIds', processed.slice(-200)); } - await ackEvent(cfg, event._id); + const deliveryId = event.payload?.deliveryId; + await ackEvent( + cfg, + event._id, + typeof deliveryId === 'string' && deliveryId ? deliveryId : undefined, + ); } catch (err) { + if (err instanceof StaleDeliveryError) { + // Superseded: the kernel handed this event to another runtime + // after our claim expired. Not an error to record as ours; stop + // this runtime's retry path. The processed-id ring is what keeps + // a later redelivery from re-running an already handled turn. + await this.state.storage.delete(stagedKey(event._id)); + continue; + } batchErrors += 1; await this.state.storage.put('lastError', `event ${event._id}: ${String((err as Error).message)}`); } diff --git a/workers/agent-runtime/src/cap.ts b/workers/agent-runtime/src/cap.ts index 03dc50643..f3027ab6b 100644 --- a/workers/agent-runtime/src/cap.ts +++ b/workers/agent-runtime/src/cap.ts @@ -13,6 +13,13 @@ export interface CapEvent { payload?: { content?: string; podId?: string; [k: string]: unknown }; } +export class StaleDeliveryError extends Error { + constructor(eventId: string) { + super(`stale_delivery: event ${eventId} was redelivered to another runtime`); + this.name = 'StaleDeliveryError'; + } +} + const headers = (cfg: CapConfig) => ({ Authorization: `Bearer ${cfg.runtimeToken}`, 'Content-Type': 'application/json', @@ -29,11 +36,19 @@ export const listEvents = async (cfg: CapConfig): Promise => { return Array.isArray(body) ? body : body.events || []; }; -export const ackEvent = async (cfg: CapConfig, eventId: string): Promise => { +export const ackEvent = async (cfg: CapConfig, eventId: string, deliveryId?: string): Promise => { const res = await fetch(`${cfg.apiUrl}/api/agents/runtime/events/${eventId}/ack`, { method: 'POST', headers: headers(cfg), + body: JSON.stringify(deliveryId ? { deliveryId } : {}), }); + // 'You were replaced' is not a retryable failure: stop, do not post again. + // Branch on the body's code, not the bare status (Otto): a generic 409 from + // anything else must stay a retryable error, not a silent drop. + if (res.status === 409) { + const body = (await res.json().catch(() => ({}))) as { code?: string }; + if (body.code === 'stale_delivery') throw new StaleDeliveryError(eventId); + } if (!res.ok) throw new Error(`ackEvent ${res.status}`); }; diff --git a/workers/agent-runtime/src/staging.ts b/workers/agent-runtime/src/staging.ts index 19f65e45b..996e47aaf 100644 --- a/workers/agent-runtime/src/staging.ts +++ b/workers/agent-runtime/src/staging.ts @@ -26,7 +26,12 @@ export const resolveStagedReply = async ( now: number = Date.now(), ): Promise<{ reply: string; fromStage: boolean }> => { const existing = await storage.get(stagedKey(eventId)); - if (existing !== undefined) return { reply: existing.reply, fromStage: true }; + // Shape guard (Otto): a malformed or legacy-shaped entry must not yield + // reply: undefined and TypeError downstream — treat it as absent, drop it. + if (existing !== undefined && typeof existing?.reply === 'string') { + return { reply: existing.reply, fromStage: true }; + } + if (existing !== undefined) await storage.delete(stagedKey(eventId)); const reply = await run(); await pruneStaged(storage, now); await storage.put(stagedKey(eventId), { reply, at: now } satisfies StagedEntry); diff --git a/workers/agent-runtime/test/agent-do.delivery-nonce.test.ts b/workers/agent-runtime/test/agent-do.delivery-nonce.test.ts new file mode 100644 index 000000000..93514e718 --- /dev/null +++ b/workers/agent-runtime/test/agent-do.delivery-nonce.test.ts @@ -0,0 +1,53 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +const cap = vi.hoisted(() => ({ + listEvents: vi.fn(), + ackEvent: vi.fn(), + postMessage: vi.fn(), +})); + +vi.mock('../src/cap', () => ({ + ...cap, + StaleDeliveryError: class StaleDeliveryError extends Error {}, +})); + +import { AgentRuntimeDO } from '../src/agent-do'; + +const cfg = { apiUrl: 'https://api.test', runtimeToken: 'cm_agent_x' }; + +const stateWith = (values: Record) => { + const data = new Map(Object.entries(values)); + const storage = { + get: vi.fn(async (key: string) => data.get(key)), + put: vi.fn(async (key: string | Record, value?: unknown) => { + if (typeof key === 'string') data.set(key, value); + else Object.entries(key).forEach(([entry, stored]) => data.set(entry, stored)); + }), + delete: vi.fn(async (key: string) => data.delete(key)), + setAlarm: vi.fn(async () => undefined), + }; + return { storage, state: { storage } as unknown as DurableObjectState }; +}; + +describe('AgentRuntimeDO D6 acknowledgement', () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it('echoes the nonce from the actual polled-event payload', async () => { + const { state } = stateWith({ runtimeToken: cfg.runtimeToken, pollSeconds: 5 }); + cap.listEvents.mockResolvedValue([ + { + _id: 'event-1', + type: 'unknown', + payload: { deliveryId: 'delivery-from-claim' }, + }, + ]); + cap.ackEvent.mockResolvedValue(undefined); + const runtime = new AgentRuntimeDO(state, { COMMONLY_API_URL: cfg.apiUrl } as never); + + await runtime.alarm(); + + expect(cap.ackEvent).toHaveBeenCalledWith(cfg, 'event-1', 'delivery-from-claim'); + }); +}); diff --git a/workers/agent-runtime/test/cap.test.ts b/workers/agent-runtime/test/cap.test.ts index 91a2ea8d5..daa7fdedc 100644 --- a/workers/agent-runtime/test/cap.test.ts +++ b/workers/agent-runtime/test/cap.test.ts @@ -1,5 +1,5 @@ import { describe, it, expect, vi, beforeEach } from 'vitest'; -import { listEvents, ackEvent, postMessage, getPodContext } from '../src/cap'; +import { listEvents, ackEvent, postMessage, getPodContext, StaleDeliveryError } from '../src/cap'; const cfg = { apiUrl: 'https://api.test', runtimeToken: 'cm_agent_x' }; @@ -30,6 +30,25 @@ describe('CAP client — the four verbs a BYO wrapper speaks', () => { await expect(getPodContext(cfg, 'p1')).rejects.toThrow('getPodContext 401'); }); + it('ack presents the delivery nonce when the claim carried one (ADR-026 D6)', async () => { + fetchMock.mockResolvedValue({ ok: true, status: 200 }); + await ackEvent(cfg, 'e1', 'nonce-abc'); + const [, init] = fetchMock.mock.calls[0]; + expect(JSON.parse(init.body)).toEqual({ deliveryId: 'nonce-abc' }); + }); + + it('ack 409 with code stale_delivery is a StaleDeliveryError — stop, never retry', async () => { + fetchMock.mockResolvedValue({ ok: false, status: 409, json: async () => ({ code: 'stale_delivery' }) }); + await expect(ackEvent(cfg, 'e1', 'old')).rejects.toBeInstanceOf(StaleDeliveryError); + }); + + it('a 409 WITHOUT the stale_delivery code stays a generic retryable error (Otto)', async () => { + fetchMock.mockResolvedValue({ ok: false, status: 409, json: async () => ({ code: 'something_else' }) }); + await expect(ackEvent(cfg, 'e1', 'old')).rejects.toThrow('ackEvent 409'); + fetchMock.mockResolvedValue({ ok: false, status: 409, json: async () => { throw new Error('no body'); } }); + await expect(ackEvent(cfg, 'e1', 'old')).rejects.toThrow('ackEvent 409'); + }); + it('posts a message as JSON to the pod route', async () => { fetchMock.mockResolvedValue({ ok: true }); await postMessage(cfg, 'pod-1', 'hello'); diff --git a/workers/agent-runtime/test/staging.test.ts b/workers/agent-runtime/test/staging.test.ts index 6addcfffb..7c088bdc9 100644 --- a/workers/agent-runtime/test/staging.test.ts +++ b/workers/agent-runtime/test/staging.test.ts @@ -53,6 +53,14 @@ describe('reply staging (#1344 — no model re-run on redelivery)', () => { expect(await pruneStaged(s, later + STAGE_TTL_MS + 1)).toBe(1); // only 'newer' left to prune }); + it('a malformed staged entry is treated as absent and dropped, never returned (Otto)', async () => { + const s = mem(); + s.raw.set(stagedKey('e9'), 'bare-string-legacy-shape'); + const r = await resolveStagedReply(s, 'e9', async () => 'fresh'); + expect(r).toEqual({ reply: 'fresh', fromStage: false }); + expect((s.raw.get(stagedKey('e9')) as { reply: string }).reply).toBe('fresh'); + }); + it('a model failure stages nothing, so the redelivery retries the model (not silence)', async () => { const s = mem(); await expect(resolveStagedReply(s, 'e2', async () => { throw new Error('529'); })).rejects.toThrow('529');