diff --git a/packages/happy-cli/src/daemon/run.ts b/packages/happy-cli/src/daemon/run.ts index 92f167bd73..8bf36ad8b0 100644 --- a/packages/happy-cli/src/daemon/run.ts +++ b/packages/happy-cli/src/daemon/run.ts @@ -35,6 +35,7 @@ import { } from './sessionEnvironment'; import { startHappyTerminalDaemon } from './happyTerminalBoot'; import { appendDaemonSpawnModeArgs, shouldForwardDaemonPermissionMode } from './spawnModeArgs'; +import { classifyResumeConflict, isPidAlive, machineBootTimeMs, resolveLiveOwnerPid, type SessionPresence } from './sessionLiveness'; /** Shell-escape a string for safe interpolation into tmux commands. */ function shellescape(s: string): string { @@ -174,6 +175,10 @@ export async function startDaemon(): Promise { // Setup state - key by PID const pidToTrackedSession = new Map(); + // Fixed for the daemon's lifetime: used to reject PIDs recorded before the + // machine's current boot, which name whatever reused the number since. + const bootTimeMs = machineBootTimeMs(os.uptime(), Date.now()); + // Retain session data after process exits so resume can still find it. // Pre-populate from disk so sessions survive daemon restarts. const sessionIdToFinishedSession = new Map(); @@ -668,24 +673,131 @@ export async function startDaemon(): Promise { return sessionIdToFinishedSession.get(happySessionId); }; - const fetchServerSessionMetadata = async (sessionId: string, encryptionKey: Uint8Array, encryptionVariant: 'legacy' | 'dataKey'): Promise => { + type ServerSessionRow = { id: string; metadata: string; active?: boolean }; + + /** + * The server's row for one session, or a reason we could not get it. The two + * failures are kept apart because they are not interchangeable to a caller + * deciding whether to kill a process. + */ + const fetchServerSession = async (sessionId: string): Promise<{ ok: true; row: ServerSessionRow } | { ok: false; reason: 'unreachable' | 'unknown-session' }> => { try { const response = await axios.get(`${configuration.serverUrl}/v1/sessions`, { headers: { Authorization: `Bearer ${credentials.token}` }, timeout: 10_000, }); - const sessions = (response.data as { sessions: { id: string; metadata: string }[] }).sessions; + const sessions = (response.data as { sessions: ServerSessionRow[] }).sessions; const matched = sessions.find(s => s.id === sessionId); - if (!matched) return null; - const decrypted = decrypt(encryptionKey, encryptionVariant, decodeBase64(matched.metadata)); + if (!matched) return { ok: false, reason: 'unknown-session' }; + return { ok: true, row: matched }; + } catch (error) { + logger.debug(`[DAEMON RUN] Failed to fetch session from server: ${error instanceof Error ? error.message : error}`); + return { ok: false, reason: 'unreachable' }; + } + }; + + const fetchServerSessionMetadata = async (sessionId: string, encryptionKey: Uint8Array, encryptionVariant: 'legacy' | 'dataKey'): Promise => { + const fetched = await fetchServerSession(sessionId); + if (!fetched.ok) return null; + try { + const decrypted = decrypt(encryptionKey, encryptionVariant, decodeBase64(fetched.row.metadata)); return decrypted as Metadata | null; } catch (error) { - logger.debug(`[DAEMON RUN] Failed to fetch session metadata from server: ${error instanceof Error ? error.message : error}`); + logger.debug(`[DAEMON RUN] Failed to decrypt session metadata from server: ${error instanceof Error ? error.message : error}`); return null; } }; - const resumeSession = async (happySessionId: string, options?: { model?: string; permissionMode?: string }): Promise => { + const fetchSessionPresence = async (sessionId: string): Promise => { + const fetched = await fetchServerSession(sessionId); + if (!fetched.ok) return { ok: false, reason: fetched.reason }; + return { ok: true, active: Boolean(fetched.row.active) }; + }; + + /** Every PID currently tracked for a session — see `ResolveLiveOwnerInput.trackedPids`. */ + const findTrackedPidsBySessionId = (happySessionId: string): number[] => { + const pids: number[] = []; + for (const [pid, session] of pidToTrackedSession.entries()) { + if (session.happySessionId === happySessionId) pids.push(pid); + } + return pids; + }; + + /** + * Stop the process that owns a session, and confirm it is gone. + * + * Confirming is the whole job. The only caller wants to spawn a replacement, + * and spawning while the old process is still up is exactly the two-owner + * state this path exists to prevent — so "the signal was sent" is not an + * answer, and a process that will not die has to be reported rather than + * quietly stepped over. + */ + const terminateSessionOwner = async (happySessionId: string, pid: number): Promise => { + // Signal the whole process group, not just the Happy CLI parent: the + // harness runs its backend as a grandchild (see stopSession below), which a + // bare SIGTERM to the parent never reaches. Daemon-spawned sessions are + // `detached`, so the parent leads a group of exactly its own descendants. + // A session started from a terminal is not a group leader, and then no + // group carries this ID at all — a group's ID is its leader's PID, and that + // PID belongs to the process we are aiming at — so the negative signal can + // only ESRCH, never reach a bystander. The direct signal below covers it. + const signal = (sig: NodeJS.Signals) => { + if (process.platform !== 'win32') { + try { + process.kill(-pid, sig); + return; + } catch { /* not a group leader, or already gone */ } + } + try { + process.kill(pid, sig); + } catch { /* already gone */ } + }; + + const waitForExit = async (timeoutMs: number): Promise => { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (!isPidAlive(pid)) return true; + await new Promise((resolve) => setTimeout(resolve, 100)); + } + return !isPidAlive(pid); + }; + + logger.debug(`[DAEMON RUN] Stopping unresponsive owner PID ${pid} of session ${happySessionId}`); + signal('SIGTERM'); + if (await waitForExit(5_000)) return true; + + logger.debug(`[DAEMON RUN] PID ${pid} ignored SIGTERM, escalating to SIGKILL`); + signal('SIGKILL'); + return waitForExit(2_000); + }; + + /** + * Resume requests for one session, collapsed onto a single attempt while it + * runs. + * + * The owner check below reads state that the spawn it guards is about to + * change, and a spawned process takes a second or two to report itself. Two + * requests arriving inside that window — a double tap, or two clients acting + * on the same stale view — would both look at a session with no owner and + * both spawn one. Sharing the in-flight attempt closes that window without a + * lock, and makes a repeated request idempotent rather than additive. + */ + const resumesInFlight = new Map>(); + + const resumeSession = (happySessionId: string, options?: { model?: string; permissionMode?: string }): Promise => { + const inFlight = resumesInFlight.get(happySessionId); + if (inFlight) { + logger.debug(`[DAEMON RUN] Resume of session ${happySessionId} is already in progress — joining it instead of starting a second one`); + return inFlight; + } + const attempt = resumeSessionAttempt(happySessionId, options).finally(() => { + resumesInFlight.delete(happySessionId); + }); + resumesInFlight.set(happySessionId, attempt); + return attempt; + }; + + const resumeSessionAttempt = async (happySessionId: string, options?: { model?: string; permissionMode?: string }): Promise => { try { const tracked = findTrackedSessionById(happySessionId); if (!tracked) { @@ -698,6 +810,43 @@ export async function startDaemon(): Promise { return { type: 'error', errorMessage: `Session ${happySessionId} has no stored encryption data. It was likely started before this feature was available. Restart the daemon and start a new session to enable resume.` }; } + // A session has exactly one runtime. Its message stream, metadata version + // and agent-state version are counters that one process advances, so a + // second process spawned for the same session gives the server two + // writers on all three: the user's next message is delivered to both, and + // both act on it. Nothing downstream de-duplicates that. + // + // Resume has no guard against it today, and the request can legitimately + // arrive while a process is still running — the caller decides from its + // own view of the session, and a second device (or a client that has not + // yet seen the reconnect) shows it as stopped for as long as its state is + // behind. So check for an owner here rather than trusting the caller. + // + // See sessionLiveness.ts for why `active === false` on a live process + // means wedged rather than merely disconnected, and why an unanswerable + // server is treated as "still running". + const liveOwnerPid = resolveLiveOwnerPid({ + trackedPids: findTrackedPidsBySessionId(happySessionId), + // Read from disk rather than the startup snapshot: a session that + // registered after this daemon booted is only on disk. + persisted: readPersistedSessions()[happySessionId], + bootTimeMs, + }); + if (liveOwnerPid !== undefined) { + const presence = await fetchSessionPresence(happySessionId); + const conflict = classifyResumeConflict(liveOwnerPid, presence); + if (conflict === 'already-running') { + logger.debug(`[DAEMON RUN] Session ${happySessionId} is still owned by live PID ${liveOwnerPid} (server presence: ${presence.ok ? `active=${presence.active}` : presence.reason}) — returning the running session instead of spawning a second process`); + return { type: 'success', sessionId: happySessionId }; + } + logger.debug(`[DAEMON RUN] Session ${happySessionId} is owned by PID ${liveOwnerPid} but the server reports no runtime attached — replacing it`); + if (!await terminateSessionOwner(happySessionId, liveOwnerPid)) { + // Refusing is the safe direction: the old process is still alive, so + // spawning now would produce the duplicate this check exists to stop. + return { type: 'error', errorMessage: `Session ${happySessionId} is held by process ${liveOwnerPid}, which did not exit. Stop it manually and try again.` }; + } + } + // Webhook metadata may be stale (missing claudeSessionId/codexThreadId set after startup). // Fetch fresh metadata from server if needed. let metadata = tracked.happySessionMetadataFromLocalWebhook; @@ -908,11 +1057,7 @@ export async function startDaemon(): Promise { // Prune stale sessions for (const [pid, _] of pidToTrackedSession.entries()) { - try { - // Check if process is still alive (signal 0 doesn't kill, just checks) - process.kill(pid, 0); - } catch (error) { - // Process is dead, remove from tracking + if (!isPidAlive(pid)) { logger.debug(`[DAEMON RUN] Removing stale session with PID ${pid} (process no longer exists)`); pidToTrackedSession.delete(pid); } diff --git a/packages/happy-cli/src/daemon/sessionLiveness.test.ts b/packages/happy-cli/src/daemon/sessionLiveness.test.ts new file mode 100644 index 0000000000..e0c24ecfd7 --- /dev/null +++ b/packages/happy-cli/src/daemon/sessionLiveness.test.ts @@ -0,0 +1,136 @@ +import { describe, expect, it } from 'vitest'; +import type { PersistedSession } from '@/persistence'; +import { + classifyResumeConflict, + isPidAlive, + machineBootTimeMs, + resolveLiveOwnerPid, + type SessionPresence, +} from './sessionLiveness'; + +const BOOT = 1_000_000; +const alive = (pids: number[]) => (pid: number | undefined) => pid !== undefined && pids.includes(pid); + +describe('isPidAlive', () => { + it('reports the current process as alive', () => { + expect(isPidAlive(process.pid)).toBe(true); + }); + + it('reports a PID the kernel rejects as dead', () => { + const kill = () => { throw new Error('ESRCH'); }; + expect(isPidAlive(4242, kill)).toBe(false); + }); + + it('treats absent and non-positive PIDs as dead without probing', () => { + const kill = () => { throw new Error('should not be called'); }; + expect(isPidAlive(undefined, kill)).toBe(false); + expect(isPidAlive(0, kill)).toBe(false); + expect(isPidAlive(-1, kill)).toBe(false); + }); +}); + +describe('machineBootTimeMs', () => { + it('subtracts uptime from now', () => { + expect(machineBootTimeMs(60, 1_000_000)).toBe(1_000_000 - 60_000); + }); +}); + +describe('resolveLiveOwnerPid', () => { + it('returns nothing when the session has no tracked or persisted process', () => { + expect(resolveLiveOwnerPid({ trackedPids: [], bootTimeMs: BOOT, isAlive: alive([]) })).toBeUndefined(); + }); + + it('returns a tracked PID that is still alive', () => { + expect(resolveLiveOwnerPid({ trackedPids: [77], bootTimeMs: BOOT, isAlive: alive([77]) })).toBe(77); + }); + + it('skips a dead tracked entry left behind next to a live one', () => { + // Tracking is keyed by PID and is cleared lazily, so a session can hold a + // stale entry alongside its real one. Answering from the stale entry would + // report "no owner" and wave a duplicate spawn through. + expect(resolveLiveOwnerPid({ trackedPids: [11, 22], bootTimeMs: BOOT, isAlive: alive([22]) })).toBe(22); + }); + + it('falls back to the on-disk record when the daemon tracks nothing', () => { + // The case after a daemon restart: children are detached and outlive it. + const owner = resolveLiveOwnerPid({ + trackedPids: [], + persisted: { metadata: { hostPid: 99 }, savedAt: BOOT + 5_000 }, + bootTimeMs: BOOT, + isAlive: alive([99]), + }); + expect(owner).toBe(99); + }); + + it('ignores an on-disk PID recorded before the current boot', () => { + // A reboot resets the PID space, so that number now belongs to whatever + // reused it — probing it would be a confident false positive. + const owner = resolveLiveOwnerPid({ + trackedPids: [], + persisted: { metadata: { hostPid: 99 }, savedAt: BOOT - 1 }, + bootTimeMs: BOOT, + isAlive: alive([99]), + }); + expect(owner).toBeUndefined(); + }); + + it('reads the PID out of a record shaped exactly as the daemon stores it', () => { + // Pins the parameter against the real `PersistedSession`. The PID lives + // under `metadata`, and a version of this that reached for a top-level + // field still type-checked, still passed every other test here, and + // resolved every owner to `undefined` — a guard that never fires is + // indistinguishable from one with nothing to guard against. + const stored: PersistedSession = { + encryptionKey: '', + encryptionVariant: 'dataKey', + seq: 0, + metadataVersion: 0, + agentStateVersion: 0, + metadata: { path: '/tmp/project', hostPid: 1234 } as PersistedSession['metadata'], + savedAt: BOOT + 1, + }; + const owner = resolveLiveOwnerPid({ + trackedPids: [], + persisted: stored, + bootTimeMs: BOOT, + isAlive: alive([1234]), + }); + expect(owner).toBe(1234); + }); + + it('returns nothing when the on-disk PID is dead', () => { + const owner = resolveLiveOwnerPid({ + trackedPids: [], + persisted: { metadata: { hostPid: 99 }, savedAt: BOOT + 5_000 }, + bootTimeMs: BOOT, + isAlive: alive([]), + }); + expect(owner).toBeUndefined(); + }); +}); + +describe('classifyResumeConflict', () => { + const active: SessionPresence = { ok: true, active: true }; + const inactive: SessionPresence = { ok: true, active: false }; + + it('allows the spawn when no process owns the session', () => { + expect(classifyResumeConflict(undefined, inactive)).toBe('none'); + }); + + it('reports a healthy owner as already running', () => { + expect(classifyResumeConflict(50, active)).toBe('already-running'); + }); + + it('reports a live process the server no longer sees as wedged', () => { + expect(classifyResumeConflict(50, inactive)).toBe('wedged'); + }); + + it.each(['unreachable', 'unknown-session'] as const)( + 'treats a %s server as already running rather than wedged', + (reason) => { + // A failed probe must not read as "nobody is attached": that answer + // gets the owning process killed. Refusing costs the user a retry. + expect(classifyResumeConflict(50, { ok: false, reason })).toBe('already-running'); + }, + ); +}); diff --git a/packages/happy-cli/src/daemon/sessionLiveness.ts b/packages/happy-cli/src/daemon/sessionLiveness.ts new file mode 100644 index 0000000000..296ac3319b --- /dev/null +++ b/packages/happy-cli/src/daemon/sessionLiveness.ts @@ -0,0 +1,165 @@ +/** + * Is a Happy session's process still running, and if so, is it still doing its + * job? + * + * A Happy session is single-owner by construction: its message stream, its + * metadata version and its agent-state version are all counters advanced by one + * runtime. Spawning a second process for a session that already has one gives + * the server two writers on those counters, and every user message is then + * delivered to — and acted on by — both. So `resumeSession` has to answer this + * question before it spawns anything. + */ + +/** + * Does this PID currently exist? + * + * `process.kill(pid, 0)` does not signal the target — signal 0 only performs the + * permission-and-existence check, which is exactly the probe we want. + */ +export function isPidAlive(pid: number | undefined, kill: (pid: number, signal: 0) => void = process.kill): boolean { + if (!pid || pid <= 0) { + return false; + } + try { + kill(pid, 0); + return true; + } catch { + return false; + } +} + +/** + * When the machine booted, in epoch milliseconds. + * + * A reboot resets the PID space, so a PID recorded before the current boot says + * nothing about the process that holds that number now — `isPidAlive` on it is a + * plausible-looking false positive. Records carry a `savedAt` stamped while the + * session was running, so `savedAt < bootTime` proves the record predates the + * boot and the process it names is gone. + */ +export function machineBootTimeMs(uptimeSeconds: number, nowMs: number): number { + return nowMs - uptimeSeconds * 1000; +} + +/** + * A session record as it survives a daemon restart, on disk. + * + * Shaped to accept `PersistedSession` as-is. An adapter here would type-check + * against a mistyped field and quietly resolve every PID to `undefined` — a + * guard that always says "no owner" is indistinguishable from one that works. + */ +export interface PersistedOwnerRecord { + /** The metadata the session reported for itself, including its PID. */ + metadata?: { hostPid?: number }; + /** When that report was written — necessarily while the process was alive. */ + savedAt: number; +} + +export interface ResolveLiveOwnerInput { + /** + * Every PID this daemon currently tracks for the session. + * + * Plural on purpose. Tracking is keyed by PID, and an entry for a process + * that has died is only cleared when its exit event or the next heartbeat + * sweep gets to it — so a session can legitimately have a stale entry + * alongside a live one, and asking only the first would report the dead one + * and wave a duplicate through. Entries here are always created during the + * current daemon's life, hence during the current boot, so they need no boot + * gate. + */ + trackedPids: number[]; + /** + * The on-disk record for the session. Needed because a session outlives the + * daemon that spawned it (children are detached), so after a daemon restart + * the only pointer to a still-running process is this record. + */ + persisted?: PersistedOwnerRecord; + /** Epoch ms of the current boot — see {@link machineBootTimeMs}. */ + bootTimeMs: number; + /** Injected for tests. */ + isAlive?: (pid: number | undefined) => boolean; +} + +/** + * The PID of the process that still owns this session, or `undefined` if no + * process does. + */ +export function resolveLiveOwnerPid(input: ResolveLiveOwnerInput): number | undefined { + const isAlive = input.isAlive ?? ((pid: number | undefined) => isPidAlive(pid)); + + for (const pid of input.trackedPids) { + if (pid > 0 && isAlive(pid)) { + return pid; + } + } + + const persisted = input.persisted; + const hostPid = persisted?.metadata?.hostPid; + if (!persisted || !hostPid) { + return undefined; + } + // Pre-boot record: the PID it names belongs to whatever reused the number. + if (persisted.savedAt < input.bootTimeMs) { + return undefined; + } + return isAlive(hostPid) ? hostPid : undefined; +} + +/** + * What the server knows about a session's runtime. + * + * `ok: false` is deliberately not collapsed into `active: false`: "the server + * says nobody is attached" and "we could not ask the server" have to drive + * different decisions, and merging them is what turns a failed probe into a + * confident answer. + */ +export type SessionPresence = + | { ok: true; active: boolean } + | { ok: false; reason: 'unreachable' | 'unknown-session' }; + +export type ResumeConflict = + /** Nothing owns the session — spawn normally. */ + | 'none' + /** A healthy process owns it; the caller's view is stale. */ + | 'already-running' + /** A process owns it but has stopped serving; replace it. */ + | 'wedged'; + +/** + * Decide what a resume request should do about the process that already owns the + * session. + * + * Only two states can produce a resume request for a session whose process is + * still up: + * + * - the process is up and the server still sees it, so the CALLER is stale — + * another device, or a client that has not yet received the reconnect. The + * running session is the right answer; a rival process would corrupt it. + * - the process is up but the server has not seen it, so it is holding a dead + * socket. That is precisely the state the user is trying to escape, and + * refusing would strand them. + * + * The server's `active` flag is safe to read this way because of how it is + * written: a running session heartbeats every 2s (`Session` in + * `claude/session.ts`), `active` only goes false on an explicit shutdown signal + * (`session-end`, or the deactivate route the CLI calls on SIGTERM) or after 10 + * minutes of silence (`presence/timeout.ts`). A dropped connection does not flip + * it — the heartbeat is `volatile`, so a blip merely stops refreshing + * `lastActiveAt`, and the 10-minute sweep is the grace period. `active === false` + * on a process that is still alive therefore means it announced it was leaving + * and did not, or it has been silent for 300× its heartbeat interval. + * + * When the server cannot be asked, the answer is `already-running`, not + * `wedged`. The two mistakes are not symmetric: refusing to spawn costs the user + * a retry, while killing on a guess costs whatever the process was in the middle + * of doing. + */ +export function classifyResumeConflict(liveOwnerPid: number | undefined, presence: SessionPresence): ResumeConflict { + if (liveOwnerPid === undefined) { + return 'none'; + } + if (presence.ok && !presence.active) { + return 'wedged'; + } + return 'already-running'; +}