diff --git a/.specs/features/store-busy-healing/spec.md b/.specs/features/store-busy-healing/spec.md index 3222ad4..94cc485 100644 --- a/.specs/features/store-busy-healing/spec.md +++ b/.specs/features/store-busy-healing/spec.md @@ -12,7 +12,10 @@ A causa do segundo escritor (N daemons) é tratada em `daemon-single-instance`. - [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 +- [x] Reconciliador no boot (`reviveStoreBusyFailures`, antes do loop de `recover()`): linha `failed` com `failure.code=STORE_BUSY` ou `failure.detail` contendo `database is locked`, com log em disco e PID, volta a `working` e passa pelo mesmo reattach de um restart + - harness vivo (mesma identidade de PID): segue tailando do offset persistido + - harness morto (ou PID reciclado): o log é drenado do offset e o desfecho é classificado; eventos drenados e `completedAt` recebem o mtime do log (os parsers carimbam hora de leitura) +- [x] Harness morto sem frame terminal e com exit não observado (reattach): se o último evento (fora `usage.updated`) é `turn.completed`, sem sinal fatal e sem assinatura de crash no stderr, fecha `completed` com `reason: "turn completed; exit not observed"`; senão segue `failed` "exited without reporting a terminal event" ## Decisions @@ -21,6 +24,10 @@ A causa do segundo escritor (N daemons) é tratada em `daemon-single-instance`. | 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`) | +| Reviver `failed` de lock | Só no boot, só com log + PID, nunca `origin=open` | Reusa o caminho de reattach testado; `open` não tem driver para reatachar | +| Horário do backlog drenado | mtime do log (máx. de stdout/stderr) quando o harness já morreu | Parsers usam `new Date()` na leitura; o mtime é o limite superior honesto. O uso por hora agrupa por `created_at`, então só a linha do tempo e `completedAt` mudam | +| Exit não observado | `turn.completed` final conta como fim do turno | `codex exec` sai logo após `turn.completed`; o trabalho do turno está feito. Vale também para restart comum do daemon. Limite: só enxerga eventos lidos desde o reattach | +| Idempotência | O desfecho novo não tem `database is locked` no `detail` | Um boot seguinte não revive a mesma linha de novo | ## Acceptance Criteria @@ -28,7 +35,11 @@ A causa do segundo escritor (N daemons) é tratada em `daemon-single-instance`. 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` +5. WHEN o boot encontra uma linha `failed` por lock com harness vivo THEN ela SHALL voltar a `working` e os eventos novos do log SHALL ser persistidos +6. WHEN o harness dessa linha já morreu THEN o log SHALL ser drenado do offset, a sessão SHALL terminar em estado terminal sem `database is locked`, e eventos drenados e `completedAt` SHALL usar o mtime do log +7. WHEN a falha não é de lock, ou não há log THEN a linha SHALL ficar intacta +8. WHEN um reattach vê o PID morrer e o último evento é `turn.completed` THEN a sessão SHALL fechar `completed`; WHEN o log para antes disso, ou há sinal fatal, exit não-zero ou crash no stderr THEN SHALL fechar `failed` ## 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. +`tests/daemon-store-busy.test.ts` (AC 1-4), `tests/daemon-heal-store-busy.test.ts` (AC 5-8), `tests/terminal-synth.test.ts` (AC 8). AC 1 reproduz a intercalação real com uma segunda `DatabaseSync` e falhava antes do fix com só `session.failed` no log de eventos. diff --git a/src/daemon/daemon.ts b/src/daemon/daemon.ts index ea309f7..766906e 100644 --- a/src/daemon/daemon.ts +++ b/src/daemon/daemon.ts @@ -132,6 +132,9 @@ class Daemon { private startTime = Date.now(); private startupReconcilePromise: Promise = Promise.resolve(); private sessionLocks = new Set(); + // Sessions recover() revived from a store-busy failure: the sequence the + // drain starts after, and the log mtime when the harness is already dead. + private healing = new Map(); // Power-shutdown state. `shuttingDown` is set synchronously by the signal // handler so concurrent handleRequest calls are refused during the drain. private shuttingDown = false; @@ -398,6 +401,7 @@ class Daemon { // streaming; a dead one is drained and classified from its log tail. private async recover(): Promise { const paths = getPaths(); + this.reviveStoreBusyFailures(paths.logsDir); const actives = this.sessions.listActive(); for (const s of actives) { // Power-shutdown rows are terminal: never reattach, never flip. A @@ -468,7 +472,10 @@ class Daemon { // Never attach to or signal a PID whose identity changed while the // daemon was away. The original harness is gone; the replacement is // somebody else's process. - if (pidReused) { + // A revived row died long ago: its PID may be anyone's by now, but its + // log is still its own, so drain it as dead instead of failing it. + const healing = this.healing.has(s.id); + if (pidReused && !healing) { const failure: FailureInfo = { code: "HARNESS_CRASH", blame: "harness", @@ -490,13 +497,14 @@ class Daemon { continue; } - const alive = processPresent; + // Only a healing row reaches here with a reused PID; it counts as dead. + const alive = processPresent && !pidReused; const stdoutPath = path.join(paths.logsDir, `${s.id}.ndjson`); const stderrPath = path.join(paths.logsDir, `${s.id}.stderr.log`); const hasDetachedLogs = fs.existsSync(stdoutPath) || fs.existsSync(stderrPath); // A live PID without a recorded identity is unsafe to attach: it may be // a recycled process. A dead PID is safe to drain from its own log. - const identityVerified = !processPresent || (recordedStart !== undefined && currentStart === recordedStart); + const identityVerified = !alive || (recordedStart !== undefined && currentStart === recordedStart); if (typeof driver.attach === "function" && pid != null && hasDetachedLogs && identityVerified) { try { @@ -543,6 +551,59 @@ class Daemon { } } + // A session the old event loop failed on "database is locked" was never + // over: the harness kept writing its log past the persisted offset. Flip + // such rows back to working so the reattach loop below drains the log + // (dead harness) or keeps tailing it (live one), exactly as after a restart. + private reviveStoreBusyFailures(logsDir: string): void { + for (const s of this.sessions.listStoreBusyFailures()) { + if (s.origin === "open") continue; + const logs = [path.join(logsDir, `${s.id}.ndjson`), path.join(logsDir, `${s.id}.stderr.log`)] + .filter((file) => fs.existsSync(file)); + const metadata = readSessionProcessMetadata(s.id); + const pid = metadata?.pid ?? s.pid; + if (logs.length === 0 || pid == null) continue; + const recordedStart = metadata?.pidStartTime ?? s.pidStartTime; + const alive = processAlive(pid) && (recordedStart === undefined || processStartTime(pid) === recordedStart); + // Parsers stamp events with read time. For a dead harness the last log + // write is the best bound on when the backlog really happened. + const endedAt = alive + ? undefined + : new Date(Math.max(...logs.map((file) => fs.statSync(file).mtimeMs))); + const last = this.db.getHandle().prepare( + `SELECT COALESCE(MAX(sequence), 0) AS seq FROM events WHERE session_id = ?`, + ).get(s.id) as { seq: number }; + this.healing.set(s.id, { fromSequence: last.seq, endedAt }); + this.sessions.setStatus(s.id, "working", { + failure: null, + completedAt: null, + lastEvent: "healing after a store-busy failure", + }); + appendDaemonLog(`healing ${s.id} from store-busy failure (${alive ? "live" : "dead"} harness)`); + } + } + + // Once a revived session's stream ends, pin the drained backlog and the + // terminal time to the log mtime instead of the replay time. + private finishHeal(sessionId: string): void { + const heal = this.healing.get(sessionId); + if (!heal) return; + this.healing.delete(sessionId); + if (!heal.endedAt) return; + const endedAt = heal.endedAt.toISOString(); + try { + this.db.getHandle().prepare(` + UPDATE events + SET timestamp = ?, normalized_payload = json_set(normalized_payload, '$.timestamp', ?) + WHERE session_id = ? AND sequence > ? AND timestamp > ? + `).run(endedAt, endedAt, sessionId, heal.fromSequence, endedAt); + const s = this.sessions.get(sessionId); + if (s && isTerminalStatus(s.status)) this.sessions.update(sessionId, { completedAt: heal.endedAt }); + } catch (error) { + appendDaemonLog(`heal timestamps for ${sessionId} failed: ${error instanceof Error ? error.message : String(error)}`); + } + } + private async reconcileOpenUsage(sessionId: string, nativeIds?: readonly string[]): Promise { const session = this.sessions.get(sessionId); if (!session || session.origin !== "open" || session.agent !== "claude") return; @@ -1769,6 +1830,7 @@ class Daemon { this.sessions.setStatus(sessionId, "failed", { lastEvent: error.slice(0, 200), failure }); } } + this.finishHeal(sessionId); // A message queued while the turn ran starts now as the next turn. // tryDispatch rechecks resumability under the lifecycle lock; a stale // or unresumable slot stays put for a manual send. diff --git a/src/drivers/session-runtime.ts b/src/drivers/session-runtime.ts index 5a57563..0c7a677 100644 --- a/src/drivers/session-runtime.ts +++ b/src/drivers/session-runtime.ts @@ -33,6 +33,8 @@ export interface RuntimeHooks { hasTerminal: boolean; hasMessage: boolean; stderr: string; + // Last non-usage event is turn.completed (see synthesizeTerminalEvent). + endedOnTurnCompleted: boolean; }) => AgentEvent[]; } @@ -333,6 +335,7 @@ export class SessionRuntime { const hasTerminal = this.buffer.some((e) => e.type === "session.completed" || e.type === "session.failed"); if (!hasTerminal && !this.stopRequested && !this.shutdownRequested) { const hasMessage = this.buffer.some((e) => e.type === "message" || e.type === "text.delta"); + const lastFrame = this.buffer.findLast((e) => e.type !== "usage.updated"); for (const ev of this.hooks.synthesizeTerminal({ sessionId: this.sessionId, exitCode, @@ -340,6 +343,7 @@ export class SessionRuntime { hasTerminal, hasMessage, stderr: this.stderrBuf, + endedOnTurnCompleted: lastFrame?.type === "turn.completed", })) { this.push(ev); } diff --git a/src/drivers/terminal.ts b/src/drivers/terminal.ts index d8991f3..53c5edd 100644 --- a/src/drivers/terminal.ts +++ b/src/drivers/terminal.ts @@ -15,9 +15,28 @@ export function synthesizeTerminalEvent(input: { hasTerminal: boolean; hasMessage: boolean; stderr: string; + // The last frame (usage aside) was turn.completed. + endedOnTurnCompleted?: boolean; }): AgentEvent | null { if (input.hasTerminal) return null; const ts = new Date().toISOString(); + // No exit code and no signal: the death was learned by polling the pid + // after a reattach. A harness that writes no terminal frame (codex) still + // said its turn finished; a crash signature in stderr overrides that. + if ( + input.exitCode === null && + !input.signal && + input.endedOnTurnCompleted && + classifyFailure(input.stderr).code !== "HARNESS_CRASH" + ) { + return { + type: "session.completed", + sessionId: input.sessionId, + timestamp: ts, + reason: "turn completed; exit not observed", + raw: { stderr: input.stderr.slice(0, 2000) }, + } as AgentEvent; + } if (input.exitCode === 0) { // Exit 0 with produced output is completion. Exit 0 with NO output but // stderr content is how "exit 0 anyway" crashes look — treat as failure. diff --git a/src/store/sessions.ts b/src/store/sessions.ts index 83ec2ec..de10979 100644 --- a/src/store/sessions.ts +++ b/src/store/sessions.ts @@ -223,6 +223,19 @@ export class SessionStore { return rows.map(rowToSession); } + // Rows the pre-STORE_BUSY event loop failed on a locked SQLite store + // ("database is locked" was classified UNKNOWN), plus STORE_BUSY rows. + listStoreBusyFailures(): Session[] { + const rows = this.db.prepare(` + SELECT * FROM sessions + WHERE status = 'failed' AND failure IS NOT NULL + AND (json_extract(failure, '$.code') = 'STORE_BUSY' + OR json_extract(failure, '$.detail') LIKE '%database is locked%') + ORDER BY updated_at DESC + `).all() as unknown as SessionRow[]; + return rows.map(rowToSession); + } + getByRunId(runId: string): Session[] { const rows = this.db.prepare( `SELECT * FROM sessions WHERE run_id = ? ORDER BY updated_at DESC`, diff --git a/tests/daemon-heal-store-busy.test.ts b/tests/daemon-heal-store-busy.test.ts new file mode 100644 index 0000000..79d68a5 --- /dev/null +++ b/tests/daemon-heal-store-busy.test.ts @@ -0,0 +1,174 @@ +import { spawn, spawnSync, type ChildProcess } from "node:child_process"; +import fs from "node:fs"; +import path from "node:path"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { Daemon } from "../src/daemon/daemon.js"; +import type { AgentEvent } from "../src/core/events.js"; +import type { FailureInfo } from "../src/core/errors.js"; +import { processStartTime } from "../src/utils/process.js"; +import { makeTempDir, removeTempDir, seam, seed } from "./helpers/daemon-seam.js"; + +// The df94 incident: the event loop died on "database is locked" and marked +// a live codex session failed. The harness kept writing its log, so every +// missing event is still on disk past the persisted offset. On boot the +// daemon heals such rows from the log instead of leaving them failed. + +let dir: string; +let daemon: Daemon | undefined; +let child: ChildProcess | undefined; + +const firstTurn = [ + { type: "thread.started", thread_id: "01a0d643-0000-7000-8000-000000000001" }, + { type: "turn.started" }, + { type: "item.completed", item: { id: "item_0", type: "agent_message", text: "first turn" } }, +]; +const secondTurn = [ + { type: "item.completed", item: { id: "item_1", type: "agent_message", text: "second turn" } }, + { type: "turn.completed", usage: { input_tokens: 5400, output_tokens: 320 } }, +]; +const ndjson = (lines: object[]) => lines.map((line) => JSON.stringify(line) + "\n").join(""); + +// What the pre-fix catch in attachDriverEvents wrote. +const lockedFailure: FailureInfo = { code: "UNKNOWN", blame: "harness", retryable: true, detail: "database is locked" }; + +beforeEach(() => { + dir = makeTempDir("heal-store-busy-"); + process.env.RUN_AGENT_DIR = dir; +}); + +afterEach(async () => { + child?.kill("SIGKILL"); + child = undefined; + // recover() starts each event loop without awaiting it, and the status + // flips to terminal before the loop's tail runs. Closing the store under + // a running loop is an unhandled rejection, so wait for every healed + // loop to reach finishHeal first. + if (daemon) { + const healing = (daemon as unknown as { healing: Map }).healing; + await vi.waitFor(() => expect(healing.size).toBe(0), { timeout: 8000 }); + } + try { if (daemon) seam(daemon).db.close(); } catch {} + daemon = undefined; + delete process.env.RUN_AGENT_DIR; + removeTempDir(dir); +}); + +function writeLog(id: string, content: string): string { + const logs = path.join(dir, "logs"); + fs.mkdirSync(logs, { recursive: true }); + const file = path.join(logs, `${id}.ndjson`); + fs.writeFileSync(file, content); + return file; +} + +function seedLockedFailure(id: string, pid: number, pidStartTime: string | undefined, failure: FailureInfo = lockedFailure): void { + seed(daemon!, id, "failed", { + agent: "codex", + nativeSessionId: firstTurn[0]!.thread_id as string, + pid, + pidStartTime, + logOffset: Buffer.byteLength(ndjson(firstTurn)), + stderrOffset: 0, + failure, + lastEvent: failure.detail, + completedAt: new Date("2026-09-25T02:08:21.000Z"), + }); + const events = seam(daemon!).events; + const ts = "2026-09-25T02:08:00.000Z"; + events.append(id, { type: "message", sessionId: id, timestamp: ts, content: "first turn", sourceKey: "log:0:0" } as AgentEvent); + events.append(id, { type: "session.failed", sessionId: id, timestamp: "2026-09-25T02:08:21.000Z", error: "database is locked", failure } as AgentEvent); +} + +function deadPid(): number { + return spawnSync("true").pid!; +} + +describe("healing sessions failed by a locked store", { timeout: 15000 }, () => { + it("drains a dead harness's log and closes the session at the log's mtime", async () => { + daemon = new Daemon(); + const log = writeLog("s-dead", ndjson([...firstTurn, ...secondTurn])); + const endedAt = new Date("2026-09-25T02:10:15.000Z"); + fs.utimesSync(log, endedAt, endedAt); + seedLockedFailure("s-dead", deadPid(), "boot-1"); + + await seam(daemon).recover(); + + // Codex writes no terminal frame and a reattach cannot see its exit + // code; the trailing turn.completed is what closes it as completed. + await vi.waitFor(() => expect(seam(daemon!).sessions.get("s-dead")?.status).toBe("completed"), { timeout: 8000 }); + const session = seam(daemon).sessions.get("s-dead")!; + expect(session.completedAt?.toISOString()).toBe(endedAt.toISOString()); + expect(session.failure).toBeUndefined(); + + const events = seam(daemon).events.list("s-dead"); + const healed = events.slice(events.findIndex((e) => e.type === "session.failed") + 1); + expect(healed.map((e) => e.type)).toEqual(expect.arrayContaining(["usage.updated", "turn.completed"])); + expect(healed.at(-1)?.type).toBe("session.completed"); + expect(healed.some((e) => e.type === "message" && (e as { content?: string }).content === "second turn")).toBe(true); + // Parsers stamp read time; a drained backlog cannot postdate the log. + for (const event of healed) expect(event.timestamp).toBe(endedAt.toISOString()); + }); + + it("reattaches a live harness and keeps it working", async () => { + daemon = new Daemon(); + writeLog("s-live", ndjson([...firstTurn, secondTurn[0]!])); + child = spawn("sleep", ["30"], { stdio: "ignore" }); + const pid = child.pid!; + await vi.waitFor(() => expect(processStartTime(pid)).toBeDefined()); + seedLockedFailure("s-live", pid, processStartTime(pid)); + + await seam(daemon).recover(); + + const session = seam(daemon).sessions.get("s-live")!; + expect(session.status).toBe("working"); + expect(session.completedAt).toBeUndefined(); + await vi.waitFor(() => { + const contents = seam(daemon!).events.list("s-live").map((e) => (e as { content?: string }).content); + expect(contents).toContain("second turn"); + }, { timeout: 8000 }); + }); + + it("heals rows already classified STORE_BUSY", async () => { + daemon = new Daemon(); + writeLog("s-code", ndjson([...firstTurn, ...secondTurn])); + seedLockedFailure("s-code", deadPid(), "boot-1", { code: "STORE_BUSY", blame: "infra", retryable: true, detail: "database is locked" }); + + await seam(daemon).recover(); + + await vi.waitFor(() => expect(seam(daemon!).sessions.get("s-code")?.status).toBe("completed"), { timeout: 8000 }); + }); + + it("leaves failures that are not a locked store alone", async () => { + daemon = new Daemon(); + writeLog("s-task", ndjson([...firstTurn, ...secondTurn])); + seedLockedFailure("s-task", deadPid(), "boot-1", { code: "TASK_ERROR", blame: "task", retryable: false, detail: "exit code 1" }); + + await seam(daemon).recover(); + await new Promise((resolve) => setTimeout(resolve, 300)); + + expect(seam(daemon).sessions.get("s-task")?.status).toBe("failed"); + expect(seam(daemon).events.last("s-task")?.type).toBe("session.failed"); + }); + + it("leaves a locked-store failure without a log to drain", async () => { + daemon = new Daemon(); + seedLockedFailure("s-nolog", deadPid(), "boot-1"); + + await seam(daemon).recover(); + + expect(seam(daemon).sessions.get("s-nolog")?.status).toBe("failed"); + }); +}); + +describe("healing a dead harness that never finished its turn", { timeout: 15000 }, () => { + it("closes failed when the log stops before turn.completed", async () => { + daemon = new Daemon(); + writeLog("s-cut", ndjson([...firstTurn, secondTurn[0]!])); + seedLockedFailure("s-cut", deadPid(), "boot-1"); + + await seam(daemon).recover(); + + await vi.waitFor(() => expect(seam(daemon!).sessions.get("s-cut")?.status).toBe("failed"), { timeout: 8000 }); + expect(seam(daemon).sessions.get("s-cut")?.lastEvent).toMatch(/exited without reporting a terminal event/); + }); +}); diff --git a/tests/terminal-synth.test.ts b/tests/terminal-synth.test.ts new file mode 100644 index 0000000..0a0bb6b --- /dev/null +++ b/tests/terminal-synth.test.ts @@ -0,0 +1,34 @@ +import { describe, expect, it } from "vitest"; +import { synthesizeTerminalEvent } from "../src/drivers/terminal.js"; + +// A reattached runtime learns of a death by polling the pid, so it never +// sees an exit code. A turn.completed as the last frame is the harness's +// own word that the turn finished; without it, unknown stays failed. +const base = { sessionId: "s1", harness: "codex", hasTerminal: false, hasMessage: true, stderr: "" }; + +describe("synthesizeTerminalEvent with an unobserved exit", () => { + it("completes when the log ended on turn.completed", () => { + const ev = synthesizeTerminalEvent({ ...base, exitCode: null, signal: null, endedOnTurnCompleted: true }); + expect(ev).toMatchObject({ type: "session.completed", reason: "turn completed; exit not observed" }); + }); + + it("stays failed when the log did not end on turn.completed", () => { + const ev = synthesizeTerminalEvent({ ...base, exitCode: null, signal: null, endedOnTurnCompleted: false }); + expect(ev).toMatchObject({ type: "session.failed", error: "codex exited without reporting a terminal event" }); + }); + + it("stays failed on a fatal signal even after turn.completed", () => { + const ev = synthesizeTerminalEvent({ ...base, exitCode: null, signal: "SIGSEGV", endedOnTurnCompleted: true }); + expect(ev?.type).toBe("session.failed"); + }); + + it("stays failed on an observed non-zero exit even after turn.completed", () => { + const ev = synthesizeTerminalEvent({ ...base, exitCode: 1, signal: null, endedOnTurnCompleted: true }); + expect(ev?.type).toBe("session.failed"); + }); + + it("stays failed when stderr carries a crash signature", () => { + const ev = synthesizeTerminalEvent({ ...base, exitCode: null, signal: null, endedOnTurnCompleted: true, stderr: "Unhandled promise rejection: EPIPE" }); + expect(ev).toMatchObject({ type: "session.failed", failure: { code: "HARNESS_CRASH" } }); + }); +});