diff --git a/workers/agent-runtime/src/agent-do.ts b/workers/agent-runtime/src/agent-do.ts index fe5ae3b388..24d07059b4 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,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 { 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..19f65e45b2 --- /dev/null +++ b/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(key: string): Promise; + put(key: string, value: unknown): Promise; + delete(key: string): Promise; + list(options: { prefix: string }): Promise>; +} + +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.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 => { + 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 new file mode 100644 index 0000000000..6addcfffbf --- /dev/null +++ b/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(); + 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), + list: async ({ prefix }: { prefix: string }) => new Map([...m].filter(([k]) => k.startsWith(prefix))) as Map, + 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); + }); +});