feat(runtime): present the delivery nonce on ack; 409 stops the turn (D6 consumer)

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()`

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)}`);
         }

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<CapEvent[]> => {
   return Array.isArray(body) ? body : body.events || [];
 };
 
-export const ackEvent = async (cfg: CapConfig, eventId: string): Promise<void> => {
+export const ackEvent = async (cfg: CapConfig, eventId: string, deliveryId?: string): Promise<void> => {
   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}`);
 };
 

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<StagedEntry>(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);

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<string, unknown>) => {
+  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<string, unknown>, 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');
+  });
+});

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');

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');