From f82a0ea487f664e70a40556c3316829148cd769b Mon Sep 17 00:00:00 2001 From: Aaron Queen Date: Mon, 5 Oct 2026 18:02:18 -0600 Subject: [PATCH] feat(runtime): preserve writers during ordinary watcher startup --- CHANGELOG.md | 2 + README.md | 10 +- __tests__/mcp-projectpath-lifecycle.test.ts | 14 + __tests__/runtime-control.test.ts | 321 +++++++++++++++++++- __tests__/writer-lock.test.ts | 36 ++- site/src/content/docs/reference/api.md | 25 +- src/bin/codegraph.ts | 5 +- src/mcp/daemon-paths.ts | 8 +- src/mcp/daemon.ts | 17 +- src/mcp/engine.ts | 18 +- src/mcp/index.ts | 74 ++++- src/mcp/project-lifecycle.ts | 8 +- src/mcp/writer-lock.ts | 13 +- src/runtime-control.ts | 92 +++++- 14 files changed, 596 insertions(+), 47 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ffbe2678eb..07105cd850 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -32,6 +32,8 @@ and adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ### New Features +- Installers can start or reuse a checkout's active watcher without replacing existing writers, and MCP launchers can preserve existing daemons across reconnects with `--preserve-existing`. + - Installers can verify daemon readiness and safely hand writer ownership across build promotion and rollback through a supported runtime-control API. - `codegraph_explore` now finds quoted prose in script strings and template text through a capped source scan, ignoring case and punctuation without requiring a re-index. diff --git a/README.md b/README.md index a96e1fdbe3..78491ed77f 100644 --- a/README.md +++ b/README.md @@ -110,7 +110,7 @@ C# property accessors and expression-bodied properties contribute calls and refe `codegraph status` reports files that need re-indexing and files with recorded parse errors. `status --json` includes `index.filesNeedingReindex` and `index.filesWithParseErrors`; `files --json` includes each file's extraction errors. A transient parser failure preserves the previous graph and retries on the next sync. -The MCP launcher can replace a daemon from an older release when its hello confirms coordinated writer handover. Legacy daemons stay running while new sessions serve reads without auto-sync; stop the old MCP sessions and daemon, then reconnect with the current install. A daemon exits when its installation is deleted or its package version changes. Different managed builds of the same release retain the fork's version-identity checks. +The MCP launcher can replace a daemon from an older release when its hello confirms coordinated writer handover. `serve --mcp --path --preserve-existing` opts out of replacement on initial connection and reconnect, requires an index at that exact root, and keeps fallback reads without a watcher. Legacy daemons stay running while new sessions serve reads without auto-sync; stop the old MCP sessions and daemon, then reconnect with the current install. A daemon exits when its installation is deleted or its package version changes. Different managed builds of the same release retain the fork's version-identity checks. Dispatch and framework coverage the fork adds, by kind: @@ -848,6 +848,14 @@ process. Keep the coordination module available separately from the executable being replaced, including during rollback. Live legacy writers require explicit session quiescence before cutover. See the [runtime control contract](site/src/content/docs/reference/api.md#installer-runtime-control). +For ordinary checkout startup, `startRuntimeWatcher(root, { expectedVersion, +cliPath, runtimePath?, timeoutMs? })` reuses or elects a shared daemon without +replacement. It returns `{ pid, version, projectRoot, watching: true }` only after +verifying the exact root, build, active watcher and writer ownership. An older, +uncertain, direct or promotion holder blocks startup. A readable index alone is +insufficient. The caller selects an absolute CLI path from the same validated +build and its runtime; this operation does not install, promote or move artifacts. + **Embedding requirements** - Install from npm (`npm i @colbymchenry/codegraph`) so the matching diff --git a/__tests__/mcp-projectpath-lifecycle.test.ts b/__tests__/mcp-projectpath-lifecycle.test.ts index 10ea19abc6..03d54d70a4 100644 --- a/__tests__/mcp-projectpath-lifecycle.test.ts +++ b/__tests__/mcp-projectpath-lifecycle.test.ts @@ -110,6 +110,20 @@ describe('MCP explicit projectPath lifecycle (#1835)', { timeout: 30_000 }, () = finally { reader.close(); } } + it('preserves uncertain explicit-project ownership during activation and retries', async () => { + await engine.stop(); + engine = new MCPEngine({ preserveExisting: true }); + engines.push(engine); + const file = path.join(serviceA, '.codegraph', 'writer.pid'); + onOpen = () => fs.writeFileSync(file, 'uncertain'); + expect(await search(serviceA, 'alphaOriginal')).toContain('alphaOriginal'); + expect(opened).toHaveLength(1); + expect(opened[0]!.isWatching()).toBe(false); + await engine.getToolHandler().execute('codegraph_status', { projectPath: serviceA }); + expect(fs.readFileSync(file, 'utf8')).toBe('uncertain'); + expect(opened[0]!.isWatching()).toBe(false); + }); + /** * A daemon owner starts with idle exit OFF. Armed at startup, its 500ms * timer raced this process's first connection and lost on a loaded machine, diff --git a/__tests__/runtime-control.test.ts b/__tests__/runtime-control.test.ts index be9c3bec07..a1546f507c 100644 --- a/__tests__/runtime-control.test.ts +++ b/__tests__/runtime-control.test.ts @@ -1,29 +1,58 @@ -import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; 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, spawnSync, type ChildProcess } from 'node:child_process'; import { once } from 'node:events'; +import { syncBuiltinESMExports } from 'node:module'; import { getRuntimeIdentity, checkRuntimeReady, reserveRuntimeWriter, claimRuntimeWriter, releaseRuntimeWriter, - stopRuntime, type WriterReservation, + stopRuntime, startRuntimeWatcher, type WriterReservation, } from '../src/runtime-control'; import { readWriterLock, releaseWriterLock, markWriterReady, tryAcquireWriterLock } from '../src/mcp/writer-lock'; import { ToolHandler } from '../src/mcp/tools'; import { getDaemonPidPath, getDaemonSocketPath, encodeLockInfo } from '../src/mcp/daemon-paths'; +import { CodeGraph } from '../src'; +import { CodeGraphPackageVersion } from '../src/mcp/version'; +import childProcess from 'child_process'; +import { MCPServer } from '../src/mcp'; +import * as proxy from '../src/mcp/proxy'; +import * as daemonRegistry from '../src/mcp/daemon-registry'; +import { getTelemetry } from '../src/telemetry'; +import * as updateCheck from '../src/upgrade/update-check'; +import { getWriterPidPath } from '../src/mcp/writer-lock'; let root: string; let server: net.Server | null = null; let statusProjectPath: unknown; const actors: ChildProcess[] = []; +const detachedActors: number[] = []; +function spySpawn() { + const spy = vi.spyOn(childProcess, 'spawn'); + syncBuiltinESMExports(); + return spy; +} beforeEach(() => { statusProjectPath = null; root = fs.mkdtempSync(path.join(os.tmpdir(), 'cg-runtime-')); fs.mkdirSync(path.join(root, '.codegraph')); }); afterEach(async () => { - for (const actor of actors.splice(0)) { + vi.restoreAllMocks(); + syncBuiltinESMExports(); + vi.unstubAllEnvs(); + for (const pid of detachedActors.splice(0)) { + if (!daemonRegistry.isProcessAlive(pid)) continue; + process.kill(pid, 'SIGKILL'); + const deadline = Date.now() + 5000; + while (daemonRegistry.isProcessAlive(pid) && Date.now() < deadline) { + await new Promise((resolve) => setTimeout(resolve, 25)); + } + expect(daemonRegistry.isProcessAlive(pid)).toBe(false); + } + const winner = readWriterLock(root)?.pid; + for (const actor of actors.splice(0).sort((a, b) => Number(a.pid === winner) - Number(b.pid === winner))) { if (actor.exitCode === null && actor.signalCode === null) { const exited = once(actor, 'exit'); actor.kill('SIGKILL'); @@ -33,7 +62,7 @@ afterEach(async () => { if (server) await new Promise((resolve) => server!.close(() => resolve())); server = null; releaseWriterLock(root); - fs.rmSync(root, { recursive: true, force: true }); + fs.rmSync(root, { recursive: true, force: true, maxRetries: 5, retryDelay: 100 }); }); describe('runtime writer reservations', () => { @@ -137,14 +166,22 @@ describe('runtime writer reservations', () => { }); }); -async function fakeDaemon(options: { wrongHello?: boolean; wrongBuild?: boolean; statusError?: boolean; guidance?: boolean; closeEarly?: boolean; legacy?: boolean; changeRecord?: boolean } = {}): Promise { +async function fakeDaemon(options: { wrongHello?: boolean; wrongBuild?: boolean; statusError?: boolean; guidance?: boolean; closeEarly?: boolean; legacy?: boolean; changeRecord?: boolean; + version?: string; watcherRoot?: string; inactive?: boolean; initializing?: boolean; stopWatchingDuringStatus?: boolean; + changeWriterGeneration?: boolean; +} = {}): Promise { const guidance = options.guidance ? await new ToolHandler(null).execute('codegraph_status', {}) : null; const socketPath = getDaemonSocketPath(root); - const info = { pid: process.pid, version: 'test-build', socketPath, startedAt: Date.now() }; + const version = options.version ?? 'test-build'; + const info = { pid: process.pid, version, socketPath, startedAt: Date.now() }; + let watching = !options.inactive; fs.writeFileSync(getDaemonPidPath(root), encodeLockInfo(info)); server = net.createServer((socket) => { socket.setEncoding('utf8'); - socket.write(JSON.stringify({ protocol: 1, pid: process.pid, codegraph: options.wrongHello ? 'other' : 'test-build', ...(options.legacy ? {} : { writerProtocol: 1 }) }) + '\n'); + socket.write(JSON.stringify({ protocol: 1, pid: process.pid, codegraph: options.wrongHello ? 'other' : version, + ...(options.legacy ? {} : { writerProtocol: 1 }), + watcher: { projectRoot: options.watcherRoot ?? fs.realpathSync.native(root), active: watching, ready: !options.initializing }, + }) + '\n'); let buffer = ''; socket.on('data', (chunk) => { buffer += chunk; @@ -153,11 +190,14 @@ async function fakeDaemon(options: { wrongHello?: boolean; wrongBuild?: boolean; const msg = JSON.parse(buffer.slice(0, end)); buffer = buffer.slice(end + 1); if (msg.method === 'initialize') { if (options.closeEarly) { socket.end(); return; } - socket.write(JSON.stringify({ jsonrpc: '2.0', id: 1, result: { serverInfo: { version: options.wrongBuild ? 'other' : 'test-build' } } }) + '\n'); + socket.write(JSON.stringify({ jsonrpc: '2.0', id: 1, result: { serverInfo: { version: options.wrongBuild ? 'other' : version } } }) + '\n'); } else if (msg.id === 2) { statusProjectPath = msg.params?.arguments?.projectPath; if (options.changeRecord) fs.writeFileSync(getDaemonPidPath(root), encodeLockInfo({ ...info, version: 'other' })); - socket.write(JSON.stringify({ jsonrpc: '2.0', id: 2, result: { isError: !!options.statusError, content: [{ type: 'text', text: options.guidance ? guidance!.content[0].text : '**CodeGraph Status**\n**Server build:** test-build\n**Files indexed:** 1\n**Total nodes:** 2\n**Total edges:** 1' }] } }) + '\n'); + if (options.stopWatchingDuringStatus) watching = false; + if (options.changeWriterGeneration) fs.writeFileSync(getWriterPidPath(root), JSON.stringify({ + pid: process.pid, mode: 'daemon', ready: true, startedAt: 999 })); + socket.write(JSON.stringify({ jsonrpc: '2.0', id: 2, result: { isError: !!options.statusError, content: [{ type: 'text', text: options.guidance ? guidance!.content[0].text : `**CodeGraph Status**\n**Server build:** ${version}\n**Files indexed:** 1\n**Total nodes:** 2\n**Total edges:** 1` }] } }) + '\n'); } } }); @@ -166,6 +206,27 @@ async function fakeDaemon(options: { wrongHello?: boolean; wrongBuild?: boolean; } describe('runtime identity and readiness', () => { + it.each(['inactive', 'initializing', 'legacy', 'stopWatchingDuringStatus', 'changeRecord', 'changeWriterGeneration'] as const)( + 'never calls a readable daemon watcher-ready when %s', async (failure) => { + tryAcquireWriterLock(root, 'daemon'); + markWriterReady(root); + await fakeDaemon({ [failure]: true }); + await expect(checkRuntimeReady(root, { pid: process.pid, version: 'test-build' }, 1000, + { requireWatcher: true })).rejects.toThrow(); + }, + ); + + it('rejects a watcher from another checkout and preserves both ownership records', async () => { + tryAcquireWriterLock(root, 'daemon'); + markWriterReady(root); + await fakeDaemon({ watcherRoot: path.dirname(root) }); + const before = fs.readFileSync(getDaemonPidPath(root), 'utf8'); + const writer = readWriterLock(root); + await expect(checkRuntimeReady(root, { pid: process.pid, version: 'test-build' }, 1000, + { requireWatcher: true })).rejects.toThrow('exact project watcher'); + expect(fs.readFileSync(getDaemonPidPath(root), 'utf8')).toBe(before); + expect(readWriterLock(root)).toEqual(writer); + }); it('verifies a daemon build through hello, MCP initialization and status', async () => { await fakeDaemon(); expect(await getRuntimeIdentity(root)).toEqual({ pid: process.pid, version: 'test-build' }); @@ -211,3 +272,245 @@ describe('runtime identity and readiness', () => { expect(fs.readFileSync(getDaemonPidPath(root), 'utf8')).toBe(before); }); }); + +describe('ordinary watcher startup', () => { + const options = () => ({ expectedVersion: CodeGraphPackageVersion, + cliPath: path.resolve(__dirname, '../dist/bin/codegraph.js'), timeoutMs: 10000 }); + + async function initialize(): Promise { + const cg = await CodeGraph.init(root, { index: false }); + cg.close(); + } + + it('reuses a verified watcher without spawning or changing ownership', async () => { + await initialize(); + tryAcquireWriterLock(root, 'daemon'); + markWriterReady(root); + await fakeDaemon({ version: CodeGraphPackageVersion }); + const spawnProbe = spySpawn(); + const before = fs.readFileSync(getDaemonPidPath(root), 'utf8'); + expect(await startRuntimeWatcher(root, options())).toEqual({ pid: process.pid, + version: CodeGraphPackageVersion, projectRoot: fs.realpathSync.native(root), watching: true }); + expect(spawnProbe).not.toHaveBeenCalled(); + expect(fs.readFileSync(getDaemonPidPath(root), 'utf8')).toBe(before); + }); + + it.each(['legacy', 'direct', 'fallback', 'promotion', 'invalid', 'wrong-build'])( + 'preserves a %s holder rather than launching a second writer', async (holder) => { + await initialize(); + if (holder === 'legacy') fs.writeFileSync(getDaemonPidPath(root), String(process.pid)); + else if (holder === 'invalid') fs.writeFileSync(getDaemonPidPath(root), 'uncertain'); + else if (holder === 'wrong-build') await fakeDaemon(); + else if (holder === 'promotion') reserveRuntimeWriter(root, process.pid); + else tryAcquireWriterLock(root, holder); + const spawnProbe = spySpawn(); + const writer = readWriterLock(root); + await expect(startRuntimeWatcher(root, options())).rejects.toThrow(/preserved/); + expect(spawnProbe).not.toHaveBeenCalled(); + expect(readWriterLock(root)).toEqual(writer); + }, + ); + + it('fails compatibility and exact-root validation before any launch', async () => { + const spawnProbe = spySpawn(); + await expect(startRuntimeWatcher(root, { ...options(), expectedVersion: 'different' })).rejects.toThrow('build'); + await expect(startRuntimeWatcher(root, options())).rejects.toThrow('exact project root'); + expect(spawnProbe).not.toHaveBeenCalled(); + expect(readWriterLock(root)).toBeNull(); + }); + + it('does not borrow a parent or linked index for an exact checkout', async () => { + await initialize(); + const childRoot = path.join(root, 'child'); + fs.mkdirSync(childRoot); + const launch = spySpawn(); + await expect(startRuntimeWatcher(childRoot, options())).rejects.toThrow('exact project root'); + fs.symlinkSync(path.join(root, '.codegraph'), path.join(childRoot, '.codegraph'), 'junction'); + await expect(startRuntimeWatcher(childRoot, options())).rejects.toThrow('linked project index'); + expect(launch).not.toHaveBeenCalled(); + expect(readWriterLock(root)).toBeNull(); + }); + + it('refuses unknown loaded and expected build identities before candidate launch', () => { + const result = spawnSync(process.execPath, ['-e', ` + require(process.argv[1]).CodeGraphPackageVersion = '0.0.0-unknown'; + require('node:child_process').spawn = () => { throw new Error('candidate launch reached'); }; + require(process.argv[2]).startRuntimeWatcher(process.argv[3], JSON.parse(process.argv[4])) + .then(() => { console.error('unexpected readiness'); process.exit(1); }) + .catch(error => { console.log(error.message); }); + `, path.resolve(__dirname, '../dist/mcp/version.js'), path.resolve(__dirname, '../dist/runtime-control.js'), + root, JSON.stringify({ ...options(), expectedVersion: '0.0.0-unknown' })], { encoding: 'utf8', timeout: 10000 }); + expect(result.status, result.stderr).toBe(0); + expect(result.stdout).toContain('build is unavailable'); + expect(readWriterLock(root)).toBeNull(); + }); + + it('keeps a detached ordinary candidate usable when its initiating caller exits before readiness', async () => { + await initialize(); + const caller = spawn(process.execPath, ['-e', ` + const childProcess = require('node:child_process'); + const original = childProcess.spawn; + childProcess.spawn = (...args) => { + const candidate = original(...args); + process.send({pid:candidate.pid}, () => process.exit(0)); + return candidate; + }; + require(process.argv[1]).startRuntimeWatcher(process.argv[2], JSON.parse(process.argv[3])) + .catch(error => { console.error(error); process.exit(1); }); + `, path.resolve(__dirname, '../dist/runtime-control.js'), root, JSON.stringify(options())], + { stdio: ['ignore', 'ignore', 'pipe', 'ipc'] }); + actors.push(caller); + const exited = once(caller, 'exit'); + const [message] = await once(caller, 'message') as [{ pid: number }]; + detachedActors.push(message.pid); + await exited; + expect(caller.exitCode).toBe(0); + const deadline = Date.now() + 10000; + let identity; + do { + identity = await getRuntimeIdentity(root, { requireWatcher: true, timeoutMs: 500 }); + if (identity) break; + await new Promise((resolve) => setTimeout(resolve, 25)); + } while (Date.now() < deadline); + expect(identity?.pid).toBe(message.pid); + const launch = spySpawn(); + expect(await startRuntimeWatcher(root, options())).toMatchObject({ pid: message.pid, watching: true }); + expect(launch).not.toHaveBeenCalled(); + }); + + it('preserves an uncertain writer appearing after daemon election but before project activation', async () => { + await initialize(); + const candidate = spawn(process.execPath, ['-e', ` + const fs = require('node:fs'); + const { MCPEngine } = require(process.argv[1] + '/engine'); + const { Daemon, tryAcquireDaemonLock } = require(process.argv[1] + '/daemon'); + const root = process.argv[2], file = root + '/.codegraph/writer.pid'; + const initialize = MCPEngine.prototype.ensureInitialized; + MCPEngine.prototype.ensureInitialized = async function(...args) { + fs.writeFileSync(file, 'uncertain'); + await initialize.apply(this, args); + await this.getToolHandler().execute('codegraph_status', {}); + process.send({writer:fs.readFileSync(file, 'utf8'), watcher:this.getWatcherState()}); + }; + tryAcquireDaemonLock(root); + new Daemon(root, {preserveExisting:true, idleTimeoutMs:0}).start() + .catch(error => { console.error(error); process.exit(1); }); + `, path.resolve(__dirname, '../dist/mcp'), root], + { stdio: ['ignore', 'ignore', 'pipe', 'ipc'], env: { ...process.env, CODEGRAPH_QUERY_POOL_SIZE: '0' } }); + actors.push(candidate); + const [state] = await once(candidate, 'message'); + expect(state).toMatchObject({ writer: 'uncertain', watcher: { active: false } }); + expect(fs.readFileSync(getWriterPidPath(root), 'utf8')).toBe('uncertain'); + }); + + it('converges racing ordinary starters on one real watcher without a promotion lease', async () => { + await initialize(); + const original = childProcess.spawn; + const spawned: ChildProcess[] = []; + spySpawn().mockImplementation(((...args: Parameters) => { + const child = original(...args); + spawned.push(child); + actors.push(child); + return child; + }) as typeof spawn); + const results = await Promise.allSettled([startRuntimeWatcher(root, options()), startRuntimeWatcher(root, options())]); + const successes = results.flatMap((r) => r.status === 'fulfilled' ? [r.value] : []); + expect(successes.length).toBeGreaterThan(0); + expect(new Set(successes.map((r) => r.pid)).size).toBe(1); + const winner = successes[0]!; + expect(spawned.some((child) => child.pid === winner.pid)).toBe(true); + expect(readWriterLock(root)).toMatchObject({ pid: winner.pid, mode: 'daemon', ready: true }); + expect(reserveRuntimeWriter(root, process.pid)).toBeNull(); + await checkRuntimeReady(root, winner, 1000, { requireWatcher: true }); + }); + + it.each(['promotion', 'older', 'invalid-writer'])( + 'preserves a %s owner appearing after the empty-slot precheck, at real candidate election', async (event) => { + await initialize(); + const original = childProcess.spawn; + let before: string; + let protectedFile: string; + spySpawn().mockImplementation(((...args: Parameters) => { + if (event === 'promotion') { + reserveRuntimeWriter(root, process.pid); + protectedFile = getWriterPidPath(root); + } else if (event === 'older') { + protectedFile = getDaemonPidPath(root); + fs.writeFileSync(protectedFile, encodeLockInfo({ pid: process.pid, version: '0.0.1', + socketPath: getDaemonSocketPath(root), startedAt: 1 })); + } else { + protectedFile = getWriterPidPath(root); + fs.writeFileSync(protectedFile, 'uncertain'); + } + before = fs.readFileSync(protectedFile, 'utf8'); + const child = original(...args); + actors.push(child); + return child; + }) as typeof spawn); + await expect(startRuntimeWatcher(root, { ...options(), timeoutMs: 1000 })).rejects.toThrow('did not become ready'); + expect(actors).toHaveLength(1); + expect(fs.readFileSync(protectedFile!, 'utf8')).toBe(before!); + }, + ); + + it('times out with a disabled real watcher without terminating a possibly shared daemon', async () => { + await initialize(); + const original = childProcess.spawn; + spySpawn().mockImplementation(((command: string, args: string[], spawnOptions: childProcess.SpawnOptions) => { + const child = original(command, args, { ...spawnOptions, + env: { ...spawnOptions.env, CODEGRAPH_NO_WATCH: '1' } }); + actors.push(child); + return child; + }) as typeof spawn); + await expect(startRuntimeWatcher(root, { ...options(), timeoutMs: 1000 })).rejects.toThrow('did not become ready'); + expect(actors).toHaveLength(1); + expect(actors[0]!.exitCode).toBeNull(); + expect(readWriterLock(root)?.pid).toBe(actors[0]!.pid); + }); +}); + +describe('preserving MCP launch policy', () => { + it('preserves owners across all three-attempt orderings of older, unreachable, uncertain and promotion holders', async () => { + const cg = await CodeGraph.init(root, { index: false }); + cg.close(); + vi.stubEnv('CODEGRAPH_NO_DAEMON', '0'); + vi.stubEnv('CODEGRAPH_DAEMON_INTERNAL', '0'); + vi.spyOn(getTelemetry(), 'startInterval').mockImplementation(() => {}); + vi.spyOn(updateCheck, 'checkForUpdateInBackground').mockImplementation(() => {}); + const connect = vi.spyOn(proxy, 'connectWithHello'); + const stop = vi.spyOn(daemonRegistry, 'stopOlderDaemon'); + const launch = spySpawn(); + vi.spyOn(process.stderr, 'write').mockImplementation(() => true); + const events = ['older', 'unreachable', 'uncertain', 'promotion'] as const; + const failures: string[] = []; + let attempts = 0; + for (const a of events) for (const b of events) for (const c of events) { + const sequence = [a, b, c]; + vi.spyOn(proxy, 'runLocalHandshakeProxy').mockImplementation(async (deps) => { + for (const event of sequence) { + connect.mockResolvedValue(event === 'older' ? 'older-version' : null); + fs.writeFileSync(getDaemonPidPath(root), event === 'uncertain' ? 'uncertain' : encodeLockInfo({ + pid: process.pid, version: event === 'older' ? '0.0.1' : CodeGraphPackageVersion, + socketPath: getDaemonSocketPath(root), startedAt: Date.now(), + })); + fs.writeFileSync(getWriterPidPath(root), JSON.stringify({ pid: process.pid, + mode: event === 'promotion' ? 'promotion:held' : 'daemon', startedAt: 1, ready: false })); + const before = fs.readFileSync(getDaemonPidPath(root), 'utf8'); + const writer = fs.readFileSync(getWriterPidPath(root), 'utf8'); + const socket = await deps.getDaemonSocket(); + attempts++; + if (socket !== null || fs.readFileSync(getDaemonPidPath(root), 'utf8') !== before || + fs.readFileSync(getWriterPidPath(root), 'utf8') !== writer) failures.push(sequence.join(' -> ')); + } + const fallback = deps.makeEngine(); + expect(fallback.isReadOnly()).toBe(true); + await fallback.stop(); + }); + await new MCPServer(root, { preserveExisting: true }).start(); + } + expect(attempts).toBe(192); + expect(failures).toEqual([]); + expect(stop).not.toHaveBeenCalled(); + expect(launch).not.toHaveBeenCalled(); + }); +}); diff --git a/__tests__/writer-lock.test.ts b/__tests__/writer-lock.test.ts index c18268111b..1e2c0635cd 100644 --- a/__tests__/writer-lock.test.ts +++ b/__tests__/writer-lock.test.ts @@ -3,11 +3,13 @@ * stale-dead-pid / live-holder refusal. */ -import { afterEach, describe, expect, it } from 'vitest'; +import { afterEach, describe, expect, it, vi } from 'vitest'; import { spawn, type ChildProcess } from 'child_process'; import * as fs from 'fs'; import * as os from 'os'; import * as path from 'path'; +import filesystem from 'fs'; +import { syncBuiltinESMExports } from 'node:module'; import { MCPEngine } from '../src/mcp/engine'; import { decodeWriterLockInfo, @@ -23,6 +25,8 @@ describe('writer lock (#1740)', () => { let holder: ChildProcess | null = null; afterEach(() => { + vi.restoreAllMocks(); + syncBuiltinESMExports(); try { holder?.kill('SIGKILL'); } catch { /* already gone */ } holder = null; for (const child of foreignHolders) { @@ -61,6 +65,36 @@ describe('writer lock (#1740)', () => { releaseWriterLock(root); }); + it('preserves an uncertain writer record at ordinary candidate acquisition', () => { + const root = makeProject(); + const file = getWriterPidPath(root); + fs.writeFileSync(file, 'uncertain'); + expect(tryAcquireWriterLock(root, 'daemon', 'writer.pid', { preserveUncertain: true }).kind).toBe('taken'); + expect(fs.readFileSync(file, 'utf8')).toBe('uncertain'); + }); + + it.each(['uncertain', 'invalid-pid', 'same-pid-successor'])( + 'preserves a %s record arriving during stale-writer cleanup', (successor) => { + const root = makeProject(); + const file = getWriterPidPath(root); + const deadPid = 2147483646; + fs.writeFileSync(file, JSON.stringify({ pid: deadPid, mode: 'daemon', startedAt: 1 })); + const next = successor === 'uncertain' ? 'uncertain' : JSON.stringify({ + pid: successor === 'invalid-pid' ? 0 : deadPid, mode: 'direct', startedAt: 2, + }); + const original = filesystem.readFileSync; + let reads = 0; + vi.spyOn(filesystem, 'readFileSync').mockImplementation(((target: fs.PathOrFileDescriptor, ...args: unknown[]) => { + if (target === file && ++reads === 2) fs.writeFileSync(file, next); + return Reflect.apply(original, filesystem, [target, ...args]); + }) as typeof filesystem.readFileSync); + syncBuiltinESMExports(); + expect(tryAcquireWriterLock(root, 'daemon', 'writer.pid', { preserveUncertain: true }).kind).toBe('taken'); + expect(fs.readFileSync(file, 'utf8')).toBe(next); + expect(reads).toBeGreaterThanOrEqual(2); + }, + ); + it('reports taken when a live foreign pid holds the lock', () => { const root = makeProject(); // A foreign process that is genuinely alive, rather than a pid assumed to diff --git a/site/src/content/docs/reference/api.md b/site/src/content/docs/reference/api.md index 204ebd8bcd..9a71090b37 100644 --- a/site/src/content/docs/reference/api.md +++ b/site/src/content/docs/reference/api.md @@ -77,12 +77,33 @@ validated build. The same functions are exported from the package entry point. | Function | Result | |---|---| -| `getRuntimeIdentity(root)` | Hello-verified `{pid, version}`, or `null` | +| `getRuntimeIdentity(root, options?)` | Hello-verified `{pid, version}`, or `null`; `{timeoutMs, requireWatcher}` can require the exact active watcher in the hello | | `stopRuntime(root, options?)` | `{root, pid, outcome, version?}` with verified signalling | | `reserveRuntimeWriter(root, holderPid)` | Opaque JSON reservation for a live updater, or `null` when the slot is occupied | | `claimRuntimeWriter(root, reservation)` | `true` when this bootstrap process claims that exact reservation | | `releaseRuntimeWriter(root, reservation)` | `true` when the reservation is gone; preserves successor ownership | -| `checkRuntimeReady(root, identity, timeoutMs?)` | The expected identity after hello, MCP initialization and status verification | +| `checkRuntimeReady(root, identity, timeoutMs?, options?)` | The expected identity after hello, MCP initialization and status verification; `{requireWatcher: true}` additionally requires the exact active watcher and writer ownership | +| `startRuntimeWatcher(root, options)` | `{pid, version, projectRoot, watching: true}` after ordinary reuse/election and active-watcher verification | + +Ordinary startup accepts `{expectedVersion, cliPath, runtimePath?, timeoutMs?}`. +The caller supplies an absolute CLI path from the validated coordination build; +`runtimePath` defaults to the calling runtime, and `timeoutMs` defaults to 120000. +The module must match `expectedVersion` before launch. An index must already +exist at the exact canonical root: no ancestor or child index is adopted. +The operation never stops an existing daemon, claims a promotion reservation or +starts a fallback watcher. Concurrent starters use existing daemon election and +writer guards; they reuse a verified winner or refuse with preserved ownership. +Legacy, uncertain, wrong-build, direct and promotion owners block startup. +Disabled, failed and initializing watchers cannot return ready. + +`serve --mcp --path --preserve-existing` applies the same preservation +policy on every connection attempt, including reconnect and detached election. +It may serve fallback reads without auto-sync; this is not watcher readiness. +`new MCPServer(root, {preserveExisting: true})` selects this policy for library +clients. Ordinary startup launches its detached candidate through this existing +path. A timeout leaves a possibly shared elected daemon alone; its usual +client/idle lifecycle retires an unused daemon. Installation, immutable artifact +selection and service limits remain the caller's responsibility. Only `term`, `kill`, `not-running` and `no-daemon` stop outcomes allow a reservation attempt. `unverified`, `still-running` and `legacy-writer` require diff --git a/src/bin/codegraph.ts b/src/bin/codegraph.ts index 61adfbaf62..1915e8fc5b 100644 --- a/src/bin/codegraph.ts +++ b/src/bin/codegraph.ts @@ -2322,7 +2322,8 @@ program .option('-p, --path ', 'Project path (optional for MCP mode, uses rootUri from client)') .option('--mcp', 'Run as MCP server (stdio transport)') .option('--no-watch', 'Disable the file watcher (no auto-sync; useful on slow filesystems like WSL2 /mnt drives)') - .action(async (options: { path?: string; mcp?: boolean; watch?: boolean }) => { + .option('--preserve-existing', 'Never replace an existing writer; require an exact indexed project path') + .action(async (options: { path?: string; mcp?: boolean; watch?: boolean; preserveExisting?: boolean }) => { const projectPath = options.path ? resolveProjectPath(options.path) : undefined; // Commander sets watch=false when --no-watch is passed. Route it through @@ -2351,7 +2352,7 @@ program } // Start MCP server - it handles initialization lazily based on rootUri from client const { MCPServer } = await import('../mcp/index'); - const server = new MCPServer(projectPath); + const server = new MCPServer(projectPath, { preserveExisting: options.preserveExisting }); await server.start(); // Server will run until terminated } else { diff --git a/src/mcp/daemon-paths.ts b/src/mcp/daemon-paths.ts index 494364a4fe..e7e6430e4a 100644 --- a/src/mcp/daemon-paths.ts +++ b/src/mcp/daemon-paths.ts @@ -123,7 +123,7 @@ export function canProbeDaemonIdentity(info: DaemonLockInfo): boolean { * its socket. A bare PID liveness probe is insufficient because OSes reuse PIDs * after an OOM/SIGKILL (#1553). */ -export function probeDaemonIdentity(info: DaemonLockInfo, timeoutMs = 1_000, requireWriterProtocol = false): Promise { +export function probeDaemonIdentity(info: DaemonLockInfo, timeoutMs = 1_000, requireWriterProtocol = false, watcherRoot?: string): Promise { if (!canProbeDaemonIdentity(info)) return Promise.resolve(false); return new Promise((resolve) => { let socket: net.Socket; @@ -152,12 +152,14 @@ export function probeDaemonIdentity(info: DaemonLockInfo, timeoutMs = 1_000, req const newline = buffer.indexOf('\n'); if (newline < 0) return; try { - const hello = JSON.parse(buffer.slice(0, newline)) as Record; + const hello = JSON.parse(buffer.slice(0, newline)); finish( hello.protocol === 1 && hello.pid === info.pid && (info.version === 'unknown' || hello.codegraph === info.version) && - (!requireWriterProtocol || hello.writerProtocol === 1) + (!requireWriterProtocol || hello.writerProtocol === 1) && + (watcherRoot === undefined || (hello.watcher?.projectRoot === watcherRoot && + hello.watcher?.active === true && hello.watcher?.ready === true)) ); } catch { finish(false); diff --git a/src/mcp/daemon.ts b/src/mcp/daemon.ts index 6618887907..da4ec8cc73 100644 --- a/src/mcp/daemon.ts +++ b/src/mcp/daemon.ts @@ -43,6 +43,7 @@ import * as fs from 'fs'; import * as net from 'net'; import * as path from 'path'; +import { canonicalProjectRoot } from '../directory'; import { MCPEngine } from './engine'; import { MCPSession } from './session'; import { SocketTransport } from './transport'; @@ -56,6 +57,7 @@ import { } from './daemon-paths'; import { CodeGraphPackageJsonPath, CodeGraphPackageReleaseVersion, CodeGraphPackageVersion } from './version'; import { + readWriterLock, releaseWriterLock, swapWriterLock, tryAcquireWriterLock, @@ -173,6 +175,7 @@ export interface DaemonHello { socketPath: string; // echoed back so the proxy can log it protocol: 1; // bump if the hello shape changes writerProtocol?: 1; // ownership mutations use the OS coordination lock + watcher?: { projectRoot: string | null; active: boolean; ready: boolean }; } /** @@ -228,22 +231,24 @@ export class Daemon { private pidPath: string; /** The launcher holding the writer slot for this daemon, when it replaced an older one (#2335). */ private handoverFrom: number | null; + private preserveExisting: boolean; constructor( private projectRoot: string, - opts: { idleTimeoutMs?: number; maxIdleMs?: number; handoverFrom?: number | null } = {}, + opts: { idleTimeoutMs?: number; maxIdleMs?: number; handoverFrom?: number | null; preserveExisting?: boolean } = {}, ) { this.socketPath = getDaemonSocketPath(projectRoot); this.pidPath = getDaemonPidPath(projectRoot); this.idleTimeoutMs = opts.idleTimeoutMs ?? resolveIdleTimeoutMs(); this.maxIdleMs = opts.maxIdleMs ?? resolveMaxIdleMs(); this.handoverFrom = opts.handoverFrom ?? null; + this.preserveExisting = opts.preserveExisting ?? false; // Daemon mode serves many concurrent clients on one event loop, so off-load // read-tool dispatch to a worker pool — otherwise concurrent explores // serialize and starve the MCP transport (clients time out). Direct mode // (one stdio client) leaves the pool off; `CODEGRAPH_QUERY_POOL_SIZE=0` // disables it here too. - this.engine = new MCPEngine({ queryPool: true }); + this.engine = new MCPEngine({ queryPool: true, preserveExisting: this.preserveExisting }); this.engine.setProjectPathHint(projectRoot); } @@ -256,7 +261,8 @@ export class Daemon { // #1740: claim the project writer lock before opening/watching so a // concurrent direct-mode serve --mcp cannot start a second watcher. assertNoRebuild(this.projectRoot); - let writer = tryAcquireWriterLock(this.projectRoot, 'daemon'); + let writer = tryAcquireWriterLock(this.projectRoot, 'daemon', 'writer.pid', + { preserveUncertain: this.preserveExisting }); // The launcher that stopped an older daemon has held the slot for us since // (#2335); take it over without letting it fall free. if (writer.kind === 'taken' && this.handoverFrom !== null && @@ -449,12 +455,17 @@ export class Daemon { socket.once('close', () => this.sockets.delete(socket)); // Hello first so the proxy can verify versions before piping any // application bytes. The proxy reads exactly one line, then forwards. + const writer = readWriterLock(this.projectRoot); + const watcher = this.engine.getWatcherState(); const hello: DaemonHello = { codegraph: CodeGraphPackageVersion, pid: process.pid, socketPath: this.socketPath, protocol: 1, writerProtocol: 1, + watcher: { ...watcher, projectRoot: watcher.projectRoot ? canonicalProjectRoot(watcher.projectRoot) : null, + ready: writer?.pid === process.pid && + writer.mode === 'daemon' && writer.ready === true }, }; socket.write(JSON.stringify(hello) + '\n'); diff --git a/src/mcp/engine.ts b/src/mcp/engine.ts index f2daaf098a..c477d31322 100644 --- a/src/mcp/engine.ts +++ b/src/mcp/engine.ts @@ -34,6 +34,8 @@ const loadCodeGraph = (): typeof import('../index').default => const RETRY_SUBSCAN_TTL_MS = 5_000; export interface MCPEngineOptions { + /** Preserve uncertain writer records during project activation and retries. */ + preserveExisting?: boolean; /** Serve existing index contents without syncing, watching, or claiming a writer slot. */ readOnly?: boolean; /** @@ -94,7 +96,8 @@ export class MCPEngine { private queryPool: QueryPool | null = null; constructor(opts: MCPEngineOptions = {}) { - this.opts = { readOnly: opts.readOnly ?? false, watch: opts.watch ?? true, queryPool: opts.queryPool ?? false, queryPoolDefaultMax: opts.queryPoolDefaultMax }; + this.opts = { readOnly: opts.readOnly ?? false, watch: opts.watch ?? true, queryPool: opts.queryPool ?? false, + queryPoolDefaultMax: opts.queryPoolDefaultMax, preserveExisting: opts.preserveExisting ?? false }; this.toolHandler = new ToolHandler(null); this.toolHandler.setProjectLifecycle({ open: (root, open) => { @@ -103,7 +106,7 @@ export class MCPEngine { assertNoRebuild(root); if (this.opts.readOnly) return loadCodeGraph().openSync(root, { readOnly: true }); if (!this.opts.watch) return open(); - const lease = acquireProject(root, open, this.watchOptions()); + const lease = acquireProject(root, open, this.watchOptions(), this.opts.preserveExisting); this.explicitProjects.set(lease.cg, lease); return lease.cg; }, @@ -121,7 +124,8 @@ export class MCPEngine { }); if (opts.writerLockRoot && !this.opts.readOnly) { assertNoRebuild(opts.writerLockRoot); - const writer = tryAcquireWriterLock(opts.writerLockRoot, 'fallback'); + const writer = tryAcquireWriterLock(opts.writerLockRoot, 'fallback', 'writer.pid', + { preserveUncertain: this.opts.preserveExisting }); if (writer.kind === 'taken') { throw new Error(writerLockHeldMessage(writer.existing, writer.pidPath)); } @@ -178,6 +182,12 @@ export class MCPEngine { return this.projectPath; } + /** Actual default watcher state, never inferred from a readable index. */ + getWatcherState(): { projectRoot: string | null; active: boolean } { + return { projectRoot: this.cg?.getProjectRoot() ?? null, + active: !this.closed && !this.opts.readOnly && (this.cg?.isWatching() ?? false) }; + } + /** Shared ToolHandler — sessions delegate tool dispatch through this. */ getToolHandler(): ToolHandler { return this.toolHandler; @@ -426,7 +436,7 @@ export class MCPEngine { if (this.opts.readOnly || !this.cg || this.watcherStarted || !this.opts.watch) return; const opened = this.cg; - this.defaultLease = acquireProject(opened.getProjectRoot(), () => opened, this.watchOptions()); + this.defaultLease = acquireProject(opened.getProjectRoot(), () => opened, this.watchOptions(), this.opts.preserveExisting); this.cg = this.defaultLease.cg; if (this.cg !== opened) opened.close(); this.toolHandler.setDefaultCodeGraph(this.cg); diff --git a/src/mcp/index.ts b/src/mcp/index.ts index 1f0653c9e7..9f19df5212 100644 --- a/src/mcp/index.ts +++ b/src/mcp/index.ts @@ -36,8 +36,8 @@ import * as fs from 'fs'; import * as path from 'path'; -import { spawn, StdioOptions } from 'child_process'; -import { resolveServerRoot, getCodeGraphDir } from '../directory'; +import { spawn, StdioOptions, type ChildProcess } from 'child_process'; +import { resolveServerRoot, getCodeGraphDir, canonicalProjectRoot, isInitialized } from '../directory'; import { StdioTransport } from './transport'; import { MCPEngine } from './engine'; import { MCPSession } from './session'; @@ -279,8 +279,10 @@ function resolveDaemonRoot(explicitPath: string | null): string | null { * launcher proxies through the single winner. `handover` tells the daemon this * launcher holds the writer slot for it (see {@link DAEMON_HANDOVER_ENV}). */ -function spawnDetachedDaemon(root: string, handover = false): void { - const scriptPath = process.argv[1]; +export function spawnDetachedDaemon(root: string, handover = false, options: { + preserveExisting?: boolean; cliPath?: string; runtimePath?: string; +} = {}): ChildProcess { + const scriptPath = options.cliPath ?? process.argv[1]; if (!scriptPath) { // No resolvable CLI entry point to re-invoke — let the caller fall back to // direct mode rather than spawn something broken. @@ -304,8 +306,9 @@ function spawnDetachedDaemon(root: string, handover = false): void { if (handover) env[DAEMON_HANDOVER_ENV] = String(process.pid); else delete env[DAEMON_HANDOVER_ENV]; const child = spawn( - process.execPath, - [...process.execArgv, scriptPath, 'serve', '--mcp', '--path', root], + options.runtimePath ?? process.execPath, + [...(options.cliPath ? [] : process.execArgv), scriptPath, 'serve', '--mcp', '--path', root, + ...(options.preserveExisting ? ['--preserve-existing'] : [])], { detached: true, stdio, @@ -320,6 +323,7 @@ function spawnDetachedDaemon(root: string, handover = false): void { // and the session is served in-process. child.on('error', () => { /* no daemon — see above */ }); child.unref(); + return child; } finally { // The child holds its own dup of the log fd now; the launcher doesn't need it. if (logFd !== null) { @@ -433,7 +437,7 @@ export class MCPServer { /** Project root whose writer.pid we hold in direct mode (#1740); released on stop. */ private writerLockRoot: string | null = null; - constructor(projectPath?: string) { + constructor(projectPath?: string, private options: { preserveExisting?: boolean } = {}) { this.projectPath = projectPath || null; } @@ -473,10 +477,20 @@ export class MCPServer { // Direct mode if the user opted out. Setting the env var is sufficient to // get the pre-#411 single-process behavior. if (daemonOptOutSet()) { + if (this.options.preserveExisting) throw new Error('Preserving startup requires shared daemon mode.'); return this.startDirect('CODEGRAPH_NO_DAEMON set'); } const root = resolveDaemonRoot(this.projectPath); + if (this.options.preserveExisting && (!this.projectPath || !root || + canonicalProjectRoot(root) !== canonicalProjectRoot(this.projectPath) || !isInitialized(root))) { + throw new Error('Preserving startup requires an initialized index at the exact project root.'); + } + if (this.options.preserveExisting && root) { + for (const file of [getCodeGraphDir(root), path.join(getCodeGraphDir(root), 'codegraph.db')]) { + if (fs.lstatSync(file).isSymbolicLink()) throw new Error('Preserving startup refuses a linked project index.'); + } + } if (!root) { // No initialized project found — daemon mode has nowhere to put its // socket. The fresh-checkout / outside-project case; behave as before. @@ -493,6 +507,7 @@ export class MCPServer { await this.runProxyWithLocalHandshake(root); return; } catch (err) { + if (this.options.preserveExisting) throw err; // Belt-and-braces: a throw during proxy SETUP (before the client was served) // is still safe to recover from with a direct-mode session. const msg = err instanceof Error ? err.message : String(err); @@ -644,16 +659,29 @@ export class MCPServer { // In daemon mode stderr IS `.codegraph/daemon.log`; stamp every line so // kills/restarts can be placed in time (#1431 — the log was undatable). timestampStderrLines(); - const root = resolveDaemonRoot(this.projectPath) ?? this.projectPath ?? process.cwd(); + const root = this.options.preserveExisting + ? canonicalProjectRoot(this.projectPath ?? process.cwd()) + : resolveDaemonRoot(this.projectPath) ?? this.projectPath ?? process.cwd(); + if (this.options.preserveExisting && !isInitialized(root)) { + throw new Error('Preserving startup requires an initialized index at the exact project root.'); + } + if (this.options.preserveExisting) { + for (const file of [getCodeGraphDir(root), path.join(getCodeGraphDir(root), 'codegraph.db')]) { + if (fs.lstatSync(file).isSymbolicLink()) throw new Error('Preserving startup refuses a linked project index.'); + } + } // Read once and dropped, so nothing this daemon spawns inherits it. const handoverPid = Number(process.env[DAEMON_HANDOVER_ENV]); const handoverFrom = Number.isInteger(handoverPid) && handoverPid > 0 ? handoverPid : null; delete process.env[DAEMON_HANDOVER_ENV]; + if (this.options.preserveExisting && handoverFrom !== null) { + throw new Error('Preserving startup cannot claim a replacement handover.'); + } for (let attempt = 0; attempt < TAKEOVER_MAX_RETRIES; attempt++) { const lock = tryAcquireDaemonLock(root); if (lock.kind === 'acquired') { - const daemon = new Daemon(root, { handoverFrom }); + const daemon = new Daemon(root, { handoverFrom, preserveExisting: this.options.preserveExisting }); await daemon.start(); this.daemon = daemon; this.mode = 'daemon'; @@ -667,8 +695,14 @@ export class MCPServer { // Taken. If the holder is alive, another daemon already serves (or is // binding) — we're redundant; exit cleanly so the launcher proxies to it. const existing = lock.existing; + if (this.options.preserveExisting && existing && (!Number.isSafeInteger(existing.pid) || existing.pid <= 0)) { + throw new Error('Uncertain daemon ownership is preserved; refusing candidate takeover.'); + } let disprovedLiveIdentity = false; if (existing && existing.pid > 0 && isProcessAlive(existing.pid)) { + if (this.options.preserveExisting) { + throw new Error(`Existing daemon pid ${existing.pid} is preserved; refusing candidate takeover.`); + } // Give a newly-elected daemon time to bind, then require its socket hello // to match the lock PID/version. PID existence alone accepts an unrelated // process after OS PID reuse and permanently wedges startup (#1553). @@ -703,6 +737,9 @@ export class MCPServer { // ambiguous under the legacy lock format, so both cases fail closed. await clearStaleDaemonArtifacts(root); } else if (lock.lockContents !== null) { + if (this.options.preserveExisting && !existing) { + throw new Error('Uncertain daemon ownership is preserved; refusing candidate takeover.'); + } clearStaleDaemonLock(lock.pidPath, existing?.pid, { expectedLockContents: lock.lockContents, }); @@ -751,12 +788,24 @@ export class MCPServer { // before #2278 named their pipe after the root as typed). Either way it // makes way for one from this install (#2335). const answered = probe === 'older-version'; - const older = await replaceOlderDaemon(root, answered); + if (this.options.preserveExisting) { + if (probe) return null; + let existing: ReturnType = null; + try { + existing = decodeLockInfo(fs.readFileSync(getDaemonPidPath(root), 'utf8')); + if (!existing || isProcessAlive(existing.pid)) return null; + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ENOENT') return null; + } + const writer = readWriterLock(root); + if (writer && isProcessAlive(writer.pid)) return null; + } + const older = this.options.preserveExisting ? 'none' : await replaceOlderDaemon(root, answered); if (older === 'kept' && answered) return null; // None reachable — spawn one (detached) and poll for its bind. After a // replacement this launcher holds the writer slot for it meanwhile. try { - spawnDetachedDaemon(root, older === 'stopped'); + spawnDetachedDaemon(root, older === 'stopped', this.options); for (let attempt = 0; attempt < DAEMON_CONNECT_MAX_RETRIES; attempt++) { await sleep(DAEMON_CONNECT_RETRY_DELAY_MS); const s = await connectAnyCandidate(); @@ -771,7 +820,8 @@ export class MCPServer { if (older === 'stopped') releaseWriterLock(root); } }; - await runLocalHandshakeProxy({ getDaemonSocket, makeEngine: () => makeFallbackEngine(root), root }); + await runLocalHandshakeProxy({ getDaemonSocket, makeEngine: () => this.options.preserveExisting + ? readOnlyFallback('preserving startup did not verify a shared daemon') : makeFallbackEngine(root), root }); } /** Standard SIGINT/SIGTERM handlers that route to our `stop()` (direct mode). */ diff --git a/src/mcp/project-lifecycle.ts b/src/mcp/project-lifecycle.ts index e9600eb406..453ef26fcf 100644 --- a/src/mcp/project-lifecycle.ts +++ b/src/mcp/project-lifecycle.ts @@ -20,6 +20,7 @@ interface Project { socket: Socket | null; timer: NodeJS.Timeout; options: Parameters[0]; + preserveUncertain: boolean; } export interface ProjectLease { @@ -35,18 +36,20 @@ export function acquireProject( root: string, open: () => CodeGraph, options: Parameters[0], + preserveUncertain = false, ): ProjectLease { root = realpathSync(root); const key = canonicalProjectRoot(root); let project = projects.get(key); if (!project) { const cg = open(); - project = { key, cg, refs: 0, owner: false, caughtUp: false, retirement: null, gate: null, socket: null, options, + project = { key, cg, refs: 0, owner: false, caughtUp: false, retirement: null, gate: null, socket: null, options, preserveUncertain, timer: setInterval(() => { void ready(root, project!); }, 1000) }; project.timer.unref(); projects.set(key, project); } const entry = project; + entry.preserveUncertain ||= preserveUncertain; entry.refs++; let released = false; return { @@ -111,7 +114,8 @@ function ready(root: string, project: Project): Promise { async function activate(root: string, project: Project): Promise { if (!isInitialized(root)) return; - const writer = tryAcquireWriterLock(root, 'fallback'); + const writer = tryAcquireWriterLock(root, 'fallback', 'writer.pid', + { preserveUncertain: project.preserveUncertain }); if (writer.kind === 'taken') { // Keep a real daemon session, so its idle timeout cannot strand this reader. // Direct writers have no socket; the periodic retry takes over on their exit. diff --git a/src/mcp/writer-lock.ts b/src/mcp/writer-lock.ts index f545a8fc55..289d5f463f 100644 --- a/src/mcp/writer-lock.ts +++ b/src/mcp/writer-lock.ts @@ -116,10 +116,11 @@ export function tryAcquireWriterLock( projectRoot: string, mode: string, lockName: 'writer.pid' | 'rebuild.pid' = 'writer.pid', + options: { preserveUncertain?: boolean } = {}, ): WriterAcquireResult { return acquireWriterRecord(projectRoot, { pid: process.pid, mode, startedAt: Date.now(), ready: false, - }, lockName); + }, lockName, true, options.preserveUncertain); } /** Reserve for a live external updater. A unique mode distinguishes repeated leases by one PID. */ @@ -137,6 +138,7 @@ function acquireWriterRecord( info: WriterLockInfo, lockName: 'writer.pid' | 'rebuild.pid', reuseOwnRecord = true, + preserveUncertain = false, ): WriterAcquireResult { const pidPath = getWriterPidPath(projectRoot, lockName); fs.mkdirSync(path.dirname(pidPath), { recursive: true }); @@ -146,6 +148,7 @@ function acquireWriterRecord( try { if (!isProcessAlive(info.pid)) return { kind: 'taken', existing: readWriterLock(projectRoot, lockName), pidPath }; + let observedRaw: string | null = null; const attempt = (): WriterAcquireResult => { const tmp = `${pidPath}.${process.pid}.tmp`; let acquired = false; @@ -184,7 +187,8 @@ function acquireWriterRecord( let existing: WriterLockInfo | null = null; try { - existing = decodeWriterLockInfo(fs.readFileSync(pidPath, 'utf8')); + observedRaw = fs.readFileSync(pidPath, 'utf8'); + existing = decodeWriterLockInfo(observedRaw); } catch { /* unreadable */ } return { kind: 'taken', existing, pidPath }; }; @@ -197,11 +201,16 @@ function acquireWriterRecord( } if (result.kind === 'taken') { const existing = result.existing; + if (preserveUncertain && (!existing || !Number.isSafeInteger(existing.pid) || existing.pid <= 0)) return result; if (!existing || existing.pid <= 0 || !isProcessAlive(existing.pid)) { // Stale — clear (pid-verified) and retry once. try { const raw = fs.readFileSync(pidPath, 'utf8'); const cur = decodeWriterLockInfo(raw); + if (preserveUncertain && (raw !== observedRaw || !cur || + !Number.isSafeInteger(cur.pid) || cur.pid <= 0)) { + return { kind: 'taken', existing: cur, pidPath }; + } if (!cur || cur.pid === existing?.pid) { if (!cur || cur.pid <= 0 || !isProcessAlive(cur.pid)) { fs.unlinkSync(pidPath); diff --git a/src/runtime-control.ts b/src/runtime-control.ts index 083d09886a..82875af128 100644 --- a/src/runtime-control.ts +++ b/src/runtime-control.ts @@ -1,17 +1,20 @@ -/** Supported daemon control for installers. Process launch and artifact swaps belong to the caller. */ +/** Supported daemon control and ordinary startup. Artifact selection and swaps belong to the caller. */ import * as fs from 'fs'; import * as net from 'net'; -import { canonicalProjectRoot } from './directory'; +import * as path from 'path'; +import { canonicalProjectRoot, isInitialized, getCodeGraphDir } from './directory'; import { getDaemonPidPath, decodeLockInfo, probeDaemonIdentity } from './mcp/daemon-paths'; import { isProcessAlive, stopDaemonAt, type StopResult } from './mcp/daemon-registry'; -import { reserveWriterLock, updateWriterLock, type WriterLockInfo } from './mcp/writer-lock'; +import { reserveWriterLock, updateWriterLock, readWriterLock, getWriterPidPath, type WriterLockInfo } from './mcp/writer-lock'; +import { CodeGraphPackageVersion } from './mcp/version'; export const RUNTIME_CONTROL_PROTOCOL = 1; export interface RuntimeIdentity { pid: number; version: string } +export interface RuntimeWatcher extends RuntimeIdentity { projectRoot: string; watching: true } /** Opaque, JSON-serializable reservation. Retain the exact value until claim or release. */ export interface WriterReservation { pid: number; mode: string; startedAt: number } -export async function getRuntimeIdentity(projectRoot: string): Promise { +export async function getRuntimeIdentity(projectRoot: string, options: { timeoutMs?: number; requireWatcher?: boolean } = {}): Promise { projectRoot = canonicalProjectRoot(projectRoot); let raw: string; try { raw = fs.readFileSync(getDaemonPidPath(projectRoot), 'utf8'); } @@ -20,7 +23,8 @@ export async function getRuntimeIdentity(projectRoot: string): Promise { projectRoot = canonicalProjectRoot(projectRoot); if (!Number.isFinite(timeoutMs) || timeoutMs <= 0) throw new Error('Readiness timeout must be positive.'); + const deadline = Date.now() + timeoutMs; const raw = fs.readFileSync(getDaemonPidPath(projectRoot), 'utf8'); + const writerRaw = options.requireWatcher ? fs.readFileSync(getWriterPidPath(projectRoot), 'utf8') : null; const info = decodeLockInfo(raw); if (!info || info.pid !== expected.pid || info.version !== expected.version) { throw new Error('Daemon identity changed before readiness.'); @@ -112,6 +119,11 @@ export async function checkRuntimeReady( if (msg.protocol !== 1 || msg.pid !== expected.pid || msg.codegraph !== expected.version) { return finish(new Error('Daemon identity changed during readiness.')); } + if (options.requireWatcher && (msg.writerProtocol !== 1 || + msg.watcher?.projectRoot !== projectRoot || msg.watcher?.active !== true || + msg.watcher?.ready !== true)) { + return finish(new Error('The exact project watcher is not ready.')); + } phase = 'initialize'; socket.write(JSON.stringify({ jsonrpc: '2.0', id: 1, method: 'initialize', params: { protocolVersion: '2024-11-05', capabilities: {}, clientInfo: { name: 'codegraph-runtime-control', version: '1' }, @@ -142,9 +154,77 @@ export async function checkRuntimeReady( } }); }); - const after = await getRuntimeIdentity(projectRoot); + if (Date.now() >= deadline) throw new Error('MCP readiness timed out.'); + const after = await getRuntimeIdentity(projectRoot, { timeoutMs: deadline - Date.now(), requireWatcher: options.requireWatcher }); if (after?.pid !== expected.pid || after.version !== expected.version) { throw new Error('Daemon identity changed after readiness.'); } + if (options.requireWatcher) { + const writer = readWriterLock(projectRoot); + if (writer?.pid !== expected.pid || writer.mode !== 'daemon' || writer.ready !== true || + fs.readFileSync(getDaemonPidPath(projectRoot), 'utf8') !== raw || + fs.readFileSync(getWriterPidPath(projectRoot), 'utf8') !== writerRaw) { + throw new Error('Watcher ownership changed after readiness.'); + } + } return expected; } + +/** Reuse or elect an exact-checkout watcher without replacement or promotion. */ +export async function startRuntimeWatcher(projectRoot: string, options: { + expectedVersion: string; cliPath: string; runtimePath?: string; timeoutMs?: number; +}): Promise { + projectRoot = canonicalProjectRoot(projectRoot); + const timeoutMs = options.timeoutMs ?? 120_000; + if (!Number.isFinite(timeoutMs) || timeoutMs <= 0) throw new Error('Readiness timeout must be positive.'); + if (CodeGraphPackageVersion === '0.0.0-unknown' || options.expectedVersion !== CodeGraphPackageVersion) { + throw new Error('Runtime-control build is unavailable or does not match the expected watcher build.'); + } + if (!path.isAbsolute(options.cliPath) || !fs.statSync(options.cliPath).isFile() || + (options.runtimePath && !path.isAbsolute(options.runtimePath))) { + throw new Error('Watcher startup requires an absolute CLI and runtime path.'); + } + if (!isInitialized(projectRoot)) throw new Error('Watcher startup requires an index at the exact project root.'); + for (const file of [getCodeGraphDir(projectRoot), path.join(getCodeGraphDir(projectRoot), 'codegraph.db')]) { + if (fs.lstatSync(file).isSymbolicLink()) throw new Error('Watcher startup refuses a linked project index.'); + } + const deadline = Date.now() + timeoutMs; + let launched = false; + let launchError: Error | undefined; + let lastError: unknown; + do { + const identity = await getRuntimeIdentity(projectRoot, { timeoutMs: Math.max(1, deadline - Date.now()) }); + if (identity) { + if (identity.version !== options.expectedVersion) throw new Error('Existing daemon build is preserved; watcher startup is blocked.'); + try { + await checkRuntimeReady(projectRoot, identity, Math.max(1, deadline - Date.now()), { requireWatcher: true }); + return { ...identity, projectRoot, watching: true }; + } catch (error) { lastError = error; } + } else if (!launched) { + // A failed hello is not an empty slot. Existing guards arbitrate the race + // after this read; the candidate never clears live or uncertain owners. + for (const file of [getDaemonPidPath(projectRoot), getWriterPidPath(projectRoot)]) { + try { + const raw = fs.readFileSync(file, 'utf8'); + const owner = file === getDaemonPidPath(projectRoot) ? decodeLockInfo(raw) : readWriterLock(projectRoot); + if (!owner || !Number.isSafeInteger(owner.pid) || owner.pid <= 0 || isProcessAlive(owner.pid)) { + throw new Error('Existing or uncertain writer is preserved; watcher startup is blocked.'); + } + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error; + } + } + const { spawnDetachedDaemon } = await import('./mcp'); + const child = spawnDetachedDaemon(projectRoot, false, { preserveExisting: true, + cliPath: options.cliPath, runtimePath: options.runtimePath }); + child.once('error', (error) => { launchError = error; }); + launched = true; + } + if (launchError) throw launchError; + if (Date.now() >= deadline) break; + await new Promise((resolve) => setTimeout(resolve, Math.min(25, deadline - Date.now()))); + } while (Date.now() < deadline); + // An elected daemon can already serve other clients. Its idle lifecycle, + // rather than this caller's timeout, owns retirement of an unused candidate. + throw new Error(`Watcher startup did not become ready: ${lastError instanceof Error ? lastError.message : 'deadline exceeded'}`); +}