From e3116602ee382553a9623fcad615dd92d677b69a Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Sat, 29 Aug 2026 19:42:27 -0700 Subject: [PATCH 1/5] feat(runtime): present the delivery nonce on ack; 409 stale_delivery stops the turn (ADR-026 D6 consumer) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pairs with #1347. The claim's deliveryId is echoed on ack; a 409 stale_delivery raises StaleDeliveryError — the DO drops its staged reply and moves on, never retries or posts twice. Additive: today's server ignores the body and never 409s. 2 tests; 22/22. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_013pc6nGXRS8mHvrwcXMSRDK --- workers/agent-runtime/src/agent-do.ts | 11 +++++++++-- workers/agent-runtime/src/cap.ts | 15 ++++++++++++++- workers/agent-runtime/test/cap.test.ts | 14 +++++++++++++- 3 files changed, 36 insertions(+), 4 deletions(-) diff --git a/workers/agent-runtime/src/agent-do.ts b/workers/agent-runtime/src/agent-do.ts index 24d07059b..4b4066646 100644 --- a/workers/agent-runtime/src/agent-do.ts +++ b/workers/agent-runtime/src/agent-do.ts @@ -9,7 +9,7 @@ // 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'; @@ -105,8 +105,15 @@ export class AgentRuntimeDO implements DurableObject { processed.push(event._id); await this.state.storage.put('processedEventIds', processed.slice(-200)); } - await ackEvent(cfg, event._id); + await ackEvent(cfg, event._id, event.deliveryId); } 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; drop + // the staged reply so nothing posts twice. + await this.state.storage.delete(`staged:${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..b521f315a 100644 --- a/workers/agent-runtime/src/cap.ts +++ b/workers/agent-runtime/src/cap.ts @@ -10,9 +10,19 @@ export interface CapEvent { _id: string; type: string; podId?: string; + // ADR-026 D6 (#1347): the claim's delivery nonce. Presented on ack; a + // 409 stale_delivery means this runtime was superseded for the event. + deliveryId?: string; 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 +39,14 @@ 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. + if (res.status === 409) throw new StaleDeliveryError(eventId); if (!res.ok) throw new Error(`ackEvent ${res.status}`); }; diff --git a/workers/agent-runtime/test/cap.test.ts b/workers/agent-runtime/test/cap.test.ts index 91a2ea8d5..9b4b684e0 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,18 @@ 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 stale_delivery is a StaleDeliveryError — stop, never retry', async () => { + fetchMock.mockResolvedValue({ ok: false, status: 409 }); + await expect(ackEvent(cfg, 'e1', 'old')).rejects.toBeInstanceOf(StaleDeliveryError); + }); + it('posts a message as JSON to the pod route', async () => { fetchMock.mockResolvedValue({ ok: true }); await postMessage(cfg, 'pod-1', 'hello'); From 34ddfe565f6b3468360c16c2e095d5bb90f6dae5 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Sat, 29 Aug 2026 19:44:24 -0700 Subject: [PATCH 2/5] fix(runtime): branch stale_delivery on the body code, not the bare 409; document that D6 makes acks single-winner, not posts (Otto) Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_013pc6nGXRS8mHvrwcXMSRDK --- workers/agent-runtime/src/cap.ts | 7 ++++++- workers/agent-runtime/test/cap.test.ts | 11 +++++++++-- 2 files changed, 15 insertions(+), 3 deletions(-) diff --git a/workers/agent-runtime/src/cap.ts b/workers/agent-runtime/src/cap.ts index b521f315a..3d45d1e72 100644 --- a/workers/agent-runtime/src/cap.ts +++ b/workers/agent-runtime/src/cap.ts @@ -46,7 +46,12 @@ export const ackEvent = async (cfg: CapConfig, eventId: string, deliveryId?: str body: JSON.stringify(deliveryId ? { deliveryId } : {}), }); // 'You were replaced' is not a retryable failure: stop, do not post again. - if (res.status === 409) throw new StaleDeliveryError(eventId); + // 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/test/cap.test.ts b/workers/agent-runtime/test/cap.test.ts index 9b4b684e0..daa7fdedc 100644 --- a/workers/agent-runtime/test/cap.test.ts +++ b/workers/agent-runtime/test/cap.test.ts @@ -37,11 +37,18 @@ describe('CAP client — the four verbs a BYO wrapper speaks', () => { expect(JSON.parse(init.body)).toEqual({ deliveryId: 'nonce-abc' }); }); - it('ack 409 stale_delivery is a StaleDeliveryError — stop, never retry', async () => { - fetchMock.mockResolvedValue({ ok: false, status: 409 }); + 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'); From c5bd69efb0c2ada4befa5bf81acaed6a8be09ca3 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Sat, 29 Aug 2026 19:44:53 -0700 Subject: [PATCH 3/5] docs(runtime): D6 makes acks single-winner, not posts (Otto) Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_013pc6nGXRS8mHvrwcXMSRDK --- workers/agent-runtime/README.md | 10 ++++++++++ 1 file changed, 10 insertions(+) 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()` From c6a7821574dd0936c29519ad5391b974037944a0 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Sat, 29 Aug 2026 21:36:10 -0700 Subject: [PATCH 4/5] =?UTF-8?q?fix(runtime):=20#1349=20on=20the=20merged?= =?UTF-8?q?=20staging=20=E2=80=94=20shared=20stagedKey,=20shape=20guard=20?= =?UTF-8?q?on=20staged=20entries=20(Otto)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_013pc6nGXRS8mHvrwcXMSRDK --- workers/agent-runtime/src/agent-do.ts | 4 ++-- workers/agent-runtime/src/staging.ts | 7 ++++++- workers/agent-runtime/test/staging.test.ts | 8 ++++++++ 3 files changed, 16 insertions(+), 3 deletions(-) diff --git a/workers/agent-runtime/src/agent-do.ts b/workers/agent-runtime/src/agent-do.ts index 4b4066646..1ccb7801c 100644 --- a/workers/agent-runtime/src/agent-do.ts +++ b/workers/agent-runtime/src/agent-do.ts @@ -12,7 +12,7 @@ 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; @@ -111,7 +111,7 @@ export class AgentRuntimeDO implements DurableObject { // Superseded: the kernel handed this event to another runtime // after our claim expired. Not an error to record as ours; drop // the staged reply so nothing posts twice. - await this.state.storage.delete(`staged:${event._id}`); + await this.state.storage.delete(stagedKey(event._id)); continue; } batchErrors += 1; 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/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'); From 91e641e67d9550112a44fa13bef2a66a17248ac7 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Sun, 30 Aug 2026 03:26:29 -0700 Subject: [PATCH 5/5] fix(runtime): read delivery nonce from claimed payload --- workers/agent-runtime/src/agent-do.ts | 12 +++-- workers/agent-runtime/src/cap.ts | 3 -- .../test/agent-do.delivery-nonce.test.ts | 53 +++++++++++++++++++ 3 files changed, 62 insertions(+), 6 deletions(-) create mode 100644 workers/agent-runtime/test/agent-do.delivery-nonce.test.ts diff --git a/workers/agent-runtime/src/agent-do.ts b/workers/agent-runtime/src/agent-do.ts index 1ccb7801c..0c32d957d 100644 --- a/workers/agent-runtime/src/agent-do.ts +++ b/workers/agent-runtime/src/agent-do.ts @@ -105,12 +105,18 @@ export class AgentRuntimeDO implements DurableObject { processed.push(event._id); await this.state.storage.put('processedEventIds', processed.slice(-200)); } - await ackEvent(cfg, event._id, event.deliveryId); + 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; drop - // the staged reply so nothing posts twice. + // 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; } diff --git a/workers/agent-runtime/src/cap.ts b/workers/agent-runtime/src/cap.ts index 3d45d1e72..f3027ab6b 100644 --- a/workers/agent-runtime/src/cap.ts +++ b/workers/agent-runtime/src/cap.ts @@ -10,9 +10,6 @@ export interface CapEvent { _id: string; type: string; podId?: string; - // ADR-026 D6 (#1347): the claim's delivery nonce. Presented on ack; a - // 409 stale_delivery means this runtime was superseded for the event. - deliveryId?: string; payload?: { content?: string; podId?: string; [k: string]: unknown }; } 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'); + }); +});