§5.5 stale-reply withholding: thread-follow replies yield to a room that moved mid-turn

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[] {

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<void> {
     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<void> {
   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<typeof createLogger>) {
+  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 <snapshot.db> --from <iso> --to <iso> [--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<void> {
+  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<void> {
   const codexOk = await codexReady();
   console.log(`${codexOk ? "ok      " : "MISSING "}codex logged in`);
@@ -253,6 +341,8 @@ async function main(): Promise<void> {
       return cmdDoctor();
     case "status":
       return cmdStatus();
+    case "replay":
+      return cmdReplay();
     default:
       console.log(HELP);
   }

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();
+}