fix(daemon): Retry a busy store instead of failing live sessions

.specs/features/store-busy-healing/spec.md

@@ -0,0 +1,34 @@
+# Store Busy Healing Specification
+
+## Problem Statement
+
+`attachDriverEvents` persistia cada evento num `BEGIN` deferred: `events.append` lê (dedupe por `source_key`, `MAX(sequence)`) antes do INSERT. Se outra conexão commita nesse intervalo, o upgrade leitura→escrita recebe `SQLITE_BUSY_SNAPSHOT` (errcode 517) na hora, sem `busy_timeout`. O `catch` tratava isso como falha do harness: gravava `session.failed` (`UNKNOWN`, `blame: harness`), marcava a linha `failed` e saía do loop, enquanto o harness seguia vivo e seus eventos deixavam de ser persistidos (sessão `df94`: tokens do segundo turno perdidos, `wait` com estado terminal falso).
+
+A causa do segundo escritor (N daemons) é tratada em `daemon-single-instance`. Esta spec torna o loop robusto a qualquer escritor concorrente (ex.: `codedeck usage backfill`, lock mantido > 5s).
+
+## Goals
+
+- [x] `BEGIN IMMEDIATE` no loop de eventos: o lock de escrita é pego antes da leitura, `busy_timeout` volta a valer
+- [x] `SQLITE_BUSY` (errcode base 5, incluindo 517) não encerra o loop nem marca a sessão: o mesmo evento é retentado com backoff (25ms dobrando, teto 2s) enquanto o daemon roda
+- [x] Erros de persistência não-busy mantêm o comportamento atual (`failed`)
+- [x] `classifyFailure("database is locked")` → `STORE_BUSY`, `blame: infra`, `retryable: true`
+- [ ] (próxima fase) Reconciliador: sessão `failed` com `STORE_BUSY`/`UNKNOWN` de lock e PID vivo com a mesma identidade é reatachada a partir dos offsets persistidos
+
+## Decisions
+
+| Decisão | Escolha | Racional |
+| ------- | ------- | -------- |
+| Retry sem limite | Sim, enquanto não `shuttingDown` | O harness continua vivo e a saída está no log em disco; desistir recria o bug. Um lock eterno é problema do daemon, logado uma vez em `daemon.log` |
+| Idempotência do retry | Transação inteira refeita (evento + offsets + status) | ROLLBACK desfaz tudo; dedupe por `sourceKey` protege replays |
+| Shutdown durante o retry | Sai do loop sem gravar nada | O drain de shutdown é dono do desfecho (`interrupted`) |
+
+## Acceptance Criteria
+
+1. WHEN outro processo commita entre a leitura e a escrita da transação do evento THEN todos os eventos SHALL ser persistidos e a sessão SHALL terminar `completed`
+2. WHEN `append` lança `SQLITE_BUSY`/`BUSY_SNAPSHOT` THEN o daemon SHALL retentar o mesmo evento e nenhum `session.failed` SHALL ser gravado
+3. WHEN `append` lança um erro não-busy THEN a sessão SHALL ser marcada `failed` (sem regressão)
+4. WHEN o texto do erro é `database is locked` THEN `classifyFailure` SHALL retornar `STORE_BUSY` com `blame: infra`
+
+## Validation
+
+`tests/daemon-store-busy.test.ts` (AC 1-4). AC 1 reproduz a intercalação real com uma segunda `DatabaseSync` e falhava antes do fix com só `session.failed` no log de eventos.

src/core/errors.ts

@@ -84,6 +84,7 @@ export type FailureCode =
   | "SPAWN_FAILED" // could not start the harness binary
   | "TIMEOUT"
   | "SHUTDOWN" // daemon shut down gracefully; session left interrupted
+  | "STORE_BUSY" // the daemon's SQLite store was locked; the harness may still run
   | "UNKNOWN";
 
 export interface FailureInfo {
@@ -119,13 +120,31 @@ const SIGNAL_EXIT_CODES: Record<number, string> = {
   139: "SIGSEGV",
 };
 
+// SQLITE_BUSY (5) and its extended codes, e.g. SQLITE_BUSY_SNAPSHOT (517),
+// which a WAL read snapshot gets when another writer commits before it
+// upgrades to write. busy_timeout never retries that one.
+const SQLITE_BUSY = 5;
+const STORE_BUSY_PATTERN = /database is locked|SQLITE_BUSY|database is busy/i;
+
+export function isStoreBusy(error: unknown): boolean {
+  const errcode = (error as { errcode?: unknown } | null)?.errcode;
+  if (typeof errcode === "number") return (errcode & 0xff) === SQLITE_BUSY;
+  const message = error instanceof Error ? error.message : String(error);
+  return STORE_BUSY_PATTERN.test(message);
+}
+
 export function classifyFailure(
   text: string,
   exitCode?: number | null,
   signal?: string | null,
 ): FailureInfo {
   const clean = (text || "").trim();
   const detail = clean.slice(0, 300) || undefined;
+  // The daemon's own store, not the harness or the task: checked first so
+  // no harness signature claims it.
+  if (STORE_BUSY_PATTERN.test(clean)) {
+    return { code: "STORE_BUSY", blame: "infra", retryable: true, detail };
+  }
   for (const [pattern, reason] of HARNESS_CRASH_PATTERNS) {
     if (pattern.test(clean)) {
       return { code: "HARNESS_CRASH", blame: "harness", retryable: true, reason, detail };

src/daemon/daemon.ts

@@ -24,7 +24,7 @@ import { readSessionProcessMetadata } from "../drivers/session-runtime.js";
 import type { AgentEvent } from "../core/events.js";
 import { loadConfig, resolveDefaultSandbox } from "../config/config.js";
 import { invalidWebPortMessage, resolveWebPort } from "../config/web-port.js";
-import { classifyFailure, RunAgentError, type FailureInfo } from "../core/errors.js";
+import { classifyFailure, isStoreBusy, RunAgentError, type FailureInfo } from "../core/errors.js";
 import { parseRole } from "../core/roles.js";
 import { getCachedOrDiscoverModels, type HarnessModels } from "../core/models.js";
 import { aggregateRunUsage } from "../core/run-usage.js";
@@ -108,6 +108,11 @@ export interface DaemonOptions {
   spawnWebChild?: WebSupervisorOptions["spawnChild"];
 }
 
+// Backoff for a busy store in the event loop; capped, never given up while
+// the daemon runs (see persistDriverEvent).
+const STORE_BUSY_RETRY_BASE_MS = 25;
+const STORE_BUSY_RETRY_MAX_MS = 2000;
+
 function appendDaemonLog(line: string): void {
   try {
     fs.appendFileSync(getPaths().daemonLog, `[${new Date().toISOString()}] ${line}\n`);
@@ -1662,19 +1667,21 @@ class Daemon {
     }
   }
 
-  private async attachDriverEvents(sessionId: string, driver: AgentDriver, drvSession: DriverSession): Promise<void> {
-    try {
-      for await (const ev of driver.events(drvSession)) {
-        // A stop operation owns the terminal outcome. A terminal frame already
-        // buffered by the harness must not race it into the event log.
-        if (this.sessionLocks.has(sessionId) && (ev.type === "session.completed" || ev.type === "session.failed")) {
-          continue;
-        }
-        const db = this.db.getHandle();
-        db.exec("BEGIN");
-        let inserted = 0;
+  // Commits one event with its cursor. A busy store is the daemon's problem,
+  // not the session's: the harness keeps running and its output stays in the
+  // log file, so retry the same event until the store frees up instead of
+  // failing a live session. Returns undefined when shutdown interrupts the
+  // wait (the drain owns the outcome); any other error propagates.
+  private async persistDriverEvent(sessionId: string, driver: AgentDriver, ev: AgentEvent): Promise<number | undefined> {
+    const db = this.db.getHandle();
+    for (let attempt = 0; ; attempt += 1) {
+      try {
+        // IMMEDIATE takes the write lock up front, so busy_timeout applies.
+        // A deferred BEGIN read first and then got SQLITE_BUSY_SNAPSHOT on
+        // the upgrade, which no busy handler retries.
+        db.exec("BEGIN IMMEDIATE");
         try {
-          inserted = this.events.append(sessionId, ev, ev.raw);
+          const inserted = this.events.append(sessionId, ev, ev.raw);
           const offsets = driver.getOffsets?.(sessionId);
           if (offsets) {
             this.sessions.update(sessionId, { logOffset: offsets.log, stderrOffset: offsets.stderr });
@@ -1684,10 +1691,30 @@ class Daemon {
             this.updateSessionFromEvent(sessionId, ev, inserted);
           }
           db.exec("COMMIT");
+          return inserted;
         } catch (error) {
           try { db.exec("ROLLBACK"); } catch {}
           throw error;
         }
+      } catch (error) {
+        if (!isStoreBusy(error)) throw error;
+        if (this.shuttingDown) return undefined;
+        if (attempt === 0) appendDaemonLog(`store busy persisting ${ev.type} for ${sessionId}; retrying`);
+        await sleep(Math.min(STORE_BUSY_RETRY_BASE_MS * 2 ** attempt, STORE_BUSY_RETRY_MAX_MS));
+      }
+    }
+  }
+
+  private async attachDriverEvents(sessionId: string, driver: AgentDriver, drvSession: DriverSession): Promise<void> {
+    try {
+      for await (const ev of driver.events(drvSession)) {
+        // A stop operation owns the terminal outcome. A terminal frame already
+        // buffered by the harness must not race it into the event log.
+        if (this.sessionLocks.has(sessionId) && (ev.type === "session.completed" || ev.type === "session.failed")) {
+          continue;
+        }
+        const inserted = await this.persistDriverEvent(sessionId, driver, ev);
+        if (inserted === undefined) return;
         // Broadcast only after the durable event+cursor commit, and never
         // broadcast a replay that was already in the event store.
         if (inserted !== 0) this.broadcast(sessionId, ev);

tests/daemon-store-busy.test.ts

@@ -0,0 +1,138 @@
+import path from "node:path";
+import { DatabaseSync } from "node:sqlite";
+import { afterEach, beforeEach, describe, expect, it } from "vitest";
+import { Daemon } from "../src/daemon/daemon.js";
+import { classifyFailure } from "../src/core/errors.js";
+import type { AgentEvent } from "../src/core/events.js";
+import { makeTempDir, removeTempDir, seam, seed } from "./helpers/daemon-seam.js";
+
+// A persistence error is the daemon's problem, not the session's: while the
+// harness runs, a busy SQLite must never end the event loop or mark the
+// session failed (the "database is locked" df94 incident).
+
+let dir: string;
+let daemon: Daemon | undefined;
+
+beforeEach(() => {
+  dir = makeTempDir("store-busy-");
+  process.env.RUN_AGENT_DIR = dir;
+});
+
+afterEach(() => {
+  try { if (daemon) seam(daemon).db.close(); } catch {}
+  daemon = undefined;
+  delete process.env.RUN_AGENT_DIR;
+  removeTempDir(dir);
+});
+
+function turn(sessionId: string): AgentEvent[] {
+  const ts = () => new Date().toISOString();
+  return [
+    { type: "message", sessionId, timestamp: ts(), content: "working on it", sourceKey: "log:10:0" },
+    { type: "usage.updated", sessionId, timestamp: ts(), usage: { inputTokens: 1200, outputTokens: 80 }, sourceKey: "log:20:0" },
+    { type: "session.completed", sessionId, timestamp: ts(), reason: "completed", sourceKey: "log:30:0" },
+  ] as AgentEvent[];
+}
+
+function driverFor(events: AgentEvent[]) {
+  return {
+    async *events() {
+      for (const ev of events) yield ev;
+    },
+  };
+}
+
+function busyError(errcode: number): Error {
+  return Object.assign(new Error("database is locked"), { code: "ERR_SQLITE_ERROR", errcode });
+}
+
+async function run(sessionId: string, events: AgentEvent[]): Promise<void> {
+  await (daemon as any).attachDriverEvents(sessionId, driverFor(events), { id: sessionId });
+}
+
+describe("event loop under a busy store", () => {
+  it("survives another process committing mid-transaction (SQLITE_BUSY_SNAPSHOT)", async () => {
+    daemon = new Daemon();
+    seed(daemon, "s-snap");
+    const store = seam(daemon).events;
+    const handle = seam(daemon).db.getHandle();
+    // A second writer on the same file: what a stray daemon was.
+    const other = new DatabaseSync(path.join(dir, "run-agent.db"));
+    other.exec("PRAGMA busy_timeout = 0; CREATE TABLE other_writer (n INTEGER);");
+    const original = store.append.bind(store);
+    let interleaved = false;
+    store.append = (sessionId, event, raw) => {
+      if (!interleaved && event.type === "message") {
+        interleaved = true;
+        // Open this transaction's read snapshot, then let the other writer
+        // commit before our INSERT, exactly the production interleaving.
+        handle.prepare("SELECT COUNT(*) FROM events").get();
+        try {
+          other.exec("INSERT INTO other_writer VALUES (1)");
+        } catch (error) {
+          // An IMMEDIATE transaction already holds the write lock, so the
+          // other writer is the one refused. That is the fixed behaviour.
+          if (!/database is locked/.test((error as Error).message)) throw error;
+        }
+      }
+      return original(sessionId, event, raw);
+    };
+
+    try {
+      await run("s-snap", turn("s-snap"));
+    } finally {
+      other.close();
+    }
+
+    const types = store.list("s-snap").map((e) => e.type);
+    expect(types).toEqual(["message", "usage.updated", "session.completed"]);
+    expect(seam(daemon).sessions.get("s-snap")?.status).toBe("completed");
+  });
+
+  it("retries a busy append instead of failing the session", async () => {
+    daemon = new Daemon();
+    seed(daemon, "s-busy");
+    const store = seam(daemon).events;
+    const original = store.append.bind(store);
+    let failures = 0;
+    store.append = (sessionId, event, raw) => {
+      // Busy for longer than busy_timeout covers: plain SQLITE_BUSY, then
+      // BUSY_SNAPSHOT, then the store frees up.
+      if (event.type === "usage.updated" && failures < 2) {
+        failures += 1;
+        throw busyError(failures === 1 ? 5 : 517);
+      }
+      return original(sessionId, event, raw);
+    };
+
+    await run("s-busy", turn("s-busy"));
+
+    expect(failures).toBe(2);
+    const events = store.list("s-busy");
+    expect(events.map((e) => e.type)).toEqual(["message", "usage.updated", "session.completed"]);
+    expect(events.some((e) => e.type === "session.failed")).toBe(false);
+    expect(seam(daemon).sessions.get("s-busy")?.status).toBe("completed");
+  });
+
+  it("still fails the session on a non-busy persistence error", async () => {
+    daemon = new Daemon();
+    seed(daemon, "s-bad");
+    const store = seam(daemon).events;
+    const original = store.append.bind(store);
+    store.append = (sessionId, event, raw) => {
+      if (event.type === "usage.updated") throw new Error("disk I/O error");
+      return original(sessionId, event, raw);
+    };
+
+    await run("s-bad", turn("s-bad"));
+
+    expect(seam(daemon).sessions.get("s-bad")?.status).toBe("failed");
+  });
+});
+
+describe("classifyFailure for store errors", () => {
+  it("blames infra, not the harness, for a locked database", () => {
+    expect(classifyFailure("database is locked")).toMatchObject({ code: "STORE_BUSY", blame: "infra", retryable: true });
+    expect(classifyFailure("SQLITE_BUSY: database is busy")).toMatchObject({ code: "STORE_BUSY", blame: "infra" });
+  });
+});