diff --git a/scripts/run-tests.mjs b/scripts/run-tests.mjs index b5545720e..061c1c207 100644 --- a/scripts/run-tests.mjs +++ b/scripts/run-tests.mjs @@ -581,6 +581,7 @@ export const SUITES = { "src/lib/server/familiar-self-reports.test.ts", "src/lib/server/milestones-ledger.test.ts", "src/lib/server/stuck-created-sweep.test.ts", + "src/lib/server/stale-running-sweep.test.ts", "src/components/thread-signal-card.test.ts", "src/components/thread-signals-section.test.ts", "src/lib/thread-signal-dismissals.test.ts", @@ -1280,6 +1281,7 @@ const ALIAS_LOADER = new Set([ "src/lib/stitch-patterns.test.ts", "src/lib/server/space-usage.test.ts", "src/lib/server/stuck-created-sweep.test.ts", + "src/lib/server/stale-running-sweep.test.ts", "src/lib/server/cave-home-migration-backup-recovery.test.ts", "src/lib/server/cave-home-migration-json-merge.test.ts", "src/lib/server/cave-home-migration-status-recovery.test.ts", diff --git a/src/app/api/sessions/list/route.ts b/src/app/api/sessions/list/route.ts index 62147f483..8b0b90094 100644 --- a/src/app/api/sessions/list/route.ts +++ b/src/app/api/sessions/list/route.ts @@ -12,6 +12,10 @@ import { localConversationSessionRows, mergeSessionRows, } from "@/lib/session-list-merge"; +import { + applyStaleRunningPresentation, + sweepStaleRunningGhosts, +} from "@/lib/server/stale-running-sweep"; import { enrichSessionsWithGitContext } from "@/lib/session-git-enrich"; import { collapseFamiliarWorkspaceSessions } from "@/lib/familiar-workspace-sessions"; import { familiarWorkspacesRoot, readFamiliarWorkspaces } from "@/lib/coven-paths"; @@ -214,9 +218,17 @@ async function computeSessionsList( return isTrueProjectCwd(projectRoot); } + // Leaked `coven run` registrations (the CLI died without reporting) sit in + // "running" forever — the daemon only reconciles them at its own restart. + // Present confirmed ghosts as "orphaned" before the merge so the Running + // popover and status badges stop advertising dead processes. Read-only and + // best-effort; genuinely-live daemon PTY sessions always carry events and + // are never touched (see stale-running-sweep.ts). + const staleRunningGhosts = await sweepStaleRunningGhosts(res.data); + const sessions = await applyAutoArchiveSweep( mergeSessionRows({ - daemonSessions: res.data, + daemonSessions: applyStaleRunningPresentation(res.data, staleRunningGhosts), localConversations, state, includeArchived, diff --git a/src/lib/server/stale-running-sweep.test.ts b/src/lib/server/stale-running-sweep.test.ts new file mode 100644 index 000000000..839b2dfa4 --- /dev/null +++ b/src/lib/server/stale-running-sweep.test.ts @@ -0,0 +1,219 @@ +// @ts-nocheck +import assert from "node:assert/strict"; +import { readFileSync } from "node:fs"; +import { + applyStaleRunningPresentation, + ghostVerdictFromEventsResponse, + resetStaleRunningSweepCache, + staleRunningCandidates, + sweepStaleRunningGhosts, + STALE_RUNNING_PRESENTED_STATUS, + STALE_RUNNING_THRESHOLD_MS, +} from "./stale-running-sweep.ts"; + +const NOW = Date.parse("2026-07-24T12:00:00Z"); +const STALE_AT = new Date(NOW - STALE_RUNNING_THRESHOLD_MS - 60_000).toISOString(); +const FRESH_AT = new Date(NOW - 60_000).toISOString(); +const noActiveRun = () => false; + +const row = (over = {}) => ({ + id: "aaaaaaaa-1111-2222-3333-444444444444", + status: "running", + created_at: STALE_AT, + updated_at: STALE_AT, + ...over, +}); + +// --- candidates: only running-tone rows old enough on BOTH timestamps, with +// no in-process chat run, qualify for the probe. +{ + const opts = { nowMs: NOW, hasActiveChatRun: noActiveRun }; + assert.deepEqual( + staleRunningCandidates([row()], opts), + [row().id], + "stale running row qualifies", + ); + assert.deepEqual( + staleRunningCandidates([row({ status: "starting" }), row({ id: "b", status: "working" })], opts), + [row().id, "b"], + "the whole running-tone vocabulary qualifies (starting/working)", + ); + for (const status of ["completed", "failed", "orphaned", "killed", "idle", "created"]) { + assert.deepEqual( + staleRunningCandidates([row({ status })], opts), + [], + `${status} rows never qualify`, + ); + } + assert.deepEqual( + staleRunningCandidates([row({ created_at: FRESH_AT, updated_at: FRESH_AT })], opts), + [], + "fresh rows never qualify — a just-started run is not a ghost", + ); + assert.deepEqual( + staleRunningCandidates([row({ updated_at: FRESH_AT })], opts), + [], + "a recently-updated row never qualifies even when created long ago", + ); + assert.deepEqual( + staleRunningCandidates([row({ created_at: "garbage" })], opts), + [], + "unparsable created_at is skipped, not swept", + ); + assert.deepEqual( + staleRunningCandidates([row({ updated_at: "garbage" })], opts), + [], + "unparsable updated_at is skipped, not swept", + ); + assert.deepEqual( + staleRunningCandidates([row()], { nowMs: NOW, hasActiveChatRun: () => true }), + [], + "an in-flight chat run in this server process is never a candidate", + ); +} + +// --- verdict parser: process I/O proves a live daemon child; a metadata-only +// log is still a ghost (the real-world "codex patch_metadata" leak); an +// incomplete or malformed page is unrulable and must fail open. +{ + const page = (events, hasMore = false) => ({ events, nextCursor: null, hasMore }); + assert.equal(ghostVerdictFromEventsResponse(page([])), true, "empty event log = ghost"); + assert.equal( + ghostVerdictFromEventsResponse(page([{ kind: "patch_metadata" }])), + true, + "metadata-only log is still a ghost — no process ever produced I/O", + ); + assert.equal( + ghostVerdictFromEventsResponse(page([{ kind: "output" }])), + false, + "PTY output proves a live daemon child", + ); + assert.equal( + ghostVerdictFromEventsResponse(page([{ kind: "patch_metadata" }, { kind: "input" }])), + false, + "input events also prove process wiring", + ); + assert.equal( + ghostVerdictFromEventsResponse(page([{ kind: "patch_metadata" }], true)), + null, + "metadata with more pages behind it is unrulable — fail open", + ); + assert.equal(ghostVerdictFromEventsResponse(null), null, "missing body fails open"); + assert.equal(ghostVerdictFromEventsResponse({ events: "nope" }), null, "malformed body fails open"); +} + +// --- sweep: the events probe is the ghost/alive discriminator. +// `coven run`-registered rows never produce daemon events; daemon-spawned PTY +// sessions always do. Probe failures fail open (row keeps its status). +{ + resetStaleRunningSweepCache(); + const probes = []; + const probe = async (id) => { + probes.push(id); + if (id === "ghost") return true; // zero events + if (id === "pty") return false; // has events — a live daemon child + return null; // probe failed — unknown + }; + const rows = [ + row({ id: "ghost" }), + row({ id: "pty" }), + row({ id: "unknown" }), + row({ id: "fresh", created_at: FRESH_AT, updated_at: FRESH_AT }), + ]; + const ghosts = await sweepStaleRunningGhosts(rows, { + nowMs: NOW, + hasActiveChatRun: noActiveRun, + probe, + }); + assert.deepEqual([...ghosts], ["ghost"], "only the zero-events row is a ghost"); + assert.deepEqual( + probes.sort(), + ["ghost", "pty", "unknown"], + "every stale candidate is probed; fresh rows are not", + ); +} + +// --- verdict cache: definite verdicts stick per (id, updated_at); probe +// failures retry; a daemon-side transition (new updated_at) re-evaluates. +{ + resetStaleRunningSweepCache(); + let calls = 0; + const probe = async (id) => { + calls += 1; + return id === "ghost" ? true : id === "pty" ? false : null; + }; + const rows = [row({ id: "ghost" }), row({ id: "pty" }), row({ id: "unknown" })]; + const opts = { nowMs: NOW, hasActiveChatRun: noActiveRun, probe }; + + const first = await sweepStaleRunningGhosts(rows, opts); + assert.deepEqual([...first], ["ghost"]); + assert.equal(calls, 3, "first sweep probes all three"); + + const second = await sweepStaleRunningGhosts(rows, opts); + assert.deepEqual([...second], ["ghost"], "cached ghost verdict still presents"); + assert.equal(calls, 4, "second sweep re-probes only the failed verdict"); + + const bumped = row({ id: "ghost", updated_at: new Date(NOW - STALE_RUNNING_THRESHOLD_MS - 30_000).toISOString() }); + await sweepStaleRunningGhosts([bumped], opts); + assert.equal(calls, 5, "a changed updated_at invalidates the cached verdict"); +} + +// --- a throwing probe never breaks the sweep (best-effort contract). +{ + resetStaleRunningSweepCache(); + const ghosts = await sweepStaleRunningGhosts([row()], { + nowMs: NOW, + hasActiveChatRun: noActiveRun, + probe: async () => { + throw new Error("daemon exploded"); + }, + }); + assert.deepEqual([...ghosts], [], "probe throw fails open"); +} + +// --- presentation: ghosts are rewritten to the daemon's own restart verdict +// ("orphaned"); everything else passes through untouched. +{ + const rows = [row({ id: "ghost" }), row({ id: "live" })]; + const out = applyStaleRunningPresentation(rows, new Set(["ghost"])); + assert.equal(out[0].status, STALE_RUNNING_PRESENTED_STATUS, "ghost presents as orphaned"); + assert.equal(out[1].status, "running", "non-ghost keeps its daemon status"); + assert.equal( + applyStaleRunningPresentation(rows, new Set()), + rows, + "empty ghost set returns the same array (no churn for the common case)", + ); +} + +// --- wiring pins: the sessions/list route must classify BEFORE the merge so +// every downstream surface (Running popover, badges, archive sweeps) sees the +// presented status; the probe must stay read-only. +{ + const route = readFileSync( + new URL("../../app/api/sessions/list/route.ts", import.meta.url), + "utf8", + ); + assert.match( + route, + /sweepStaleRunningGhosts\(res\.data\)/, + "list route sweeps the daemon rows", + ); + assert.match( + route, + /daemonSessions: applyStaleRunningPresentation\(res\.data, staleRunningGhosts\)/, + "presentation applies to the daemon rows fed into mergeSessionRows", + ); + const sweep = readFileSync(new URL("./stale-running-sweep.ts", import.meta.url), "utf8"); + assert.match( + sweep, + /\/events\?limit=\$\{EVENTS_PROBE_LIMIT\}/, + "liveness probe is the read-only events endpoint", + ); + assert.doesNotMatch( + sweep, + /method:\s*"(POST|DELETE|PUT)"/, + "sweep never mutates the daemon", + ); +} + +console.log("stale-running-sweep tests passed"); diff --git a/src/lib/server/stale-running-sweep.ts b/src/lib/server/stale-running-sweep.ts new file mode 100644 index 000000000..07676f5e5 --- /dev/null +++ b/src/lib/server/stale-running-sweep.ts @@ -0,0 +1,217 @@ +/** + * stale-running-sweep.ts — present leaked "running" daemon rows as orphaned. + * + * `coven run`-registered sessions (board enrich-steps, workflows, chat + * one-shots) leave the daemon a row it does not own a process for: the CLI + * spawns the harness itself and reports terminal status back. When that CLI + * process dies without reporting — request abort SIGTERMs it, or the app/ + * server quits mid-run — the row is stuck in "running" forever. The daemon + * only reconciles at ITS OWN restart (marking them "orphaned"); `kill` + * refuses (`session_not_live`), `sacrifice` refuses ("still running"), and + * there is no HTTP delete route. Meanwhile every running-tone surface (the + * menu bar's Running-processes popover, sidebar badges) presents the ghost as + * a live process whose chat opens to "Chat history unavailable". + * + * This sweep converges the *presentation* with what the daemon itself would + * say after a restart: a running-tone row that is old enough and shows no + * process I/O in the daemon's event log is presented as "orphaned". The + * events probe is the discriminator that protects genuinely-live + * daemon-spawned PTY sessions — those are daemon children wired straight + * into the events table (they emit `output` events from their first prompt + * paint), while `coven run` sessions never produce process I/O (at most a + * stray metadata annotation like `patch_metadata`). The probe is read-only + * (`GET /api/v1/sessions/{id}/events`), so a mid-flight slow run is never + * harmed: if it later reports completion the daemon row transitions + * normally and the override stops applying. + */ + +import { callDaemon } from "@/lib/coven-daemon"; +import { hasActiveChatRun } from "@/lib/server/chat-stop-registry"; +import { parseDaemonTime } from "@/lib/server/stuck-created-sweep"; +import { sessionStatusTone } from "@/lib/session-status"; + +/** Minimal structural cut of the daemon session row the sweep reads. */ +export type StaleRunningRow = { + id: string; + status: string; + created_at: string; + updated_at: string; +}; + +/** + * A row this old that still claims a running tone, with nothing having + * touched `updated_at` since creation and no in-process chat run, is past + * any legitimate `coven run` lifetime (enrich runs are request-scoped + * minutes; chat turns are covered by the active-run registry). + */ +export const STALE_RUNNING_THRESHOLD_MS = 30 * 60 * 1000; + +/** Presented status for a confirmed ghost — the daemon's own restart verdict. */ +export const STALE_RUNNING_PRESENTED_STATUS = "orphaned"; + +/** + * Pure classifier: which rows are *candidates* for the ghost probe? + * Running tone + both timestamps beyond the threshold + no live chat run in + * this server process. Unparsable timestamps never qualify (NaN fails the + * comparison), so a malformed row is skipped, not swept. + */ +export function staleRunningCandidates( + rows: StaleRunningRow[], + opts: { + nowMs: number; + thresholdMs?: number; + hasActiveChatRun: (sessionId: string) => boolean; + }, +): string[] { + const threshold = opts.thresholdMs ?? STALE_RUNNING_THRESHOLD_MS; + return rows + .filter((row) => { + if (sessionStatusTone(row.status) !== "running") return false; + const createdMs = parseDaemonTime(row.created_at); + const updatedMs = parseDaemonTime(row.updated_at); + if (!(opts.nowMs - createdMs >= threshold)) return false; + if (!(opts.nowMs - updatedMs >= threshold)) return false; + return !opts.hasActiveChatRun(row.id); + }) + .map((row) => row.id); +} + +/** + * Probe verdict for one session: `true` = confirmed ghost (probe succeeded, + * no process I/O events), `false` = provably alive (has PTY output/input), + * `null` = unknown (probe failed — daemon busy/offline/no events capability, + * or too many metadata events to rule on). Unknown must fail open: the row + * keeps its daemon status. + */ +export type GhostVerdict = boolean | null; + +type EventsProbe = (sessionId: string) => Promise; + +/** Event kinds that prove a real harness process was wired to the daemon. + * Daemon-spawned PTY children emit `output` immediately (prompt paint); + * leaked `coven run` registrations never do — their event log is empty or + * carries only metadata annotations (`patch_metadata`, `cast`). */ +const PROCESS_IO_EVENT_KINDS = new Set(["output", "input"]); + +const EVENTS_PROBE_LIMIT = 16; + +/** + * Pure verdict from an events-endpoint response. Ghost only when the probe + * saw the session's *complete* event log (no `hasMore`) and none of it is + * process I/O. A page full of metadata with more behind it is unrulable — + * fail open rather than risk hiding a live process. + */ +export function ghostVerdictFromEventsResponse(data: unknown): GhostVerdict { + if (!data || typeof data !== "object") return null; + const events = (data as { events?: unknown }).events; + if (!Array.isArray(events)) return null; + for (const event of events) { + const kind = (event as { kind?: unknown } | null)?.kind; + if (typeof kind === "string" && PROCESS_IO_EVENT_KINDS.has(kind)) return false; + } + const hasMore = (data as { hasMore?: unknown }).hasMore; + if (hasMore === true || events.length >= EVENTS_PROBE_LIMIT) return null; + return true; +} + +async function probeSessionForGhost(sessionId: string): Promise { + const res = await callDaemon({ + path: `/api/v1/sessions/${encodeURIComponent(sessionId)}/events?limit=${EVENTS_PROBE_LIMIT}`, + }); + if (!res.ok) return null; + return ghostVerdictFromEventsResponse(res.data); +} + +// Verdict cache so the polled session list doesn't re-probe every request. +// Keyed by id|updated_at: any daemon-side transition (a slow run finally +// reporting completion) changes the key and invalidates the entry. Only +// definite verdicts are cached — probe failures retry on the next sweep. +const verdictCache = new Map(); +const VERDICT_CACHE_MAX = 512; + +function cacheVerdict(key: string, verdict: boolean): void { + if (verdictCache.size >= VERDICT_CACHE_MAX) { + const oldest = verdictCache.keys().next().value; + if (oldest !== undefined) verdictCache.delete(oldest); + } + verdictCache.set(key, verdict); +} + +/** Test seam: reset module-level cache state. */ +export function resetStaleRunningSweepCache(): void { + verdictCache.clear(); +} + +// Bound the per-sweep probe fan-out. Ghosts accumulate slowly (a handful per +// leak incident); anything beyond the cap waits for the next list request. +const MAX_PROBES_PER_SWEEP = 16; + +/** + * Classify candidates and return the set of confirmed-ghost session ids. + * Best-effort and read-only: never throws, probe failures leave rows + * untouched. Injectable probe keeps the daemon out of unit tests. + */ +export async function sweepStaleRunningGhosts( + rows: StaleRunningRow[], + opts?: { + nowMs?: number; + thresholdMs?: number; + hasActiveChatRun?: (sessionId: string) => boolean; + probe?: EventsProbe; + }, +): Promise> { + const ghosts = new Set(); + try { + const candidates = staleRunningCandidates(rows, { + nowMs: opts?.nowMs ?? Date.now(), + thresholdMs: opts?.thresholdMs, + hasActiveChatRun: opts?.hasActiveChatRun ?? hasActiveChatRun, + }); + if (candidates.length === 0) return ghosts; + + const rowById = new Map(rows.map((row) => [row.id, row])); + const probe = opts?.probe ?? probeSessionForGhost; + const toProbe: Array<{ id: string; key: string }> = []; + for (const id of candidates) { + const key = `${id}|${rowById.get(id)?.updated_at ?? ""}`; + const cached = verdictCache.get(key); + if (cached === true) ghosts.add(id); + else if (cached === undefined && toProbe.length < MAX_PROBES_PER_SWEEP) { + toProbe.push({ id, key }); + } + } + + const verdicts = await Promise.all( + toProbe.map(async ({ id, key }) => { + try { + return { id, key, verdict: await probe(id) }; + } catch { + return { id, key, verdict: null as GhostVerdict }; + } + }), + ); + for (const { id, key, verdict } of verdicts) { + if (verdict === null) continue; + cacheVerdict(key, verdict); + if (verdict) ghosts.add(id); + } + return ghosts; + } catch { + return ghosts; + } +} + +/** + * Rewrite daemon rows so confirmed ghosts carry the presented "orphaned" + * status before the merge — downstream (status tones, the Running popover, + * archive sweeps) then treats them exactly like restart-orphaned rows. + */ +export function applyStaleRunningPresentation( + rows: T[], + ghosts: Set, +): T[] { + if (ghosts.size === 0) return rows; + return rows.map((row) => + ghosts.has(row.id) ? { ...row, status: STALE_RUNNING_PRESENTED_STATUS } : row, + ); +}