Age out stale running sessions and recover leftover run state

Author: TheGreatAxios

CHANGELOG.md

@@ -13,6 +13,14 @@ parallel copies under `docs/` or `scripts/notes/`. At cut time: rename
 
 ## [Unreleased]
 
+### Fixed
+
+- Stale `running` sessions age to `interrupted` after two missed 5-minute
+  heartbeats, leftover newer parseable `run.json.*.tmp` files recover by mtime
+  over a stale `run.json`, and resume persists `interrupted` before reopening
+  as `running`. Signals stay `failed`; missing or unreadable state stays
+  `crashed`.
+
 ## [0.3.20] - 2026-09-10
 
 ### Added

docs/TUI.md

@@ -540,7 +540,7 @@ in `mouse-reporting-disabled.test.ts` for both `runListModal` and
 click-to-expand or drag-to-scroll, so leaving mouse reporting off lets the
 terminal's own text selection and copy work by default, with no Alt+M dance
 required. The resume picker lists the 10 most recently persisted sessions
-for this checkout — completed, failed, and crashed included. Recency is
+for this checkout — completed, failed, crashed, and interrupted included. Recency is
 the last write to `run.json`, not start time. Type to filter by name
 (printable keys claim the `>` row, same as the model picker); `--force`
 is not a list filter.

src/exec/runner.ts

@@ -75,6 +75,7 @@ import {
   syncRunStateHandle,
   type RunStateHandle,
 } from "../session/active-run.js";
+import { startRunHeartbeat } from "../session/run-liveness.js";
 import {
   setActiveDisposeHost,
   clearActiveDisposeHost,
@@ -389,13 +390,18 @@ export async function runExec(config: Config): Promise<ExecResult> {
     turnsUsed: 0,
     model: `${config.providerName}:${config.model}`,
   };
+  let stopHeartbeat: (() => void) | undefined;
 
   const persist = async (
     status: "running" | "done" | "failed" | "cancelled",
     extra?: { error?: string },
   ): Promise<void> => {
     if (finalized && status === "running") return;
-    if (status !== "running") finalized = true;
+    if (status !== "running") {
+      finalized = true;
+      stopHeartbeat?.();
+      stopHeartbeat = undefined;
+    }
     const model = `${config.providerName}:${config.model}`;
     const nextTurnsUsed = runSink?.getTurnCount() ?? turnsUsed;
     syncRunStateHandle(activeRunHandle, {
@@ -437,6 +443,10 @@ export async function runExec(config: Config): Promise<ExecResult> {
   };
 
   await persist("running");
+  stopHeartbeat = startRunHeartbeat({
+    shouldTick: () => !finalized,
+    tick: () => persist("running"),
+  });
 
   try {
     // Pricing seed is optional for exec; continue without rates rather than fail the run.

src/session/list-sessions.test.ts

@@ -275,3 +275,19 @@ test("listSessions reports updatedAt from run.json mtime", async () => {
   expect(row?.updatedAt).toBeGreaterThanOrEqual(stamp - 2000);
   expect(row?.updatedAt).toBeLessThanOrEqual(stamp + 2000);
 });
+
+test("listSessions ages a stale running session to interrupted", async () => {
+  const sessionId = generateSessionId();
+  const runPath = await writeRun(sessionId, {
+    status: "running",
+    task: "stale live",
+    startedAt: 1,
+  });
+  const oldSec = Math.floor(Date.now() / 1000) - 20 * 60;
+  await utimes(runPath, oldSec, oldSec);
+
+  const listed = await listSessions(cwd, home);
+  const row = listed.find((s) => s.sessionId === sessionId);
+  expect(row?.status).toBe("interrupted");
+  expect(row?.status).not.toBe("running");
+});

src/session/run-liveness.test.ts

@@ -0,0 +1,311 @@
+import { mkdtemp, rm, utimes, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+
+import { afterEach, beforeEach, describe, expect, test } from "bun:test";
+
+import {
+  clearActiveRun,
+  setActiveRun,
+  type RunStateHandle,
+} from "./active-run.js";
+import { generateSessionId, initSessionDir, sessionDir } from "./index.js";
+import {
+  ageStaleRunningState,
+  RUN_STALE_THRESHOLD_MS,
+  startRunHeartbeat,
+} from "./run-liveness.js";
+import { loadState, saveState, type RunState } from "./state.js";
+
+function baseState(over: Partial<RunState> = {}): RunState {
+  return {
+    status: "running",
+    turnsUsed: 2,
+    task: "liveness",
+    startedAt: 1_000,
+    model: "test:model",
+    ...over,
+  };
+}
+
+describe("ageStaleRunningState", () => {
+  test("ages parseable stale running to interrupted", () => {
+    const aged = ageStaleRunningState(baseState(), 1_000, {
+      nowMs: 1_000 + RUN_STALE_THRESHOLD_MS + 1,
+      sessionId: "other",
+    });
+    expect(aged.status).toBe("interrupted");
+    expect(aged.finishedAt).toBe(1_000);
+  });
+
+  test("does not age a fresh running record", () => {
+    const aged = ageStaleRunningState(baseState(), 1_000, {
+      nowMs: 1_000 + RUN_STALE_THRESHOLD_MS,
+    });
+    expect(aged.status).toBe("running");
+  });
+
+  test("does not age the active run owned by this process", () => {
+    clearActiveRun();
+    const sessionId = "live-session";
+    const handle: RunStateHandle = {
+      sessionId,
+      cwd: "/tmp",
+      task: "live",
+      startedAt: 1,
+    };
+    setActiveRun(handle);
+    try {
+      const aged = ageStaleRunningState(baseState(), 1_000, {
+        nowMs: 1_000 + RUN_STALE_THRESHOLD_MS + 1,
+        sessionId,
+      });
+      expect(aged.status).toBe("running");
+    } finally {
+      clearActiveRun();
+    }
+  });
+});
+
+describe("startRunHeartbeat", () => {
+  test("ticks while active and stops after teardown", async () => {
+    let ticks = 0;
+    let active = true;
+    const stop = startRunHeartbeat({
+      intervalMs: 20,
+      shouldTick: () => active,
+      tick: () => {
+        ticks += 1;
+      },
+    });
+    await new Promise((resolve) => setTimeout(resolve, 55));
+    expect(ticks).toBeGreaterThan(0);
+    const beforeStop = ticks;
+    active = false;
+    stop();
+    await new Promise((resolve) => setTimeout(resolve, 45));
+    expect(ticks).toBe(beforeStop);
+  });
+});
+
+describe("loadState tmp recovery and stale aging", () => {
+  let cwd = "";
+  let home = "";
+
+  beforeEach(async () => {
+    const stamp = `${Date.now()}-${Math.random().toString(16).slice(2)}`;
+    cwd = await mkdtemp(join(tmpdir(), `corbits-liveness-cwd-${stamp}-`));
+    home = await mkdtemp(join(tmpdir(), `corbits-liveness-home-${stamp}-`));
+  });
+
+  afterEach(async () => {
+    await rm(cwd, { recursive: true, force: true });
+    await rm(home, { recursive: true, force: true });
+  });
+
+  test("prefers a newer parseable tmp over stale run.json by mtime", async () => {
+    const sessionId = generateSessionId();
+    await initSessionDir(cwd, sessionId, home);
+    const dir = sessionDir(cwd, sessionId, home);
+    const runPath = join(dir, "run.json");
+    await saveState(cwd, sessionId, baseState({ task: "old-on-disk" }), home);
+    const newer = baseState({ task: "newer-tmp", turnsUsed: 9 });
+    const tmpPath = join(dir, `run.json.${process.pid}.recovery.tmp`);
+    await writeFile(tmpPath, JSON.stringify(newer, null, 2));
+    const older = Math.floor(Date.now() / 1000) - 20 * 60;
+    const newerSec = Math.floor(Date.now() / 1000);
+    await utimes(runPath, older, older);
+    await utimes(tmpPath, newerSec, newerSec);
+
+    const loaded = await loadState(cwd, sessionId, home, {
+      persistAgeOut: false,
+      staleThresholdMs: 10 * 60_000,
+    });
+    expect(loaded).toMatchObject({
+      kind: "ok",
+      state: { task: "newer-tmp", turnsUsed: 9 },
+    });
+  });
+
+  test("does not prefer a newer parseable tmp over a fresh run.json", async () => {
+    const sessionId = generateSessionId();
+    await initSessionDir(cwd, sessionId, home);
+    const dir = sessionDir(cwd, sessionId, home);
+    await saveState(cwd, sessionId, baseState({ task: "canonical" }), home);
+    const tmpPath = join(dir, `run.json.${process.pid}.inflight.tmp`);
+    await writeFile(
+      tmpPath,
+      JSON.stringify(baseState({ task: "in-flight", turnsUsed: 9 }), null, 2),
+    );
+    const runPath = join(dir, "run.json");
+    const older = Math.floor(Date.now() / 1000) - 30;
+    const newerSec = Math.floor(Date.now() / 1000);
+    await utimes(runPath, older, older);
+    await utimes(tmpPath, newerSec, newerSec);
+
+    const loaded = await loadState(cwd, sessionId, home, {
+      persistAgeOut: false,
+    });
+    expect(loaded).toMatchObject({
+      kind: "ok",
+      state: { task: "canonical" },
+    });
+  });
+
+  test("does not resurrect a newer running tmp over a fresh terminal run.json", async () => {
+    const sessionId = generateSessionId();
+    await initSessionDir(cwd, sessionId, home);
+    const dir = sessionDir(cwd, sessionId, home);
+    await saveState(
+      cwd,
+      sessionId,
+      baseState({ status: "done", finishedAt: 999, task: "landed" }),
+      home,
+    );
+    const tmpPath = join(dir, `run.json.${process.pid}.straggler.tmp`);
+    await writeFile(
+      tmpPath,
+      JSON.stringify(baseState({ task: "straggler-running" }), null, 2),
+    );
+    const runPath = join(dir, "run.json");
+    const landed = Math.floor(Date.now() / 1000);
+    await utimes(runPath, landed, landed);
+    await utimes(tmpPath, landed + 2, landed + 2);
+
+    const loaded = await loadState(cwd, sessionId, home, {
+      persistAgeOut: false,
+    });
+    expect(loaded).toMatchObject({
+      kind: "ok",
+      state: { status: "done", task: "landed" },
+    });
+  });
+
+  test("recovers a parseable tmp when run.json is missing", async () => {
+    const sessionId = generateSessionId();
+    await initSessionDir(cwd, sessionId, home);
+    const dir = sessionDir(cwd, sessionId, home);
+    const tmpPath = join(dir, `run.json.${process.pid}.orphan.tmp`);
+    await writeFile(
+      tmpPath,
+      JSON.stringify(baseState({ task: "orphan-tmp", turnsUsed: 4 }), null, 2),
+    );
+
+    const loaded = await loadState(cwd, sessionId, home, {
+      persistAgeOut: false,
+    });
+    expect(loaded).toMatchObject({
+      kind: "ok",
+      state: { task: "orphan-tmp", turnsUsed: 4 },
+    });
+  });
+
+  test("does not resurrect an older or equal-mtime tmp", async () => {
+    const sessionId = generateSessionId();
+    await initSessionDir(cwd, sessionId, home);
+    const dir = sessionDir(cwd, sessionId, home);
+    await saveState(cwd, sessionId, baseState({ task: "canonical" }), home);
+    const tmpPath = join(dir, `run.json.${process.pid}.older.tmp`);
+    await writeFile(
+      tmpPath,
+      JSON.stringify(baseState({ task: "stale-tmp" }), null, 2),
+    );
+    const runPath = join(dir, "run.json");
+    const stamp = Math.floor(Date.now() / 1000);
+    await utimes(runPath, stamp, stamp);
+    await utimes(tmpPath, stamp - 30, stamp - 30);
+
+    const loaded = await loadState(cwd, sessionId, home, {
+      persistAgeOut: false,
+    });
+    expect(loaded).toMatchObject({
+      kind: "ok",
+      state: { task: "canonical" },
+    });
+  });
+
+  test("does not prefer a malformed newer tmp", async () => {
+    const sessionId = generateSessionId();
+    await initSessionDir(cwd, sessionId, home);
+    const dir = sessionDir(cwd, sessionId, home);
+    await saveState(cwd, sessionId, baseState({ task: "canonical" }), home);
+    const tmpPath = join(dir, `run.json.${process.pid}.bad.tmp`);
+    await writeFile(tmpPath, "{ not-json");
+    const runPath = join(dir, "run.json");
+    const older = Math.floor(Date.now() / 1000) - 60;
+    const newerSec = Math.floor(Date.now() / 1000);
+    await utimes(runPath, older, older);
+    await utimes(tmpPath, newerSec, newerSec);
+
+    const loaded = await loadState(cwd, sessionId, home, {
+      persistAgeOut: false,
+    });
+    expect(loaded).toMatchObject({
+      kind: "ok",
+      state: { task: "canonical" },
+    });
+  });
+
+  test("sweeps aged temps but keeps a fresh in-flight tmp", async () => {
+    const sessionId = generateSessionId();
+    await initSessionDir(cwd, sessionId, home);
+    const dir = sessionDir(cwd, sessionId, home);
+    await saveState(cwd, sessionId, baseState(), home);
+    const agedTmp = join(dir, `run.json.${process.pid}.aged.tmp`);
+    const freshTmp = join(dir, `run.json.${process.pid}.fresh.tmp`);
+    await writeFile(agedTmp, "{");
+    await writeFile(freshTmp, "{");
+    const agedSec = Math.floor(Date.now() / 1000) - 20 * 60;
+    const freshSec = Math.floor(Date.now() / 1000);
+    await utimes(agedTmp, agedSec, agedSec);
+    await utimes(freshTmp, freshSec, freshSec);
+
+    await loadState(cwd, sessionId, home, {
+      nowMs: Date.now(),
+      tmpSweepAgeMs: 10 * 60_000,
+      persistAgeOut: false,
+    });
+
+    const { existsSync } = await import("node:fs");
+    expect(existsSync(agedTmp)).toBe(false);
+    expect(existsSync(freshTmp)).toBe(true);
+  });
+
+  test("ages stale running to interrupted and persists it", async () => {
+    const sessionId = generateSessionId();
+    await initSessionDir(cwd, sessionId, home);
+    await saveState(cwd, sessionId, baseState(), home);
+    const runPath = join(sessionDir(cwd, sessionId, home), "run.json");
+    const oldSec = Math.floor(Date.now() / 1000) - 20 * 60;
+    await utimes(runPath, oldSec, oldSec);
+
+    const loaded = await loadState(cwd, sessionId, home, {
+      nowMs: Date.now(),
+      staleThresholdMs: 10 * 60_000,
+    });
+    expect(loaded).toMatchObject({
+      kind: "ok",
+      state: { status: "interrupted", turnsUsed: 2 },
+    });
+    const again = await loadState(cwd, sessionId, home, {
+      persistAgeOut: false,
+    });
+    expect(again).toMatchObject({
+      kind: "ok",
+      state: { status: "interrupted" },
+    });
+  });
+
+  test("missing and unreadable stay non-interrupted", async () => {
+    const missingId = generateSessionId();
+    expect(await loadState(cwd, missingId, home)).toEqual({ kind: "missing" });
+
+    const badId = generateSessionId();
+    await initSessionDir(cwd, badId, home);
+    await writeFile(
+      join(sessionDir(cwd, badId, home), "run.json"),
+      "{ turnsUsed",
+    );
+    expect(await loadState(cwd, badId, home)).toEqual({ kind: "unreadable" });
+  });
+});