diff --git a/src/daemon.ts b/src/daemon.ts index 56f8326..a19de0c 100644 --- a/src/daemon.ts +++ b/src/daemon.ts @@ -336,10 +336,12 @@ type SessionState = { const INACTIVITY_TIMEOUT_MS = 10 * 60 * 1_000; // 10 minutes // Absolute ceiling for holding the daemon open past the normal inactivity -// timeout while cross-session team work is in flight (see checkInactivity). -// Bounds the case where a teammate crashes and never emits TeammateIdle, so an -// unemitted entry can't pin the daemon forever. -const TEAM_INFLIGHT_MAX_MS = 60 * 60 * 1_000; // 60 minutes +// timeout while work is still in flight — either cross-session team +// correlation (hasUnemittedTeamMembers) or an ordinary open turn / pending +// tool / tracked subagent (hasInFlightWork); see checkInactivity. Bounds the +// pathological case (a teammate that never emits TeammateIdle, or a stuck +// session) so an in-flight entry can't pin the daemon forever. +const INFLIGHT_HOLD_MAX_MS = 60 * 60 * 1_000; // 60 minutes const CONNECTION_TIMEOUT_MS = 5_000; // 5 seconds per connection const MAX_SOCKET_PAYLOAD_BYTES = 4 * 1024 * 1024; // 4 MiB per message @@ -1818,12 +1820,22 @@ export class GlobalDaemon { // still-open specialist span. Agent-teams runs have quiet windows (engineer // think-time; gaps between spawn and first teammate report) that would // otherwise trip the 10-min timeout mid-triage. Hold open until the team - // work drains, bounded by TEAM_INFLIGHT_MAX_MS so a crashed teammate that + // work drains, bounded by INFLIGHT_HOLD_MAX_MS so a crashed teammate that // never emits TeammateIdle can't pin the daemon indefinitely. - if (idle < TEAM_INFLIGHT_MAX_MS && this.hasUnemittedTeamMembers()) { + if (idle < INFLIGHT_HOLD_MAX_MS && this.hasUnemittedTeamMembers()) { this.log('DEBUG', 'Inactivity timeout reached but team correlation in flight — staying up'); return; } + // Also hold open while ordinary work is in flight: an open turn span, a + // pending tool call, or a tracked subagent. A long-running tool or turn + // (longer than the timeout, with no other session active) would otherwise + // trip the timeout mid-flight — dropping the still-open spans and forcing + // the resumed work onto a fresh, amnesiac daemon. Same INFLIGHT_HOLD_MAX_MS + // ceiling so a stuck session can't pin the daemon indefinitely. + if (idle < INFLIGHT_HOLD_MAX_MS && this.hasInFlightWork()) { + this.log('DEBUG', 'Inactivity timeout reached but work in flight — staying up'); + return; + } this.log('INFO', 'Inactivity timeout — shutting down'); void this.shutdown('inactivity'); } @@ -1837,6 +1849,18 @@ export class GlobalDaemon { return false; } + /** True if any session has work in flight: an open turn span, a pending tool + * call, or a tracked subagent. Keeps the daemon alive across the inactivity + * timeout so in-flight work isn't cut off mid-flight (see checkInactivity). */ + private hasInFlightWork(): boolean { + for (const s of this.sessions.values()) { + if (s.currentTurnSpan) return true; + if (s.pendingToolCalls.size > 0) return true; + if (s.subagents.size() > 0) return true; + } + return false; + } + private async shutdown(reason: string): Promise { if (!this.running) return; this.running = false; diff --git a/tests/daemon-idle-inflight.test.ts b/tests/daemon-idle-inflight.test.ts new file mode 100644 index 0000000..c52fbb7 --- /dev/null +++ b/tests/daemon-idle-inflight.test.ts @@ -0,0 +1,75 @@ +// SPDX-FileCopyrightText: 2026 CoreWeave, Inc. +// SPDX-License-Identifier: MIT +// SPDX-PackageName: weave-claude-code + +// The daemon idles out after a quiet window, but the inactivity check only held +// it open for in-flight cross-session *team* work. A plain long-running tool or +// turn (longer than the timeout, with no other session active) tripped the +// timeout mid-flight: the daemon exited, dropped the still-open turn/tool spans, +// and the resumed work landed on a fresh, amnesiac daemon. +// +// The fix: also hold the daemon open while any session has an open turn span, a +// pending tool call, or a tracked subagent. + +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import * as fs from 'node:fs'; +import * as path from 'node:path'; + +import { startTestDaemon } from './helpers.ts'; + +function writeTranscript(home: string, sessionId: string): string { + const dir = path.join(home, '.claude', 'projects', 'test', sessionId); + fs.mkdirSync(dir, { recursive: true }); + const file = path.join(dir, `${sessionId}.jsonl`); + fs.writeFileSync(file, [ + JSON.stringify({ type: 'user', message: { role: 'user', content: [{ type: 'text', text: 'do work' }] } }), + JSON.stringify({ + type: 'assistant', + message: { + role: 'assistant', model: 'claude-opus-4-8', id: 'm1', + usage: { input_tokens: 10, output_tokens: 5, cache_read_input_tokens: 0, cache_creation_input_tokens: 0 }, + stop_reason: 'end_turn', content: [{ type: 'text', text: 'done' }], + }, + }), + ].join('\n') + '\n'); + return file; +} + +test('daemon stays up past the inactivity timeout while a turn span is open', async () => { + const d = await startTestDaemon({ env: { WEAVE_INACTIVITY_MS: '1000' } }); + try { + const sessionId = 'inflight-001'; + const transcript = writeTranscript(d.home, sessionId); + await d.send({ hook_event_name: 'SessionStart', session_id: sessionId, transcript_path: transcript }); + await d.send({ hook_event_name: 'UserPromptSubmit', session_id: sessionId, transcript_path: transcript, prompt: 'a long-running task' }); + + // Turn span is open and no further events arrive. Past the 1s timeout + // (checks fire every ~500ms) the daemon must log that it is holding open + // for in-flight work, and must NOT decide to shut down. + const stayedUp = await d.waitForLog(/work in flight — staying up/, 3000); + assert.ok(stayedUp, `daemon should hold open while a turn is in flight; log was:\n${d.readLog()}`); + assert.doesNotMatch(d.readLog(), /Inactivity timeout — shutting down/); + assert.equal(d.hasExited(), false, 'daemon should still be running'); + } finally { + await d.stop(); + } +}); + +test('daemon still idles out once the turn closes and nothing is in flight', async () => { + const d = await startTestDaemon({ env: { WEAVE_INACTIVITY_MS: '1000' } }); + try { + const sessionId = 'inflight-002'; + const transcript = writeTranscript(d.home, sessionId); + await d.send({ hook_event_name: 'SessionStart', session_id: sessionId, transcript_path: transcript }); + await d.send({ hook_event_name: 'UserPromptSubmit', session_id: sessionId, transcript_path: transcript, prompt: 'a quick task' }); + await d.send({ hook_event_name: 'Stop', session_id: sessionId, transcript_path: transcript }); + + // Turn span closed → nothing in flight → the daemon must still decide to + // idle out (the in-flight hold must not pin it open forever). + const shuttingDown = await d.waitForLog(/Inactivity timeout — shutting down/, 3500); + assert.ok(shuttingDown, `daemon should idle out after the turn closes; log was:\n${d.readLog()}`); + } finally { + await d.stop(); + } +}); diff --git a/tests/helpers.ts b/tests/helpers.ts index 9fb1345..2e36691 100644 --- a/tests/helpers.ts +++ b/tests/helpers.ts @@ -7,8 +7,10 @@ // (install-source-local.test.ts) needed the same helper. import * as fs from 'node:fs'; +import * as net from 'node:net'; +import * as os from 'node:os'; import * as path from 'node:path'; -import { spawn } from 'node:child_process'; +import { spawn, type ChildProcess } from 'node:child_process'; import { fileURLToPath } from 'node:url'; import { MARKETPLACE_NAME } from '../src/setup.ts'; @@ -91,3 +93,108 @@ export function writeKnownMarketplace(home: string, source: Record { + return new Promise((r) => setTimeout(r, ms)); +} + +/** Poll `pred` until it returns true or `timeoutMs` elapses. Resolves to the + * final value of `pred` (true if the condition was met, false on timeout). */ +export async function waitUntil(pred: () => boolean, timeoutMs = 3000): Promise { + const start = Date.now(); + while (Date.now() - start < timeoutMs) { + if (pred()) return true; + await delay(25); + } + return pred(); +} + +export interface TestDaemon { + home: string; + socketPath: string; + logPath: string; + proc: ChildProcess; + /** Open a connection, write one JSON payload, resolve when the socket closes. */ + send(payload: object): Promise; + readLog(): string; + /** Resolve true once the log matches `re`, false on timeout. */ + waitForLog(re: RegExp, timeoutMs?: number): Promise; + /** Resolve true once the daemon process has exited, false on timeout. */ + waitForExit(timeoutMs?: number): Promise; + hasExited(): boolean; + /** Kill the daemon (if alive) and remove its throwaway home. */ + stop(): Promise; +} + +/** + * Start a daemon in a throwaway home and wait until its socket is accepting. + * `opts.settings` is merged into the generated settings.json; `opts.env` into + * the daemon's environment (e.g. WEAVE_INACTIVITY_MS). + */ +export async function startTestDaemon( + opts: { settings?: Record; env?: Record } = {}, +): Promise { + const home = fs.mkdtempSync(path.join(os.homedir(), '.weave-daemontest-')); + const configDir = path.join(home, '.weave-claude-code'); + const socketPath = path.join(configDir, 'daemon.sock'); + const logPath = path.join(configDir, 'logs', 'daemon.log'); + fs.mkdirSync(path.join(configDir, 'logs'), { recursive: true }); + fs.writeFileSync( + path.join(configDir, 'settings.json'), + JSON.stringify({ + weave_project: 'test/test', + wandb_api_key: 'fake-key-for-test', + daemon_socket: socketPath, + log_file: logPath, + debug: true, + ...opts.settings, + }), + ); + + const proc = spawn(process.execPath, ['--import', 'tsx', CLI, 'daemon'], { + env: { ...process.env, HOME: home, WANDB_BASE_URL: 'http://127.0.0.1:1', ...opts.env }, + stdio: 'ignore', + }); + let exited = false; + proc.once('exit', () => { exited = true; }); + + const readLog = (): string => (fs.existsSync(logPath) ? fs.readFileSync(logPath, 'utf8') : ''); + const send = (payload: object): Promise => + new Promise((resolve, reject) => { + const s = net.createConnection(socketPath); + s.on('error', reject); + s.on('connect', () => s.end(JSON.stringify(payload))); + s.on('close', () => resolve()); + }); + + await waitUntil(() => fs.existsSync(socketPath), 5000); + await delay(150); // let listen() settle before the first send + + return { + home, + socketPath, + logPath, + proc, + send, + readLog, + waitForLog: (re, timeoutMs = 3000) => waitUntil(() => re.test(readLog()), timeoutMs), + waitForExit: (timeoutMs = 3000) => waitUntil(() => exited, timeoutMs), + hasExited: () => exited, + stop: async () => { + if (!exited) { + try { proc.kill('SIGKILL'); } catch { /* already gone */ } + await waitUntil(() => exited, 2000); + } + fs.rmSync(home, { recursive: true, force: true }); + }, + }; +}