From 2a56987eee0299586738e5672c435b3e63634997 Mon Sep 17 00:00:00 2001 From: Noah Lindner Date: Thu, 23 Jul 2026 08:50:08 -0400 Subject: [PATCH 1/5] =?UTF-8?q?=C2=A75.5=20stale-reply=20withholding:=20a?= =?UTF-8?q?=20thread-follow=20reply=20yields=20to=20a=20room=20that=20move?= =?UTF-8?q?d=20mid-turn?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Live 2026-07-23: in the font/image incident thread, a thread-follow wake answered a question a human had already answered — her reply composed against a wake-start snapshot and landed a minute stale, then the next wake misread the human's answer as a correction and flip-flopped. SPEC §5.5 (and the §18.2 'Stale-reply withholding' row) prescribe the guard: in a wake with no direct address, replies buffer until turn end; if newer addressed messages arrived on the same conversation mid-turn, the reply is withheld and the immediately following wake carries it as an unsent draft to reconsider. A directly-addressed wake never buffers — the asker is owed the answer. - toolset: optional bufferReply seam; a buffered reply records no effect (the flush owns posted/withheld truth in the turn row) - turn: beforeRecord hook so the flush lands inside the turn's ledger record - service: buffer + flush (post / withhold + unsent-draft slot, in-memory like ear notes); failed attempts drop their buffer (same rule as clearCards) Co-Authored-By: Claude Fable 5 --- src/service.ts | 51 +++++++++++++++++++- src/turn-runner/toolset.ts | 13 ++++++ src/turn-runner/turn.ts | 5 ++ test/resident.test.ts | 96 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 164 insertions(+), 1 deletion(-) diff --git a/src/service.ts b/src/service.ts index ed64fa5..5c1fbb5 100644 --- a/src/service.ts +++ b/src/service.ts @@ -21,6 +21,7 @@ import { import { queryMemory, coreWithinBudget } from "./ledger/memory"; import { pendingMessages, messagesAfter, advanceCursor, type InboxMessage } from "./ledger/inbox"; import { openAttentionItem, closeAttentionItemsForThread, closeAttentionItem, reopenAttentionItem, openItems, earCursor, advanceEarCursor } from "./ledger/attention"; +import { recordThreadParticipation } from "./ledger/threads"; import { composeEarInstructions } from "./turn-runner/ear-soul"; import { checkpointWal } from "./ledger/db"; import { runExecution, type ExecutionOutcome } from "./turn-runner/execution-loop"; @@ -107,6 +108,9 @@ export class Service { private earRunning = new Set(); private earRerun = new Set(); private earNotes = new Map(); + // §5.5 withheld replies awaiting the next wake's reconsideration. In-memory like earNotes + // (and for the same reason): a crash loses a draft the model can simply re-derive — fail-open. + private unsentDrafts = new Map(); constructor(deps: ServiceDeps) { this.d = deps; @@ -545,6 +549,41 @@ export class Service { }); const effects: unknown[] = []; let failureCause = ""; + // §5.5 stale-reply withholding: nobody addressed this wake directly, so a reply races the + // room — the model composes against a snapshot while people keep talking (2026-07-23 live: + // she answered a question a human had already answered, a minute later). Replies buffer + // here until turn end; flushBuffered (below, run before the turn records) posts each one + // unless newer addressed messages landed on its conversation mid-turn — those are withheld + // into the next wake as unsent drafts. A directly-addressed wake never buffers: the asker + // is owed the answer even if the thread has moved. + const batchTail = pending.at(-1)!.rowid; + const buffered: { anchor: Anchor; text: string }[] = []; + const bufferReply = direct.length > 0 ? undefined : (a: Anchor, text: string) => void buffered.push({ anchor: a, text }); + const flushBuffered = async (turnStatus: TurnStatus): Promise => { + const toFlush = buffered.splice(0); // each retry attempt re-decides from scratch + if (turnStatus !== "succeeded") return; // a dead wake's half-sent words never post (same rule as clearCards) + const drafts: string[] = []; + for (const b of toFlush) { + const moved = messagesAfter(this.d.db, identityId, batchTail).some( + (m) => + m.kind === "addressed_message" && + (m.venueId ?? "") === b.anchor.venueId && + (b.anchor.threadRootId === null ? m.threadRootId === null : (m.threadRootId ?? m.ts) === b.anchor.threadRootId), + ); + if (moved) { + drafts.push(`- to <#${b.anchor.venueId}>${b.anchor.threadRootId ? ` thread=${b.anchor.threadRootId}` : ""}: ${b.text}`); + effects.push({ kind: "withheld", anchor: b.anchor, text: b.text }); + continue; + } + const streamedId = + b.anchor.venueId === anchorObj.venueId && b.anchor.threadRootId === anchorObj.threadRootId ? await stream.post(b.text) : null; + const result = streamedId ? { messageId: streamedId } : await this.postMessage(b.anchor, b.text); + recordThreadParticipation(this.d.db, this.d.clock, identityId, b.anchor.venueId, b.anchor.threadRootId ?? result.messageId); + closeAttentionItemsForThread(this.d.db, this.d.clock, identityId, b.anchor.venueId, b.anchor.threadRootId ?? null, "answered in thread"); + effects.push({ kind: "posted", anchor: b.anchor, text: b.text }); + } + if (drafts.length) this.unsentDrafts.set(identityId, [...(this.unsentDrafts.get(identityId) ?? []), ...drafts]); + }; // §14.2 gate: flipped when a reply or react lands on a directly addressed message — a // wake that answered someone before dying leaves nobody hanging, so no fallback. Every // flip must co-occur with a pushed effect (the same tool call records one): the retry @@ -586,6 +625,7 @@ export class Service { }, checklist: { messageId: null }, effects, + ...(bufferReply ? { bufferReply } : {}), }); this.refreshSoul(); // a fresh thread must open with current memory + standing instructions // The prompt is the messages, plus the two model-authored slots the ear design adds: her @@ -604,6 +644,14 @@ export class Service { : `- you reacted :${d.emoji}: to ts=${d.ts} in <#${d.venueId}>`, ); const didSection = didLines.length ? `\n\n[what you did recently]\n${didLines.join("\n")}` : ""; + // §5.5: a withheld reply surfaces to the immediately following wake — the model's own + // words, reconsidered by the model against the room as it now stands. Consumed like ear + // notes: once, by whichever wake comes next. + const heldDrafts = this.unsentDrafts.get(identityId) ?? []; + this.unsentDrafts.delete(identityId); + const draftSection = heldDrafts.length + ? `\n\n[drafted last wake but not sent — the conversation had moved on; decide fresh what (if anything) to say]\n${heldDrafts.join("\n")}` + : ""; const readSection = notes.length ? `\n\n[your first read of the room]\n${notes.map((n) => `- ${n}`).join("\n")}` : ""; const owedSection = owed.length ? `\n\n[still owed]\n${owed @@ -614,7 +662,7 @@ export class Service { }) .join("\n")}${owed.length > ATTENTION_PROMPT_CAP ? `\n(+${owed.length - ATTENTION_PROMPT_CAP} newer ones not shown — they surface as these settle)` : ""}` : ""; - const prompt = `${pending.map((m) => `${isDirectAddress(m) ? "[to you] " : ""}${inboxLine(m)}`).join("\n")}${didSection}${readSection}${owedSection}`; + const prompt = `${pending.map((m) => `${isDirectAddress(m) ? "[to you] " : ""}${inboxLine(m)}`).join("\n")}${didSection}${draftSection}${readSection}${owedSection}`; let status: TurnStatus = "failed"; // In-flight work finishes under the policy it started with (SPEC §16.2) — snapshot once. const turns = this.policy().turns; @@ -650,6 +698,7 @@ export class Service { tokensUsed: () => 0, spendAmount: () => 0, envelope: { timeoutMs: turns.interactiveTimeoutMs, tokenCeiling: turns.interactiveTokenCeiling }, + beforeRecord: flushBuffered, }); status = result.status; if (!failureCause && result.cause) failureCause = result.cause; diff --git a/src/turn-runner/toolset.ts b/src/turn-runner/toolset.ts index 33dee19..a506d54 100644 --- a/src/turn-runner/toolset.ts +++ b/src/turn-runner/toolset.ts @@ -47,6 +47,11 @@ export interface ToolsetContext { taskId?: string; // the task this execution_step turn belongs to nudgeAfterMs: number; postMessage: (anchor: Anchor, text: string) => Promise<{ messageId: string }>; + // SPEC §5.5 stale-reply withholding: set only when the turn's batch had no direct address. + // Replies then buffer with the caller until turn end, which posts each one or withholds it + // (newer addressed arrivals on its conversation) into the next wake as an unsent draft. The + // caller owns the posted/withheld effect records; replyTool records nothing for a buffered call. + bufferReply?: (anchor: Anchor, text: string) => void; // Edit an already-posted message (Slack chat.update). Enables the live checklist. Optional — a // surface without it just re-posts instead of editing in place. updateMessage?: (venueId: string, messageId: string, text: string) => Promise; @@ -278,6 +283,14 @@ function replyTool(ctx: ToolsetContext): DynamicTool { } } + // §5.5: nobody addressed this turn directly, so the reply waits for turn end — the room + // may still be talking while the model composes, and an answer to a moved-on conversation + // is the harness's to hold back, not the model's to re-litigate mid-turn. + if (ctx.bufferReply) { + ctx.bufferReply(anchor, a.text); + return { success: true, output: "queued — it posts when your turn ends, unless the conversation has moved by then (it would come back to you next time instead)" }; + } + const result = await ctx.postMessage(anchor, a.text); recordPostedThread(ctx, anchor, result.messageId); pushEffect(ctx, { kind: "posted", anchor, text: a.text }); diff --git a/src/turn-runner/turn.ts b/src/turn-runner/turn.ts index 5e4deb9..cac3762 100644 --- a/src/turn-runner/turn.ts +++ b/src/turn-runner/turn.ts @@ -32,6 +32,9 @@ export interface RunTurnParams { tokensUsed: () => number; spendAmount: () => number; envelope?: EnvelopeOpts; // interactive/ambient/distillation (SPEC §4.1.6) + // Runs after the model's turn settles and BEFORE the turn row records — the window where + // §5.5's buffered replies post or withhold, so their effects land in this turn's record. + beforeRecord?: (status: TurnStatus) => Promise; // execution_step's watchdog (SPEC §6.3): wall-clock with NO activity, not total turn time. // Requires session.msSinceLastActivity(); a stall is "killed and treated as a failed attempt." stallTimeoutMs?: number; @@ -113,6 +116,8 @@ export async function runTurn(params: RunTurnParams): Promise { status = (await done) === "failed" ? "failed" : "succeeded"; } + if (params.beforeRecord) await params.beforeRecord(status); + recordTurn(params.db, params.clock, { id: params.turnId, identityId: params.identityId, diff --git a/test/resident.test.ts b/test/resident.test.ts index 8350f63..36f1bf0 100644 --- a/test/resident.test.ts +++ b/test/resident.test.ts @@ -411,3 +411,99 @@ describe("resident delivery", () => { await service.stop(); }); }); + +// SPEC §5.5 stale-reply withholding (§18.2 row): the room can move while the model composes. +// A thread-follow turn's reply buffers until turn end; newer addressed arrivals on the same +// conversation withhold it, and the NEXT wake reconsiders it as an unsent draft. A +// directly-addressed turn's reply is never withheld. +describe("stale-reply withholding (§5.5)", () => { + // Each test's ear script wakes the mind for thread chatter — the ear's judgment isn't under + // test here, the wake's posting behavior is. + const earWakes = async (tools: Map): Promise => { + const verdict = tools.get("verdict"); + if (!verdict) return false; + await verdict.run({ decision: "wake", why: "her thread is moving", venueId: "C1", threadRootId: "1.0" }); + return true; + }; + + test("§5.5: a thread-follow reply is withheld when the conversation moved mid-turn; the next wake carries the unsent draft", async () => { + let mindWakes = 0; + let replyResult: { success: boolean; output: string } | undefined; + let emitMidTurn!: () => void; + const { db, adapter, service, minds } = harness(async (_turn, tools) => { + if (await earWakes(tools)) return; + if (++mindWakes === 2) { + // Noah answers Nina while she is still composing her own answer. + emitMidTurn(); + replyResult = (await tools.get("reply")!.run({ text: "the shipping window was clean", venueId: "C1", threadRootId: "1.0" })) as { + success: boolean; + output: string; + }; + } + }); + emitMidTurn = () => adapter.emit(msg({ text: "already answered: it shipped at 8pm", ts: "1.3", threadRootTs: "1.0", principalId: "U_NOAH" })); + await service.start(); + adapter.emit(msg({ text: "<@BOT1> keep an eye on this thread", mentionsBotId: true, ts: "1.0" })); + await service.idle(); + adapter.emit(msg({ text: "so when did this actually ship?", ts: "1.2", threadRootTs: "1.0", principalId: "U_NINA" })); + await service.idle(); + + // The reply call itself succeeds (the model is done deciding) but nothing lands in the room. + expect(replyResult!.success).toBe(true); + const everything = [...adapter.posts.map((p) => p.text), ...adapter.streams.map((s) => s.text)].join(" "); + expect(everything).not.toContain("the shipping window was clean"); + // The ledger records the withhold honestly — never a "posted" that didn't post. + const rows = db.query("SELECT effects FROM turns WHERE kind='resident'").all() as { effects: string }[]; + expect(rows.some((r) => r.effects.includes('"kind":"withheld"'))).toBe(true); + expect(rows.some((r) => r.effects.includes('"kind":"posted"') && r.effects.includes("shipping window was clean"))).toBe(false); + // The immediately following wake carries both the mover and the unsent draft. + expect(mindWakes).toBeGreaterThanOrEqual(3); + const next = minds()[2]!.prompts[0]!; + expect(next).toContain("already answered: it shipped at 8pm"); + expect(next).toContain("[drafted last wake but not sent"); + expect(next).toContain("the shipping window was clean"); + await service.stop(); + }); + + test("§5.5: a thread-follow reply with no mid-turn arrivals posts normally at turn end", async () => { + let mindWakes = 0; + const { db, adapter, service } = harness(async (_turn, tools) => { + if (await earWakes(tools)) return; + if (++mindWakes === 2) { + await tools.get("reply")!.run({ text: "covered upthread — the fix shipped", venueId: "C1", threadRootId: "1.0" }); + } + }); + await service.start(); + adapter.emit(msg({ text: "<@BOT1> watch this one", mentionsBotId: true, ts: "1.0" })); + await service.idle(); + adapter.emit(msg({ text: "any update?", ts: "1.2", threadRootTs: "1.0", principalId: "U_NINA" })); + await service.idle(); + + expect(adapter.lastStreamText()).toBe("covered upthread — the fix shipped"); + const rows = db.query("SELECT effects FROM turns WHERE kind='resident'").all() as { effects: string }[]; + expect(rows.some((r) => r.effects.includes('"kind":"posted"') && r.effects.includes("covered upthread"))).toBe(true); + expect(rows.some((r) => r.effects.includes('"kind":"withheld"'))).toBe(false); + await service.stop(); + }); + + test("§5.5: a directly-addressed turn's reply is never withheld, even when the thread moves mid-turn", async () => { + let emitMidTurn!: () => void; + const { db, adapter, service } = harness(async (_turn, tools) => { + if (await earWakes(tools)) return; + if (adapter.streams.length === 0 && adapter.posts.length === 0) { + emitMidTurn(); + await tools.get("reply")!.run({ text: "answering you directly", venueId: "C1", threadRootId: "1.0" }); + } + }); + emitMidTurn = () => adapter.emit(msg({ text: "meanwhile the thread moves on", ts: "1.1", threadRootTs: "1.0", principalId: "U_NOAH" })); + await service.start(); + adapter.emit(msg({ text: "<@BOT1> when did this ship?", mentionsBotId: true, ts: "1.0" })); + await service.idle(); + + const everything = [...adapter.posts.map((p) => p.text), ...adapter.streams.map((s) => s.text)].join(" "); + expect(everything).toContain("answering you directly"); + const rows = db.query("SELECT effects FROM turns WHERE kind='resident'").all() as { effects: string }[]; + expect(rows.some((r) => r.effects.includes('"kind":"withheld"'))).toBe(false); + await service.stop(); + }); +}); From e0d39b6327b3c3b365bca3642fc580e52b83ebb1 Mon Sep 17 00:00:00 2001 From: Noah Lindner Date: Thu, 23 Jul 2026 10:11:02 -0400 Subject: [PATCH 2/5] =?UTF-8?q?ear:=20needing=20someone=20is=20not=20needi?= =?UTF-8?q?ng=20her=20=E2=80=94=20the=20wake=20and=20debt=20boundary=20is?= =?UTF-8?q?=20ownership?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Live 2026-07-21..23: 26-44 wake verdicts/day, many for conversations the ear itself judged to belong to someone else ('noah owns the incident response and should investigate', 'kate still owes that check', 'someone should investigate', 'no owner is named yet' — all woke the mind). Third-party debts also landed in attention items and rode the wake prompt as hers, priming her to step into other people's threads. Result: replies into ~40 distinct threads/day. One boundary, stated once in the ear soul: when people are talking to each other, the conversation is theirs; a question aimed at another teammate is that person's to answer even when she knows it; record only debts aimed at her. Verdict tool description tightened to match (wake = hers AND now; open_ask = never what one teammate owes another). Unowned machine signals are untouched: alert front-running stays licensed by her memory facts, which the ear reads. Co-Authored-By: Claude Fable 5 --- src/service.ts | 2 +- src/turn-runner/ear-soul.ts | 6 ++++++ 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/src/service.ts b/src/service.ts index 5c1fbb5..f24b63b 100644 --- a/src/service.ts +++ b/src/service.ts @@ -404,7 +404,7 @@ export class Service { spec: { name: "verdict", description: - "Report one judgment about one conversation. decision: 'hold' (nothing needed from her), 'wake' (this needs her now — why becomes her own first read of it), 'open_ask' (a direct ask of her with no answer yet — record the debt; does not wake by itself), 'close_ask' / 'reopen_ask' (a recorded debt was settled / was not actually settled; pass itemId). Every why must read naturally if said aloud in the room.", + "Report one judgment about one conversation. decision: 'hold' (nothing needed from her), 'wake' (this is HERS and needs her now — why becomes her own first read of it), 'open_ask' (a direct ask of her, never what one teammate owes another — record the debt; does not wake by itself), 'close_ask' / 'reopen_ask' (a recorded debt was settled / was not actually settled; pass itemId). Every why must read naturally if said aloud in the room.", inputSchema: { type: "object", additionalProperties: false, diff --git a/src/turn-runner/ear-soul.ts b/src/turn-runner/ear-soul.ts index 805564c..52d663f 100644 --- a/src/turn-runner/ear-soul.ts +++ b/src/turn-runner/ear-soul.ts @@ -28,6 +28,12 @@ You report through the verdict tool, one verdict per conversation, and nothing e line as if she may say it aloud in the room, because she may: plain words about who is talking to whom and what is needed, never anything about tools, models, passes, or systems. +Needing someone is not needing her. When people are talking to each other, the conversation is +theirs: a question aimed at another teammate is that person's to answer even when she knows the +answer, and waking her into it costs the room more than it gives. The same boundary holds for +debts: record only asks aimed at her. What one teammate owes another is theirs, not hers to +carry or to chase. + Bias to hold. Most of what you hear needs nothing from her, and waking her for it costs the room more than it gives. But a real ask with no answer is the one failure you exist to prevent: when in doubt about an explicit request aimed at her, record the debt.`; From 2481ae3afa158ee1c54c3be846ee883fdd10fda1 Mon Sep 17 00:00:00 2001 From: Noah Lindner Date: Thu, 23 Jul 2026 10:24:46 -0400 Subject: [PATCH 3/5] ear: open asks are not hers to claim; debts must be anchored; operator closes are final MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Live 2026-07-22/23, the 'stfu youre not helping' incident: Noah posted open team asks ('Needs QA:', a GTM voice-memo request) with no addressee. The ear recorded them as HER debts ('noah is asking you to QA BEV-4611', 'the gtm team, including you') — one opened seconds after Noah told her 'not for u' in the same channel. Both were recorded with thread_root_id NULL, so her step_back (which settles by thread root) could never touch them; the ear then reopened the 7951 debt three times on 'the work is still outstanding' logic. Result: repeated blocked-QA announcements and an 'i've got capacity' interjection into a team she is not on. Three fixes: - open_ask requires an anchor; a top-level ask roots on its own ts (the router's convention), so in-thread answers and step_back can actually settle it - reopenAttentionItem refuses operator closes; step-back closes stay reopenable per SPEC ('the ear MAY reopen one that truly was hers') - ear soul: an ask to the room or a team is not hers to claim unless a name or a standing rule makes it hers; unfinished work is not an unanswered ask Co-Authored-By: Claude Fable 5 --- src/ledger/attention.ts | 6 ++++- src/service.ts | 14 +++++++--- src/turn-runner/ear-soul.ts | 5 +++- test/ear.test.ts | 54 +++++++++++++++++++++++++++++++++++++ 4 files changed, 74 insertions(+), 5 deletions(-) diff --git a/src/ledger/attention.ts b/src/ledger/attention.ts index 3cd22bf..85eadf5 100644 --- a/src/ledger/attention.ts +++ b/src/ledger/attention.ts @@ -50,7 +50,11 @@ export function closeAttentionItem(db: Database, clock: Clock, id: string, cause } export function reopenAttentionItem(db: Database, id: string): boolean { - return db.query("UPDATE attention_items SET closed_at = NULL, closed_cause = NULL WHERE id = ?").run(id).changes > 0; + // "The ear MAY reopen one that truly was hers" (SPEC §13) covers its own closes and even a + // step_back's — but never an operator's close: that judgment outranks the ear's. + return db + .query("UPDATE attention_items SET closed_at = NULL, closed_cause = NULL WHERE id = ? AND (closed_cause IS NULL OR closed_cause NOT LIKE 'operator:%')") + .run(id).changes > 0; } export function openItems(db: Database, identityId: string, limit = 50): AttentionItem[] { diff --git a/src/service.ts b/src/service.ts index f24b63b..5a1bff4 100644 --- a/src/service.ts +++ b/src/service.ts @@ -427,19 +427,27 @@ export class Service { if (a.venueId) notes.push(`<#${a.venueId}>${a.threadRootId ? ` thread=${a.threadRootId}` : ""}: ${a.why}`); else notes.push(a.why); } else if (a.decision === "open_ask") { - if (!a.venueId) return { success: false, output: "open_ask needs venueId (and threadRootId/askTs when known)" }; + if (!a.venueId || (!a.threadRootId && !a.askTs)) { + return { success: false, output: "open_ask needs venueId plus where the ask lives: its threadRootId (the thread= value), or the message's own ts as askTs for a top-level ask" }; + } openAttentionItem(this.d.db, this.d.clock, { id: this.d.newId(), identityId, venueId: a.venueId, - threadRootId: a.threadRootId ?? null, + // A top-level ask roots the thread its replies will carry (the router's own + // convention). An anchor-less debt can never be settled by an in-thread answer or + // a step_back, so it rides every wake until the ear happens to close it (live + // 2026-07-23: two orphaned QA debts she kept announcing blockers on). + threadRootId: a.threadRootId ?? a.askTs ?? null, askTs: a.askTs ?? null, what: a.why, }); } else if (a.decision === "close_ask") { if (!a.itemId || !closeAttentionItem(this.d.db, this.d.clock, a.itemId, a.why)) return { success: false, output: "no open item with that id" }; } else if (a.decision === "reopen_ask") { - if (!a.itemId || !reopenAttentionItem(this.d.db, a.itemId)) return { success: false, output: "no item with that id" }; + if (!a.itemId || !reopenAttentionItem(this.d.db, a.itemId)) { + return { success: false, output: "nothing to reopen with that id: either it does not exist, or the operator settled it and that stays settled" }; + } } return { success: true, output: "noted" }; }, diff --git a/src/turn-runner/ear-soul.ts b/src/turn-runner/ear-soul.ts index 52d663f..dba4b0c 100644 --- a/src/turn-runner/ear-soul.ts +++ b/src/turn-runner/ear-soul.ts @@ -32,7 +32,10 @@ Needing someone is not needing her. When people are talking to each other, the c theirs: a question aimed at another teammate is that person's to answer even when she knows the answer, and waking her into it costs the room more than it gives. The same boundary holds for debts: record only asks aimed at her. What one teammate owes another is theirs, not hers to -carry or to chase. +carry or to chase. An ask to the room or a team belongs to whoever steps up or gets named, and +open work is not hers to claim unless a name or a standing rule makes it hers. Unfinished work +is not an unanswered ask: once she answered, was told it is not hers, or stepped away, that +debt is settled, and only a fresh ask aimed at her opens a new one. Bias to hold. Most of what you hear needs nothing from her, and waking her for it costs the room more than it gives. But a real ask with no answer is the one failure you exist to prevent: when diff --git a/test/ear.test.ts b/test/ear.test.ts index 3535c24..b950242 100644 --- a/test/ear.test.ts +++ b/test/ear.test.ts @@ -210,6 +210,60 @@ describe("attention items (what she owes)", () => { await service.stop(); }); + test("an anchor-less open_ask is refused; askTs alone roots the debt so stepping back settles it", async () => { + // Live 2026-07-23: the ear recorded two QA debts with no thread coordinates; step_back and + // in-thread answers settle by thread root, so the orphans rode every wake and were reopened + // repeatedly. A top-level ask roots on its own ts (the router's convention). + let earCalls = 0; + let bad: { success: boolean; output: string } | undefined; + const { db, service, adapter } = harness(async (_turn, tools) => { + const verdict = tools.get("verdict"); + if (verdict) { + if (++earCalls === 1) { + bad = (await verdict.run({ decision: "open_ask", why: "qa is needed on the preview", venueId: "C1" })) as { success: boolean; output: string }; + await verdict.run({ decision: "open_ask", why: "qa is needed on the preview", venueId: "C1", askTs: "5.0" }); + await verdict.run({ decision: "wake", why: "an open qa request with no taker", venueId: "C1", threadRootId: "5.0" }); + } + return; + } + await tools.get("step_back")!.run({ why: "not mine to claim", venueId: "C1", threadRootId: "5.0" }); + }); + await service.start(); + adapter.emit(msg({ text: "Needs QA: check the upload dialog", ts: "5.0" })); + await service.idle(); + + expect(bad!.success).toBe(false); + expect(openItems(db, "eng")).toHaveLength(0); // the askTs-rooted debt settled with her step_back + await service.stop(); + }); + + test("an operator-settled debt stays settled — the ear's reopen is refused", async () => { + let earCalls = 0; + let openedId = ""; + let reopen: { success: boolean; output: string } | undefined; + const { db, clock, service, adapter } = harness(async (_turn, tools) => { + const verdict = tools.get("verdict"); + if (!verdict) return; // the mind stays idle in this test + if (++earCalls === 1) { + await verdict.run({ decision: "open_ask", why: "qa still outstanding", venueId: "C1", threadRootId: "6.0", askTs: "6.1" }); + return; + } + reopen = (await verdict.run({ decision: "reopen_ask", why: "the work is still not done", itemId: openedId })) as { success: boolean; output: string }; + }); + await service.start(); + adapter.emit(msg({ text: "needs qa", ts: "6.1", threadRootTs: "6.0" })); + await service.idle(); + openedId = openItems(db, "eng")[0]!.id; + const { closeAttentionItem } = await import("../src/ledger/attention"); + closeAttentionItem(db, clock, openedId, "operator: not her work"); + adapter.emit(msg({ text: "still not done", ts: "6.2", threadRootTs: "6.0" })); + await service.idle(); + + expect(reopen!.success).toBe(false); + expect(openItems(db, "eng")).toHaveLength(0); + await service.stop(); + }); + test("the owed section is capped and an overdue item is flagged to the mind's own judgment", async () => { let earCalls = 0; const { clock, service, adapter, mindSessions } = harness(async (_turn, tools) => { From bb9f85801d136d79ded8872cf4b7b6604e25ddda Mon Sep 17 00:00:00 2001 From: Noah Lindner Date: Thu, 23 Jul 2026 10:43:04 -0400 Subject: [PATCH 4/5] replay harness: relive a recorded incident with real model calls, captured room MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit earshot replay --db --from --to [--venue C…] [--speed N] Carves the window's surface messages out of a ledger snapshot (RawMessage round-trip from the router's stored payloads), rewinds the snapshot COPY to the moment before the window (events + contentless-FTS docs, turns, attention items incl. re-opening ones closed in-window, participation step-backs, cursors; tasks/executions/steering/timers cleared — scheduler state firing mid-replay is noise), then relives the traffic through the real Service at recorded pacing: - real codex sessions via the same factory as start (extracted, shared) — a replay driving different wiring would test the wrong bot - CaptureAdapter: replies/reactions recorded, never delivered; no streaming methods so every reply funnels through the plain-post capture point; read_thread serves the room as recorded - integration registries keep their real specs but record instead of executing (writes report done; reads report unavailable) — the report shows what she reached for without touching Linear/GitHub/Notion - --speed compresses gaps; speed 1 preserves mid-turn race timing Report prints her original window actions next to the replay's. Co-Authored-By: Claude Fable 5 --- src/main.ts | 110 ++++++++++++++++++++--- src/replay/incident.ts | 115 ++++++++++++++++++++++++ src/replay/run.ts | 193 ++++++++++++++++++++++++++++++++++++++++ test/replay.test.ts | 194 +++++++++++++++++++++++++++++++++++++++++ 4 files changed, 602 insertions(+), 10 deletions(-) create mode 100644 src/replay/incident.ts create mode 100644 src/replay/run.ts create mode 100644 test/replay.test.ts diff --git a/src/main.ts b/src/main.ts index c3ab73c..690f601 100644 --- a/src/main.ts +++ b/src/main.ts @@ -24,6 +24,8 @@ usage: earshot start run the daemon: connect to Slack, drive tasks via codex, survive restarts earshot doctor check codex login, env vars, and that the policy file validates earshot status one-shot snapshot: open tasks + running executions per identity + earshot replay relive a recorded incident from a ledger snapshot with real model calls, + against a captured room (nothing reaches Slack). See: earshot replay --help config (env): EARSHOT_DB ledger path (default ./earshot.db) @@ -138,16 +140,7 @@ async function cmdStart(): Promise { catalog, registries, newId: () => `${Date.now().toString(36)}-${(counter++).toString(36)}`, - // overrides carry a task tier's model/effort (policy.models): codex accepts -c config - // overrides ahead of the subcommand, so each worker session runs on its tier while the - // resident mind stays on the runtime default (config.toml). - sessionFactory: (tools: DynamicTool[], onEvent, overrides) => { - const flags = [overrides?.model ? `-c model=${JSON.stringify(overrides.model)}` : "", overrides?.effort ? `-c model_reasoning_effort=${JSON.stringify(overrides.effort)}` : ""] - .filter(Boolean) - .join(" "); - const config = flags ? { ...DEFAULT_CODEX_CONFIG, command: `codex ${flags} app-server` } : DEFAULT_CODEX_CONFIG; - return new AppServerSession(config, tools, onEvent ?? ((e) => e.log && log.info("codex", { line: e.log })), { scrubEnv: scrubSecrets }); - }, + sessionFactory: makeCodexSessionFactory(log), logger: log, heartbeatMs: 1000, }); @@ -191,6 +184,101 @@ async function cmdStart(): Promise { process.on("unhandledRejection", (e) => console.error("[main] unhandled rejection:", e)); } +// The one real codex wiring, shared by start and replay — a replay that drives a different +// session factory than production would test the wrong bot. overrides carry a task tier's +// model/effort (policy.models): codex accepts -c config overrides ahead of the subcommand, so +// each worker session runs on its tier while the resident mind stays on the runtime default. +function makeCodexSessionFactory(log: ReturnType) { + return (tools: DynamicTool[], onEvent?: (e: import("./turn-runner/types").AgentEvent) => void, overrides?: { model?: string; effort?: string }) => { + const flags = [overrides?.model ? `-c model=${JSON.stringify(overrides.model)}` : "", overrides?.effort ? `-c model_reasoning_effort=${JSON.stringify(overrides.effort)}` : ""] + .filter(Boolean) + .join(" "); + const config = flags ? { ...DEFAULT_CODEX_CONFIG, command: `codex ${flags} app-server` } : DEFAULT_CODEX_CONFIG; + return new AppServerSession(config, tools, onEvent ?? ((e) => e.log && log.info("codex", { line: e.log })), { scrubEnv: scrubSecrets }); + }; +} + +const REPLAY_HELP = `earshot replay — relive a recorded incident with real model calls, captured room. + +usage: + earshot replay --db --from --to [--venue C…] [--speed N] + +The snapshot is COPIED into the workspace and rewound to the window start; the original file is +never touched. Inbound messages replay at recorded pacing (--speed N compresses gaps N-fold; +speed 1 is truest to mid-turn races). Replies, reactions, and external tool calls are captured +and printed against what she originally did — nothing reaches Slack, Linear, GitHub, or Notion. + +needs: codex logged in, EARSHOT_POLICY (or ./policy.yaml), and the workspace dirs codex-trusted. + --db path to a ledger snapshot (scp it from the live box first) + --from/--to ISO-8601 UTC window bounds, e.g. 2026-07-23T12:00:00Z + --venue only replay messages from one venue id + --speed gap compression factor (default 1) + --workspace scratch dir for the replay's codex sessions (default ./replay-workspace) + --bot-id bot principal id (default SLACK_BOT_USER_ID, else UREPLAY) +`; + +async function cmdReplay(): Promise { + if (process.argv.includes("--help")) { + console.log(REPLAY_HELP); + return; + } + const arg = (name: string) => { + const i = process.argv.indexOf(`--${name}`); + return i >= 0 ? process.argv[i + 1] : undefined; + }; + const snapshot = arg("db"); + const from = arg("from"); + const to = arg("to"); + if (!snapshot || !from || !to) { + console.log(REPLAY_HELP); + process.exit(1); + } + const { loadIncident, originalActions, rewindLedger } = await import("./replay/incident"); + const { runReplay } = await import("./replay/run"); + const { copyFileSync } = await import("node:fs"); + + const workspace = arg("workspace") ?? "./replay-workspace"; + mkdirSync(workspace, { recursive: true }); + const copy = join(workspace, "replay.db"); + copyFileSync(snapshot, copy); // rewind is destructive — never open the snapshot itself + const db = openLedger(copy); + const store = makeStore(); + const log = createLogger(); + + const venue = arg("venue"); + const events = loadIncident(db, { fromIso: from, toIso: to, ...(venue ? { venueId: venue } : {}) }); + if (events.length === 0) { + console.error("no surface messages in that window"); + process.exit(1); + } + const original = originalActions(db, from, to); + const rewound = rewindLedger(db, events[0]!.rowid, from); + console.log( + `rewound to ${from}: ${rewound.events} events, ${rewound.turns} turns, ${rewound.itemsDeleted}+${rewound.itemsReopened} attention items, ` + + `${rewound.tasks} tasks, ${rewound.timers} timers cleared` + + (rewound.memoriesInWindow ? ` (caveat: ${rewound.memoriesInWindow} memories written in-window stay — no edit history to rewind)` : ""), + ); + console.log(`replaying ${events.length} messages at speed ${arg("speed") ?? "1"}…\n`); + + const captured = await runReplay({ + db, + events, + policyStore: store, + sessionFactory: makeCodexSessionFactory(log), + workspace, + botPrincipalId: arg("bot-id") ?? process.env.SLACK_BOT_USER_ID ?? "UREPLAY", + speed: Number(arg("speed") ?? "1"), + logger: log, + }); + + const show = (kind: string, detail: unknown) => ` ${kind}: ${JSON.stringify(detail)}`; + console.log("\n=== originally ==="); + for (const t of original) for (const e of t.effects as { kind?: string }[]) console.log(show(e.kind ?? "?", e)); + console.log("\n=== in replay ==="); + for (const c of captured) console.log(show(c.kind, c.detail)); + db.close(); +} + async function cmdDoctor(): Promise { const codexOk = await codexReady(); console.log(`${codexOk ? "ok " : "MISSING "}codex logged in`); @@ -253,6 +341,8 @@ async function main(): Promise { return cmdDoctor(); case "status": return cmdStatus(); + case "replay": + return cmdReplay(); default: console.log(HELP); } diff --git a/src/replay/incident.ts b/src/replay/incident.ts new file mode 100644 index 0000000..fe51271 --- /dev/null +++ b/src/replay/incident.ts @@ -0,0 +1,115 @@ +// Replay harness (dev tool, not part of the daemon): carve a real incident out of a ledger +// snapshot and rewind the snapshot to the moment before it, so the service can relive the same +// inbound traffic — real model judgment, captured room (run.ts). Rewind is destructive: always +// run it on a COPY of the ledger, never the live file (the CLI copies before opening). +import type { Database } from "bun:sqlite"; +import type { RawMessage, MessageFile } from "@bevyl-ai/agent-tools"; + +export interface IncidentEvent { + rowid: number; + receivedAt: string; + message: RawMessage; +} + +export interface IncidentWindow { + fromIso: string; + toIso: string; + venueId?: string; // omit to replay every venue active in the window +} + +interface EventRow { + rowid: number; + venue_id: string | null; + thread_root_id: string | null; + principal_id: string | null; + payload: string; + received_at: string; +} + +// Surface messages in the window, reconstructed into the RawMessage the adapter originally +// delivered. addressMode (the router's output, stored in the payload) round-trips to the inbound +// flags: a mention is the only source of mentionsBotId, and dm is the only non-channel venueKind +// the router ever records. external_signal rows are excluded — those are the system's own +// productions (worker outcomes, timers) and the replay's service re-derives them itself. +export function loadIncident(db: Database, w: IncidentWindow): IncidentEvent[] { + const rows = db + .query( + `SELECT rowid, venue_id, thread_root_id, principal_id, payload, received_at FROM events + WHERE kind IN ('addressed_message','observed_message') AND received_at >= ? AND received_at < ? + ${w.venueId ? "AND venue_id = ?" : ""} ORDER BY rowid`, + ) + .all(...(w.venueId ? [w.fromIso, w.toIso, w.venueId] : [w.fromIso, w.toIso])) as EventRow[]; + return rows.map((r) => { + const p = JSON.parse(r.payload) as { text?: string; ts?: string; isBot?: boolean; addressMode?: string; files?: MessageFile[] }; + return { + rowid: r.rowid, + receivedAt: r.received_at, + message: { + venueId: r.venue_id ?? "", + venueKind: p.addressMode === "dm" ? ("dm" as const) : ("channel" as const), + principalId: r.principal_id, + isBot: p.isBot ?? false, + text: p.text ?? "", + ts: p.ts ?? "", + threadRootTs: r.thread_root_id, + mentionsBotId: p.addressMode === "mention", + ...(p.files?.length ? { files: p.files } : {}), + }, + }; + }); +} + +export interface OriginalTurn { + startedAt: string; + kind: string; + effects: unknown[]; +} + +// What she actually did in the window — read BEFORE rewindLedger, which deletes these rows. +export function originalActions(db: Database, fromIso: string, toIso: string): OriginalTurn[] { + const rows = db + .query("SELECT started_at, kind, effects FROM turns WHERE started_at >= ? AND started_at < ? AND kind IN ('resident','attention') ORDER BY started_at") + .all(fromIso, toIso) as { started_at: string; kind: string; effects: string }[]; + return rows.map((r) => ({ startedAt: r.started_at, kind: r.kind, effects: JSON.parse(r.effects) as unknown[] })); +} + +export interface RewindReport { + events: number; + turns: number; + itemsDeleted: number; + itemsReopened: number; + tasks: number; + timers: number; + memoriesInWindow: number; // NOT rewound (no edit history) — reported so the caveat is visible +} + +// Point-in-time rewind: everything the service wrote at or after the window start is unwound so +// the replay's own passes rebuild it. Participation stepped-back during the window is un-stepped +// (it had not happened yet); the rows themselves stay — participation without traffic is inert. +// Tasks, executions, steering, and timers are cleared outright: a replay relives conversations, +// and a snapshot's scheduler state firing mid-replay is noise, not fidelity. Memory edits cannot +// be rewound (items carry no edit history); the count is reported instead. +export function rewindLedger(db: Database, cutoffRowid: number, fromIso: string): RewindReport { + const tx = db.transaction(() => { + // events_fts is contentless (content='') with an insert-only trigger, so doomed docs must be + // removed explicitly — an fts5 'delete' needs the original text back. + const doomed = db + .query("SELECT rowid, coalesce(json_extract(payload,'$.text'),'') AS text FROM events WHERE rowid >= ?") + .all(cutoffRowid) as { rowid: number; text: string }[]; + for (const d of doomed) db.query("INSERT INTO events_fts (events_fts, rowid, text) VALUES ('delete', ?, ?)").run(d.rowid, d.text); + const events = db.query("DELETE FROM events WHERE rowid >= ?").run(cutoffRowid).changes; + const turns = db.query("DELETE FROM turns WHERE started_at >= ?").run(fromIso).changes; + const itemsDeleted = db.query("DELETE FROM attention_items WHERE opened_at >= ?").run(fromIso).changes; + const itemsReopened = db.query("UPDATE attention_items SET closed_at = NULL, closed_cause = NULL WHERE closed_at >= ?").run(fromIso).changes; + db.query("UPDATE thread_participation SET stepped_back_at = NULL, stepped_back_why = NULL WHERE stepped_back_at >= ?").run(fromIso); + db.query("UPDATE resident_cursor SET delivered_rowid = min(delivered_rowid, ?)").run(cutoffRowid - 1); + db.query("UPDATE ear_cursor SET judged_rowid = min(judged_rowid, ?)").run(cutoffRowid - 1); + const timers = db.query("DELETE FROM timers").run().changes; + db.query("DELETE FROM steering").run(); + db.query("DELETE FROM executions").run(); + const tasks = db.query("DELETE FROM tasks").run().changes; + const memoriesInWindow = (db.query("SELECT count(*) AS n FROM memory_items WHERE created_at >= ?").get(fromIso) as { n: number }).n; + return { events, turns, itemsDeleted, itemsReopened, tasks, timers, memoriesInWindow }; + }); + return tx(); +} diff --git a/src/replay/run.ts b/src/replay/run.ts new file mode 100644 index 0000000..ab27208 --- /dev/null +++ b/src/replay/run.ts @@ -0,0 +1,193 @@ +// The replay run: the REAL Service (router, ear, wakes, ledger) reliving an incident's inbound +// traffic with real model calls, against a capture surface — nothing reaches Slack, external +// write tools record instead of executing, and reads are served from the snapshot itself. +import type { Database } from "bun:sqlite"; +import type { SurfaceAdapter, RawMessage, PostResult, MessageFile } from "@bevyl-ai/agent-tools"; +import { Service, type ServiceDeps } from "../service"; +import { INTEGRATION_REGISTRIES, flattenRegistries, type ToolRegistry } from "../tools/catalog"; +import { systemClock, type Clock } from "../ledger/clock"; +import type { PolicyStore } from "../policy/load"; +import type { Logger } from "../log"; +import type { IncidentEvent } from "./incident"; + +export interface CapturedAction { + at: string; + kind: "post" | "reaction" | "external_tool"; + detail: Record; +} + +type ThreadMsg = { user: string | null; text: string; ts: string; files?: MessageFile[] }; + +// A surface that captures instead of delivering. Streaming methods are deliberately absent so +// every reply funnels through the plain-post fallback — one capture point, no stream bookkeeping. +// readThread serves the room as recorded: snapshot history seeded at construction, replayed +// messages appended as they're emitted. +class CaptureAdapter implements SurfaceAdapter { + readonly captured: CapturedAction[] = []; + private handlers: Array<(msg: RawMessage) => void> = []; + private threads = new Map(); + private nextId = 1; + + constructor( + private clock: Clock, + db: Database, + ) { + const rows = db + .query("SELECT venue_id, thread_root_id, principal_id, payload FROM events WHERE kind IN ('addressed_message','observed_message') ORDER BY rowid") + .all() as { venue_id: string | null; thread_root_id: string | null; principal_id: string | null; payload: string }[]; + for (const r of rows) { + const p = JSON.parse(r.payload) as { text?: string; ts?: string; files?: MessageFile[] }; + if (!p.ts) continue; + this.append(r.thread_root_id ?? p.ts, { user: r.principal_id, text: p.text ?? "", ts: p.ts, ...(p.files?.length ? { files: p.files } : {}) }); + } + } + + private append(root: string, msg: ThreadMsg): void { + const list = this.threads.get(root) ?? []; + list.push(msg); + this.threads.set(root, list); + } + + async start(): Promise {} + stop(): void {} + + onMessage(handler: (msg: RawMessage) => void): void { + this.handlers.push(handler); + } + + emit(msg: RawMessage): void { + this.append(msg.threadRootTs ?? msg.ts, { user: msg.principalId, text: msg.text, ts: msg.ts, ...(msg.files?.length ? { files: msg.files } : {}) }); + for (const h of this.handlers) h(msg); + } + + async postMessage(venueId: string, threadRootTs: string | null, text: string): Promise { + this.captured.push({ at: this.clock(), kind: "post", detail: { venueId, threadRootTs, text } }); + return { messageId: `replay-${this.nextId++}` }; + } + + async addReaction(venueId: string, messageId: string, emoji: string): Promise { + this.captured.push({ at: this.clock(), kind: "reaction", detail: { venueId, messageId, emoji } }); + } + + async readThread(_venueId: string, threadTs: string): Promise { + return this.threads.get(threadTs) ?? []; + } + + async setTypingStatus(): Promise {} +} + +// The integration registries with their real specs but recording implementations: a write +// (any action-classed call) reports success without executing; a read reports itself +// unavailable — both are captured so the report shows what she reached for. +export function recordingRegistries(captured: CapturedAction[], clock: Clock): ToolRegistry[] { + return INTEGRATION_REGISTRIES.map((r) => ({ + ...r, + tools: Object.fromEntries( + Object.entries(r.tools).map(([name, spec]) => [ + name, + { + ...spec, + run: async (args: unknown) => { + captured.push({ at: clock(), kind: "external_tool", detail: { tool: name, args } }); + const outward = (spec.actionClasses?.(args) ?? []).length > 0; + return outward ? { success: true, output: "done" } : { success: false, output: "that lookup is not available right now" }; + }, + }, + ]), + ), + })); +} + +// read_channel / read_thread served from the snapshot's own events, mirroring main.ts's live +// slack registry (same names, so existing grants validate and expose them identically). +export function snapshotSlackRegistry(db: Database): ToolRegistry { + const messages = (where: string, params: string[], limit: number) => + db + .query( + `SELECT venue_id, thread_root_id, principal_id, payload FROM events + WHERE kind IN ('addressed_message','observed_message') AND ${where} ORDER BY rowid DESC LIMIT ?`, + ) + .all(...params, limit) + .reverse() + .map((row) => { + const r = row as { principal_id: string | null; payload: string }; + const p = JSON.parse(r.payload) as { text?: string; ts?: string }; + return { user: r.principal_id, text: p.text ?? "", ts: p.ts ?? "" }; + }); + return { + name: "slack", + skill: "Beyond the thread in front of you: pull a channel's recent history on demand, then open any conversation it roots.", + tools: { + read_channel: { + description: "Read recent messages from a Slack channel. Input: { channel, limit? } — channel as <#C…> link or id.", + inputSchema: { type: "object", additionalProperties: false, required: ["channel"], properties: { channel: { type: "string" }, limit: { type: "number" } } }, + run: async (args: unknown) => { + const a = (args ?? {}) as { channel?: string; limit?: number }; + const venueId = a.channel?.replace(/^<#|[|>].*$/g, ""); + if (!venueId) return { success: false, output: "read_channel needs a { channel }" }; + return { success: true, output: JSON.stringify(messages("venue_id = ? AND thread_root_id IS NULL", [venueId], Math.min(a.limit ?? 20, 100))) }; + }, + }, + read_thread: { + description: "Read a Slack thread's replies. Input: { channel, thread_ts, limit? }.", + inputSchema: { type: "object", additionalProperties: false, required: ["channel", "thread_ts"], properties: { channel: { type: "string" }, thread_ts: { type: "string" }, limit: { type: "number" } } }, + run: async (args: unknown) => { + const a = (args ?? {}) as { channel?: string; thread_ts?: string; limit?: number }; + if (!a.channel || !a.thread_ts) return { success: false, output: "read_thread needs { channel, thread_ts }" }; + return { success: true, output: JSON.stringify(messages("thread_root_id = ?", [a.thread_ts], Math.min(a.limit ?? 50, 200))) }; + }, + }, + }, + }; +} + +export interface ReplayOpts { + db: Database; + events: IncidentEvent[]; + policyStore: PolicyStore; + sessionFactory: ServiceDeps["sessionFactory"]; + workspace: string; + botPrincipalId: string; + speed?: number; // 1 = recorded pacing (truest to mid-turn races); N compresses gaps N-fold + clock?: Clock; + logger?: Logger; + out?: (line: string) => void; +} + +// Feed the incident through a fresh Service at recorded pacing and return everything she did. +// The db must already be rewound (incident.ts) — this function only relives and captures. +export async function runReplay(opts: ReplayOpts): Promise { + const clock = opts.clock ?? systemClock; + const out = opts.out ?? ((line: string) => console.log(line)); + const speed = opts.speed ?? 1; + const adapter = new CaptureAdapter(clock, opts.db); + const registries = [...recordingRegistries(adapter.captured, clock), snapshotSlackRegistry(opts.db)]; + let n = 0; + const service = new Service({ + db: opts.db, + clock, + policyStore: opts.policyStore, + adapter, + botPrincipalId: opts.botPrincipalId, + cwd: opts.workspace, + catalog: flattenRegistries(registries), + registries, + newId: () => `replay-${Date.now().toString(36)}-${(n++).toString(36)}`, + sessionFactory: opts.sessionFactory, + ...(opts.logger ? { logger: opts.logger } : {}), + heartbeatMs: 1000, + }); + await service.start(); + const t0 = Date.parse(opts.events[0]!.receivedAt); + const started = Date.now(); + for (const e of opts.events) { + const wait = started + (Date.parse(e.receivedAt) - t0) / speed - Date.now(); + if (wait > 0) await new Promise((r) => setTimeout(r, wait)); + const where = `${e.message.venueId}${e.message.threadRootTs ? ` thread=${e.message.threadRootTs}` : ""}`; + out(`⟳ ${e.receivedAt} [${where}] <${e.message.principalId ?? "?"}>: ${e.message.text.slice(0, 120)}`); + adapter.emit(e.message); + } + await service.idle(); + await service.stop(); + return adapter.captured; +} diff --git a/test/replay.test.ts b/test/replay.test.ts new file mode 100644 index 0000000..927bfad --- /dev/null +++ b/test/replay.test.ts @@ -0,0 +1,194 @@ +import { describe, expect, test } from "bun:test"; +import { openLedger } from "../src/ledger/db"; +import { PolicyStore } from "../src/policy/load"; +import { Service } from "../src/service"; +import { pendingMessages } from "../src/ledger/inbox"; +import { openItems, openAttentionItem, closeAttentionItem } from "../src/ledger/attention"; +import { loadIncident, originalActions, rewindLedger } from "../src/replay/incident"; +import { runReplay, recordingRegistries } from "../src/replay/run"; +import { FakeAdapter } from "./fakes/fake-adapter"; +import { FakeAgentRuntimeSession } from "./fakes/fake-runtime-session"; +import type { DynamicTool } from "../src/turn-runner/types"; +import type { Clock } from "../src/ledger/clock"; +import type { RawMessage } from "@bevyl-ai/agent-tools"; + +// The replay harness (src/replay): carve a recorded incident out of a ledger snapshot, rewind +// the snapshot to the moment before it, and relive it through the real Service against a capture +// surface. Codex is faked here per the repo's test rules — the CLI injects the real factory. + +function fakeClock(start = "2026-07-02T00:00:00Z"): Clock & { set: (iso: string) => void } { + let now = start; + const clock = (() => now) as Clock & { set: (iso: string) => void }; + clock.set = (iso: string) => { + now = iso; + }; + return clock; +} + +const POLICY_YAML = ` +surface: + kind: slack + credentials: + bot_token: $BOT +operator_principals: + - U_OPERATOR +identities: + - id: eng + venue_ids: [C1] + budget: { monthly_cap: 1000 } +turns: + backoff_ms: 1 +budget: + global_monthly_cap: 100000 +`; + +function policyStore(): PolicyStore { + return new PolicyStore(() => POLICY_YAML, { knownTools: new Set(), envAvailable: () => true }); +} + +function msg(overrides: Partial = {}): RawMessage { + return { + venueId: "C1", + venueKind: "channel", + principalId: "U1", + isBot: false, + text: "hello", + ts: `${Date.now()}.${Math.random().toString().slice(2, 8)}`, + threadRootTs: null, + mentionsBotId: false, + ...overrides, + }; +} + +// Record phase: run a real service over the fake adapter so the ledger fills exactly the way a +// live one would (router-written payloads, cursors, turns). Ids must be unique across the db's +// LIFETIME, not one service run — a reused id is silently dropped as a duplicate event. +let idCounter = 0; +async function record(db: ReturnType, clock: Clock, messages: RawMessage[], script: ConstructorParameters[1]) { + const adapter = new FakeAdapter(); + const service = new Service({ + db, + clock, + policyStore: policyStore(), + adapter, + botPrincipalId: "BOT1", + cwd: "/tmp", + earCwd: "/tmp/ear-test", + newId: () => `rec-${++idCounter}`, + sessionFactory: (tools: DynamicTool[]) => new FakeAgentRuntimeSession(tools, script), + }); + await service.start(); + for (const m of messages) { + adapter.emit(m); + await service.idle(); + } + await service.stop(); +} + +describe("replay: incident loading", () => { + test("messages round-trip: a mention regains mentionsBotId, thread and files survive, window filters apply", async () => { + const db = openLedger(":memory:"); + const clock = fakeClock("2026-07-02T00:00:00Z"); + await record(db, clock, [msg({ text: "before the window", ts: "1.0" })], async () => {}); + clock.set("2026-07-02T10:00:00Z"); + await record(db, clock, [ + msg({ text: "<@BOT1> look at this", mentionsBotId: true, ts: "2.0", files: [{ id: "F1", name: "shot.png", mimetype: "image/png", urlPrivate: "u", size: 1 }] }), + msg({ text: "a thread reply", ts: "2.1", threadRootTs: "2.0", principalId: "U2" }), + ], async () => {}); + + const events = loadIncident(db, { fromIso: "2026-07-02T10:00:00Z", toIso: "2026-07-02T11:00:00Z" }); + expect(events).toHaveLength(2); + expect(events[0]!.message).toMatchObject({ text: "<@BOT1> look at this", mentionsBotId: true, ts: "2.0", threadRootTs: null, venueKind: "channel" }); + expect(events[0]!.message.files).toHaveLength(1); + expect(events[1]!.message).toMatchObject({ text: "a thread reply", mentionsBotId: false, threadRootTs: "2.0", principalId: "U2" }); + }); +}); + +describe("replay: rewind", () => { + test("rewind unwinds the window — events, turns, attention items, cursors — and leaves the past intact", async () => { + const db = openLedger(":memory:"); + const clock = fakeClock("2026-07-02T00:00:00Z"); + await record(db, clock, [msg({ text: "<@BOT1> old business", mentionsBotId: true, ts: "1.0" })], async (_t, tools) => { + if (tools.get("verdict")) return; + await tools.get("reply")!.run({ text: "handled", venueId: "C1", threadRootId: "1.0" }); + }); + // an item opened before the window but closed during it must come back open + openAttentionItem(db, clock, { id: "old-item", identityId: "eng", venueId: "C1", threadRootId: "1.0", askTs: null, what: "an old debt" }); + clock.set("2026-07-02T10:00:00Z"); + await record(db, clock, [msg({ text: "<@BOT1> new business", mentionsBotId: true, ts: "2.0" })], async (_t, tools) => { + if (tools.get("verdict")) return; + await tools.get("reply")!.run({ text: "on it", venueId: "C1", threadRootId: "2.0" }); + }); + closeAttentionItem(db, clock, "old-item", "answered in thread"); + openAttentionItem(db, clock, { id: "new-item", identityId: "eng", venueId: "C1", threadRootId: "2.0", askTs: null, what: "a window debt" }); + + const events = loadIncident(db, { fromIso: "2026-07-02T10:00:00Z", toIso: "2026-07-02T11:00:00Z" }); + const original = originalActions(db, "2026-07-02T10:00:00Z", "2026-07-02T11:00:00Z"); + expect(original.flatMap((t) => t.effects as { kind?: string; text?: string }[]).some((e) => e.text === "on it")).toBe(true); + + const report = rewindLedger(db, events[0]!.rowid, "2026-07-02T10:00:00Z"); + expect(report.events).toBeGreaterThanOrEqual(1); + expect(report.turns).toBeGreaterThanOrEqual(1); + // the window is gone… + expect(originalActions(db, "2026-07-02T10:00:00Z", "2026-07-02T11:00:00Z")).toHaveLength(0); + expect(loadIncident(db, { fromIso: "2026-07-02T10:00:00Z", toIso: "2026-07-02T11:00:00Z" })).toHaveLength(0); + // …the past is not… + expect(loadIncident(db, { fromIso: "2026-07-02T00:00:00Z", toIso: "2026-07-02T01:00:00Z" })).toHaveLength(1); + // …the closed-in-window item is open again, the opened-in-window item is gone… + expect(openItems(db, "eng").map((i) => i.id)).toEqual(["old-item"]); + // …and nothing is pending: the cursor sits exactly at the end of the remaining events. + expect(pendingMessages(db, "eng")).toHaveLength(0); + }); +}); + +describe("replay: reliving", () => { + test("a rewound incident re-runs through the real pipeline; her actions are captured, nothing reaches the fake room", async () => { + const db = openLedger(":memory:"); + const clock = fakeClock("2026-07-02T00:00:00Z"); + await record(db, clock, [msg({ text: "<@BOT1> keep an eye out", mentionsBotId: true, ts: "1.0" })], async () => {}); + clock.set("2026-07-02T10:00:00Z"); + await record(db, clock, [msg({ text: "<@BOT1> what broke?", mentionsBotId: true, ts: "2.0", principalId: "U_NOAH" })], async (_t, tools) => { + if (tools.get("verdict")) return; + await tools.get("reply")!.run({ text: "the original answer", venueId: "C1", threadRootId: "2.0" }); + }); + + const events = loadIncident(db, { fromIso: "2026-07-02T10:00:00Z", toIso: "2026-07-02T11:00:00Z" }); + rewindLedger(db, events[0]!.rowid, "2026-07-02T10:00:00Z"); + + const prompts: string[] = []; + const captured = await runReplay({ + db, + events, + policyStore: policyStore(), + sessionFactory: (tools: DynamicTool[]) => + new FakeAgentRuntimeSession(tools, async (_t, sessionTools) => { + if (sessionTools.get("verdict")) return; + await sessionTools.get("reply")!.run({ text: "the replayed answer", venueId: "C1", threadRootId: "2.0" }); + }), + workspace: "/tmp", + botPrincipalId: "BOT1", + clock, + out: (line) => prompts.push(line), + }); + + const posts = captured.filter((c) => c.kind === "post"); + expect(posts).toHaveLength(1); + expect(posts[0]!.detail["text"]).toBe("the replayed answer"); + expect(prompts.some((l) => l.includes("what broke?"))).toBe(true); // the run narrates each replayed line + }); + + test("recording registries: a write reports done without executing, a read reports unavailable — both captured", async () => { + const captured: Parameters[0] = []; + const registries = recordingRegistries(captured, fakeClock()); + const linearWrite = registries.flatMap((r) => Object.entries(r.tools)).find(([name]) => name === "linear_write")?.[1]; + const linearRead = registries.flatMap((r) => Object.entries(r.tools)).find(([name]) => name === "linear_read")?.[1]; + expect(linearWrite).toBeDefined(); + expect(linearRead).toBeDefined(); + + const write = await linearWrite!.run!({ query: "mutation { issueCreate }" }); + const read = await linearRead!.run!({ query: "query { issues }" }); + expect(write.success).toBe(true); + expect(read.success).toBe(false); + expect(captured.map((c) => c.detail["tool"])).toEqual(["linear_write", "linear_read"]); + }); +}); From dc88d1552f79d0b92868ecd81c4e2496d568e65a Mon Sep 17 00:00:00 2001 From: Noah Lindner Date: Thu, 23 Jul 2026 11:15:25 -0400 Subject: [PATCH 5/5] duplicate-mutation guard + replay reads run real; soul stays lean MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round-1 replay findings, fixed mechanically (no prompt additions — reverted the soul lines from the first attempt): - toolset: an identical repeated outward call within one wake is refused after a success ('already done'). Shared across §14.2 retry attempts on purpose: external calls record no ledger effects, so a wake that wrote then died would otherwise re-run the write on retry. Failures never arm the guard. - replay: read-grain tools run their REAL implementations (side-effect-free by the grain contract); only writes are stubbed+captured. A sandbox where she cannot look anything up distorts her more than reads answering with today's world — round 1's failed lookups produced the duplicate ticket and the fabricated 'I checked'. Co-Authored-By: Claude Fable 5 --- src/replay/run.ts | 14 +++++---- src/turn-runner/toolset.ts | 14 +++++++++ test/replay.test.ts | 5 ++-- test/toolset.test.ts | 60 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 86 insertions(+), 7 deletions(-) diff --git a/src/replay/run.ts b/src/replay/run.ts index ab27208..edb30d3 100644 --- a/src/replay/run.ts +++ b/src/replay/run.ts @@ -76,9 +76,12 @@ class CaptureAdapter implements SurfaceAdapter { async setTypingStatus(): Promise {} } -// The integration registries with their real specs but recording implementations: a write -// (any action-classed call) reports success without executing; a read reports itself -// unavailable — both are captured so the report shows what she reached for. +// The integration registries with writes stubbed and reads real. A write (any action-classed +// call) is captured and reports success without executing; a read runs its actual +// implementation — the grain contract already guarantees reads are side-effect-free, and a +// replay where she cannot look anything up distorts her far more than reads answering with +// today's world instead of the incident's (first run: failed lookups produced a duplicate +// ticket and a fabricated "I checked"). export function recordingRegistries(captured: CapturedAction[], clock: Clock): ToolRegistry[] { return INTEGRATION_REGISTRIES.map((r) => ({ ...r, @@ -88,9 +91,10 @@ export function recordingRegistries(captured: CapturedAction[], clock: Clock): T { ...spec, run: async (args: unknown) => { - captured.push({ at: clock(), kind: "external_tool", detail: { tool: name, args } }); const outward = (spec.actionClasses?.(args) ?? []).length > 0; - return outward ? { success: true, output: "done" } : { success: false, output: "that lookup is not available right now" }; + if (!outward) return spec.run ? spec.run(args) : { success: false, output: "that lookup is not available right now" }; + captured.push({ at: clock(), kind: "external_tool", detail: { tool: name, args } }); + return { success: true, output: JSON.stringify({ success: true, note: "the write completed" }) }; }, }, ]), diff --git a/src/turn-runner/toolset.ts b/src/turn-runner/toolset.ts index a506d54..092ac90 100644 --- a/src/turn-runner/toolset.ts +++ b/src/turn-runner/toolset.ts @@ -598,6 +598,11 @@ const BUILTIN_TOOL_NAME = new Set(BUILTIN_REGISTRIES.flatMap((r) => Object.keys( function externalTools(ctx: ToolsetContext): DynamicTool[] { const tools: DynamicTool[] = []; + // No turn needs the same mutation twice: an identical repeated outward call is a blind retry + // (2026-07-23 replay: a failed verification read led straight to a duplicate ticket). The set + // is shared by a wake's §14.2 retry attempts on purpose — external calls record no ledger + // effects, so without it a wake that wrote and then died would re-run the write on retry. + const ranOutward = new Set(); for (const grant of ctx.identity.grants) { if (BUILTIN_TOOL_NAME.has(grant.tool)) continue; // built-ins (audit_query included) are constructed below, not granted specs const spec = ctx.catalog[grant.tool]; @@ -610,6 +615,15 @@ function externalTools(ctx: ToolsetContext): DynamicTool[] { run: gated(ctx, grant.tool, async (args) => { const impl = spec?.run; if (!impl) return { success: false, output: `no implementation registered for external tool ${grant.tool}` }; + if ((spec?.actionClasses?.(args) ?? []).length > 0) { + const key = `${grant.tool}${JSON.stringify(args)}`; + if (ranOutward.has(key)) { + return { success: false, output: "already done: this exact call ran earlier this turn and completed. If you meant a different change, change the arguments." }; + } + const result = await impl(args); + if (result.success) ranOutward.add(key); + return result; + } return impl(args); }), }); diff --git a/test/replay.test.ts b/test/replay.test.ts index 927bfad..08a821c 100644 --- a/test/replay.test.ts +++ b/test/replay.test.ts @@ -177,7 +177,7 @@ describe("replay: reliving", () => { expect(prompts.some((l) => l.includes("what broke?"))).toBe(true); // the run narrates each replayed line }); - test("recording registries: a write reports done without executing, a read reports unavailable — both captured", async () => { + test("recording registries: a write reports done without executing and is captured; a read runs its real implementation", async () => { const captured: Parameters[0] = []; const registries = recordingRegistries(captured, fakeClock()); const linearWrite = registries.flatMap((r) => Object.entries(r.tools)).find(([name]) => name === "linear_write")?.[1]; @@ -186,9 +186,10 @@ describe("replay: reliving", () => { expect(linearRead).toBeDefined(); const write = await linearWrite!.run!({ query: "mutation { issueCreate }" }); + // the real read runs (here it fails friendly on missing credentials — same as live without keys) const read = await linearRead!.run!({ query: "query { issues }" }); expect(write.success).toBe(true); expect(read.success).toBe(false); - expect(captured.map((c) => c.detail["tool"])).toEqual(["linear_write", "linear_read"]); + expect(captured.map((c) => c.detail["tool"])).toEqual(["linear_write"]); // only the write is stub-captured }); }); diff --git a/test/toolset.test.ts b/test/toolset.test.ts index af98f77..14bc54b 100644 --- a/test/toolset.test.ts +++ b/test/toolset.test.ts @@ -652,3 +652,63 @@ describe("per-kind tool exposure", () => { for (const gone of ["task_create", "task_steer", "task_cancel", "task_confirm"]) expect(n).not.toContain(gone); }); }); + +describe("duplicate outward calls (one wake, one write)", () => { + test("an identical repeated outward call is refused; changed arguments and reads pass", async () => { + const db = freshDb(); + const clock = fakeClock(); + seedEvent(db, "e1", clock); + let writes = 0; + let reads = 0; + const catalog: ToolCatalog = { + fake_write: { + description: "w", + inputSchema: { type: "object" }, + actionClasses: () => ["outward"], + run: async () => ({ success: true, output: `w${++writes}` }), + }, + fake_read: { + description: "r", + inputSchema: { type: "object" }, + actionClasses: () => [], + run: async () => ({ success: true, output: `r${++reads}` }), + }, + }; + const ctx = baseCtx(db, clock, { + identity: identity({ grants: [{ tool: "fake_write", preauthorizedActionClasses: ["outward"] }, { tool: "fake_read", preauthorizedActionClasses: [] }] }), + catalog, + }); + const tools = buildToolset(ctx); + + expect((await tool(tools, "fake_write").run({ title: "ticket A" })).success).toBe(true); + const repeat = await tool(tools, "fake_write").run({ title: "ticket A" }); + expect(repeat.success).toBe(false); + expect(repeat.output).toContain("already done"); + expect(writes).toBe(1); // the second identical mutation never reached the implementation + expect((await tool(tools, "fake_write").run({ title: "ticket B" })).success).toBe(true); + expect((await tool(tools, "fake_read").run({ q: "same" })).success).toBe(true); + expect((await tool(tools, "fake_read").run({ q: "same" })).success).toBe(true); + expect(reads).toBe(2); // reads repeat freely + }); + + test("a FAILED outward call may be retried with the same arguments", async () => { + const db = freshDb(); + const clock = fakeClock(); + seedEvent(db, "e1", clock); + let calls = 0; + const catalog: ToolCatalog = { + fake_write: { + description: "w", + inputSchema: { type: "object" }, + actionClasses: () => ["outward"], + run: async () => (++calls === 1 ? { success: false, output: "transient" } : { success: true, output: "ok" }), + }, + }; + const ctx = baseCtx(db, clock, { identity: identity({ grants: [{ tool: "fake_write", preauthorizedActionClasses: ["outward"] }] }), catalog }); + const tools = buildToolset(ctx); + + expect((await tool(tools, "fake_write").run({ x: 1 })).success).toBe(false); + expect((await tool(tools, "fake_write").run({ x: 1 })).success).toBe(true); // failure never arms the guard + expect(calls).toBe(2); + }); +});