fix(runtime): stage the reply before posting — idempotent redelivery (#1344)

workers/agent-runtime/src/agent-do.ts

@@ -12,6 +12,7 @@
 import { listEvents, ackEvent, postMessage, CapConfig, CapEvent } from './cap';
 import { runTurn } from './turn';
 import { buildCapTools } from './tools';
+import { resolveStagedReply, commitStagedReply } from './staging';
 
 export interface Env {
   AGENT: DurableObjectNamespace;
@@ -124,18 +125,28 @@ export class AgentRuntimeDO implements DurableObject {
     }
   }
 
+  // #1344 (Otto): a postMessage failure after a successful turn must not
+  // re-run the model on redelivery. The reply is STAGED on DO storage keyed
+  // by event id before posting; a redelivery of the same event posts the
+  // staged reply and skips the model entirely. Staged entries are cleared
+  // once the post succeeds; orphans from a stuck post are pruned after
+  // STAGE_TTL_MS on the next stage (staging.ts), so storage cannot grow.
   private async handleEvent(cfg: CapConfig, event: CapEvent): Promise<void> {
     const podId = event.podId || event.payload?.podId;
     if (!podId) return;
     if (event.type !== 'chat.mention' && event.type !== 'first_contact') return;
     // No eager context fetch: the mention is the prompt and read_pod_context
-    // earns the read when the model needs it (Otto: two CAP reads per turn
-    // and the same content twice against the byte ceiling otherwise).
-    const reply = await this.runTurn(String(event.payload?.content || ''), buildCapTools(cfg, podId));
+    // earns the read when the model needs it.
+    const { reply } = await resolveStagedReply(
+      this.state.storage,
+      event._id,
+      () => this.runTurn(String(event.payload?.content || ''), buildCapTools(cfg, podId)),
+    );
     // NO_REPLY contract: total-match suppression, same as every runtime.
-    if (reply && reply.trim() !== 'NO_REPLY') {
+    if (reply.trim() !== 'NO_REPLY') {
       await postMessage(cfg, podId, reply);
     }
+    await commitStagedReply(this.state.storage, event._id);
   }
 
   // The pi-driven turn (turn.ts): transcript on DO storage, streamSimple

workers/agent-runtime/src/staging.ts

@@ -0,0 +1,50 @@
+// Reply staging (#1344): make "model turn → post" idempotent across kernel
+// redeliveries. The reply is written to DO storage under the event id
+// BEFORE the post; a redelivery of the same event reuses it and never
+// re-runs the model. `commit` clears it once the post succeeded. Pure
+// over an injected storage so the contract is unit-testable without workerd.
+export interface StagingStorage {
+  get<T = unknown>(key: string): Promise<T | undefined>;
+  put(key: string, value: unknown): Promise<void>;
+  delete(key: string): Promise<boolean | void>;
+  list<T = unknown>(options: { prefix: string }): Promise<Map<string, T>>;
+}
+
+export const STAGE_PREFIX = 'staged:';
+export const stagedKey = (eventId: string): string => `${STAGE_PREFIX}${eventId}`;
+// A stuck post leaves its staged entry behind (no commit). The kernel retires
+// an event after 3 redeliveries, so an orphan is never reused — prune anything
+// older than this on each stage so orphans cannot accumulate (Otto, #1346).
+export const STAGE_TTL_MS = 24 * 60 * 60 * 1000;
+
+interface StagedEntry { reply: string; at: number }
+
+export const resolveStagedReply = async (
+  storage: StagingStorage,
+  eventId: string,
+  run: () => Promise<string>,
+  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 };
+  const reply = await run();
+  await pruneStaged(storage, now);
+  await storage.put(stagedKey(eventId), { reply, at: now } satisfies StagedEntry);
+  return { reply, fromStage: false };
+};
+
+export const pruneStaged = async (storage: StagingStorage, now: number = Date.now()): Promise<number> => {
+  const all = await storage.list<StagedEntry>({ prefix: STAGE_PREFIX });
+  let pruned = 0;
+  for (const [key, entry] of all) {
+    if (!entry || typeof entry.at !== 'number' || now - entry.at > STAGE_TTL_MS) {
+      await storage.delete(key);
+      pruned += 1;
+    }
+  }
+  return pruned;
+};
+
+export const commitStagedReply = async (storage: StagingStorage, eventId: string): Promise<void> => {
+  await storage.delete(stagedKey(eventId));
+};

workers/agent-runtime/test/staging.test.ts

@@ -0,0 +1,61 @@
+import { describe, it, expect } from 'vitest';
+import { resolveStagedReply, commitStagedReply, stagedKey, pruneStaged, STAGE_TTL_MS } from '../src/staging';
+
+const mem = () => {
+  const m = new Map<string, unknown>();
+  return {
+    get: async <T,>(k: string) => m.get(k) as T | undefined,
+    put: async (k: string, v: unknown) => { m.set(k, v); },
+    delete: async (k: string) => m.delete(k),
+    list: async <T,>({ prefix }: { prefix: string }) => new Map([...m].filter(([k]) => k.startsWith(prefix))) as Map<string, T>,
+    raw: m,
+  };
+};
+
+describe('reply staging (#1344 — no model re-run on redelivery)', () => {
+  it('first delivery runs the model and stages the reply before the post', async () => {
+    const s = mem();
+    let runs = 0;
+    const r = await resolveStagedReply(s, 'e1', async () => { runs += 1; return 'hello'; });
+    expect(r).toEqual({ reply: 'hello', fromStage: false });
+    expect(runs).toBe(1);
+    expect((s.raw.get(stagedKey('e1')) as { reply: string }).reply).toBe('hello');
+  });
+
+  it('a redelivery after a failed post reuses the staged reply — the model is NOT called again', async () => {
+    const s = mem();
+    let runs = 0;
+    const run = async () => { runs += 1; return 'hello'; };
+    await resolveStagedReply(s, 'e1', run);          // post then fails → no commit
+    const again = await resolveStagedReply(s, 'e1', run);
+    expect(again).toEqual({ reply: 'hello', fromStage: true });
+    expect(runs).toBe(1);
+  });
+
+  it('commit clears the stage so a later, different event with the same id cannot replay it', async () => {
+    const s = mem();
+    await resolveStagedReply(s, 'e1', async () => 'hello');
+    await commitStagedReply(s, 'e1');
+    expect(s.raw.has(stagedKey('e1'))).toBe(false);
+  });
+
+  it('orphaned stages older than the TTL are pruned on the next stage — storage cannot grow (Otto)', async () => {
+    const s = mem();
+    const t0 = 1_000_000;
+    await resolveStagedReply(s, 'old', async () => 'stuck', t0);              // post never commits
+    await resolveStagedReply(s, 'fresh', async () => 'ok', t0 + 1000);
+    expect(s.raw.has(stagedKey('old'))).toBe(true);                             // within TTL: kept
+    const later = t0 + STAGE_TTL_MS + 5000;
+    await resolveStagedReply(s, 'newer', async () => 'ok2', later);
+    expect(s.raw.has(stagedKey('old'))).toBe(false);                            // past TTL: pruned
+    expect(s.raw.has(stagedKey('fresh'))).toBe(false);                          // also past TTL by then
+    expect(s.raw.has(stagedKey('newer'))).toBe(true);
+    expect(await pruneStaged(s, later + STAGE_TTL_MS + 1)).toBe(1);             // only 'newer' left to prune
+  });
+
+  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');
+    expect(s.raw.has(stagedKey('e2'))).toBe(false);
+  });
+});