From e4e8fc2a867e2b1865c293f2a425b4a9cb63fc14 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Sat, 29 Aug 2026 19:35:30 -0700 Subject: [PATCH 1/2] =?UTF-8?q?fix(runtime):=20stage=20the=20reply=20befor?= =?UTF-8?q?e=20posting=20=E2=80=94=20redelivery=20never=20re-runs=20the=20?= =?UTF-8?q?model=20(#1344)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit resolveStagedReply writes the reply to DO storage under the event id before the post; a redelivery of the same event reuses it (model not called); commitStagedReply clears it after a successful post; a model failure stages nothing so the retry re-runs rather than going silent. 4 tests; 24/24. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_013pc6nGXRS8mHvrwcXMSRDK --- workers/agent-runtime/src/agent-do.ts | 18 +++++++-- workers/agent-runtime/src/staging.ts | 28 +++++++++++++ workers/agent-runtime/test/staging.test.ts | 46 ++++++++++++++++++++++ 3 files changed, 88 insertions(+), 4 deletions(-) create mode 100644 workers/agent-runtime/src/staging.ts create mode 100644 workers/agent-runtime/test/staging.test.ts diff --git a/workers/agent-runtime/src/agent-do.ts b/workers/agent-runtime/src/agent-do.ts index fe5ae3b388..bea08f47e9 100644 --- a/workers/agent-runtime/src/agent-do.ts +++ b/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,27 @@ 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 (and capped so a stuck post cannot grow storage). private async handleEvent(cfg: CapConfig, event: CapEvent): Promise { 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 diff --git a/workers/agent-runtime/src/staging.ts b/workers/agent-runtime/src/staging.ts new file mode 100644 index 0000000000..16e3ff9ee3 --- /dev/null +++ b/workers/agent-runtime/src/staging.ts @@ -0,0 +1,28 @@ +// 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(key: string): Promise; + put(key: string, value: unknown): Promise; + delete(key: string): Promise; +} + +export const stagedKey = (eventId: string): string => `staged:${eventId}`; + +export const resolveStagedReply = async ( + storage: StagingStorage, + eventId: string, + run: () => Promise, +): Promise<{ reply: string; fromStage: boolean }> => { + const existing = await storage.get(stagedKey(eventId)); + if (existing !== undefined) return { reply: existing, fromStage: true }; + const reply = await run(); + await storage.put(stagedKey(eventId), reply); + return { reply, fromStage: false }; +}; + +export const commitStagedReply = async (storage: StagingStorage, eventId: string): Promise => { + await storage.delete(stagedKey(eventId)); +}; diff --git a/workers/agent-runtime/test/staging.test.ts b/workers/agent-runtime/test/staging.test.ts new file mode 100644 index 0000000000..cc2ebfe902 --- /dev/null +++ b/workers/agent-runtime/test/staging.test.ts @@ -0,0 +1,46 @@ +import { describe, it, expect } from 'vitest'; +import { resolveStagedReply, commitStagedReply, stagedKey } from '../src/staging'; + +const mem = () => { + const m = new Map(); + return { + get: async (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), + 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'))).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('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); + }); +}); From 54775a5251278766da8acf0c899e80c1282d9734 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Sat, 29 Aug 2026 21:29:19 -0700 Subject: [PATCH 2/2] fix(runtime): staged replies carry a timestamp and orphans are pruned after 24h (Otto's #1346 hold) The comment claimed a cap that did not exist. Now: entries are {reply, at}; each stage prunes entries past STAGE_TTL_MS via a prefix list, so a stuck post cannot grow storage. Test pins keep-within-TTL, prune-past-TTL, and the prune count. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_013pc6nGXRS8mHvrwcXMSRDK --- workers/agent-runtime/src/agent-do.ts | 3 ++- workers/agent-runtime/src/staging.ts | 30 +++++++++++++++++++--- workers/agent-runtime/test/staging.test.ts | 19 ++++++++++++-- 3 files changed, 45 insertions(+), 7 deletions(-) diff --git a/workers/agent-runtime/src/agent-do.ts b/workers/agent-runtime/src/agent-do.ts index bea08f47e9..24d07059b4 100644 --- a/workers/agent-runtime/src/agent-do.ts +++ b/workers/agent-runtime/src/agent-do.ts @@ -129,7 +129,8 @@ export class AgentRuntimeDO implements DurableObject { // 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 (and capped so a stuck post cannot grow storage). + // 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 { const podId = event.podId || event.payload?.podId; if (!podId) return; diff --git a/workers/agent-runtime/src/staging.ts b/workers/agent-runtime/src/staging.ts index 16e3ff9ee3..19f65e45b2 100644 --- a/workers/agent-runtime/src/staging.ts +++ b/workers/agent-runtime/src/staging.ts @@ -7,22 +7,44 @@ export interface StagingStorage { get(key: string): Promise; put(key: string, value: unknown): Promise; delete(key: string): Promise; + list(options: { prefix: string }): Promise>; } -export const stagedKey = (eventId: string): string => `staged:${eventId}`; +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, + now: number = Date.now(), ): Promise<{ reply: string; fromStage: boolean }> => { - const existing = await storage.get(stagedKey(eventId)); - if (existing !== undefined) return { reply: existing, fromStage: true }; + const existing = await storage.get(stagedKey(eventId)); + if (existing !== undefined) return { reply: existing.reply, fromStage: true }; const reply = await run(); - await storage.put(stagedKey(eventId), reply); + 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 => { + const all = await storage.list({ 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 => { await storage.delete(stagedKey(eventId)); }; diff --git a/workers/agent-runtime/test/staging.test.ts b/workers/agent-runtime/test/staging.test.ts index cc2ebfe902..6addcfffbf 100644 --- a/workers/agent-runtime/test/staging.test.ts +++ b/workers/agent-runtime/test/staging.test.ts @@ -1,5 +1,5 @@ import { describe, it, expect } from 'vitest'; -import { resolveStagedReply, commitStagedReply, stagedKey } from '../src/staging'; +import { resolveStagedReply, commitStagedReply, stagedKey, pruneStaged, STAGE_TTL_MS } from '../src/staging'; const mem = () => { const m = new Map(); @@ -7,6 +7,7 @@ const mem = () => { get: async (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 ({ prefix }: { prefix: string }) => new Map([...m].filter(([k]) => k.startsWith(prefix))) as Map, raw: m, }; }; @@ -18,7 +19,7 @@ describe('reply staging (#1344 — no model re-run on redelivery)', () => { 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'))).toBe('hello'); + 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 () => { @@ -38,6 +39,20 @@ describe('reply staging (#1344 — no model re-run on redelivery)', () => { 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');