feat(runtime): initialize and reconcile ordinary watchers

CHANGELOG.md

@@ -32,6 +32,8 @@ and adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ### New Features
 
+- Ordinary watcher startup can initialize a missing project index under daemon ownership and waits for fresh file indexing and queued synthesized edges before reporting readiness.
+
 - 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.

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. `serve --mcp --path <root> --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.
+The MCP launcher can replace a daemon from an older release when its hello confirms coordinated writer handover. `serve --mcp --path <root> --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. Adding `--initialize-index` lets the elected daemon create a missing exact-root index after acquiring writer ownership. 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:
 
@@ -850,11 +850,16 @@ session quiescence before cutover. See the [runtime control contract](site/src/c
 
 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.
+replacement. The elected daemon initializes a missing index at the exact root
+under writer ownership. Startup requests a fresh reconciliation even when reusing
+a watcher, and waits for file indexing and queued synthesized edges to complete.
+It returns `{ pid, version, projectRoot, watching: true }` only after verifying
+the exact root, build, active watcher and unchanged ownership records. An older,
+uncertain, direct or promotion holder blocks startup. Failed reconciliation or a
+timeout cannot return ready. The caller selects an absolute CLI path from the
+same validated build and its runtime; this operation does not install, promote
+or move artifacts. Legacy CLI initialization that ignores writer ownership
+remains outside this guard.
 
 **Embedding requirements**
 

__tests__/sync-resynthesis.test.ts

@@ -5,10 +5,11 @@
  * when neither end's file changed.
  */
 
-import { describe, it, expect, beforeEach, afterEach } from 'vitest';
+import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest';
 import * as fs from 'fs';
 import * as path from 'path';
 import * as os from 'os';
+import { syncBuiltinESMExports } from 'node:module';
 import CodeGraph from '../src/index';
 
 describe('synthesized edges on incremental sync', () => {
@@ -35,7 +36,10 @@ describe('synthesized edges on incremental sync', () => {
   });
 
   afterEach(() => {
+    vi.restoreAllMocks();
+    syncBuiltinESMExports();
     delete process.env.CODEGRAPH_SYNTH_REFRESH_MS;
+    delete process.env.CODEGRAPH_SYNC_RESYNTHESIS;
     try { cg.close(); } catch { /* ignore */ }
     fs.rmSync(dir, { recursive: true, force: true });
   });
@@ -83,4 +87,118 @@ describe('synthesized edges on incremental sync', () => {
     await new Promise((r) => setTimeout(r, 400));
     expect(callees('loadRepo')).toContain('GET https://api.github.com/repos/${…}');
   });
+
+  it('finishes queued watcher synthesis before a no-change complete reconciliation returns', async () => {
+    process.env.CODEGRAPH_SYNTH_REFRESH_MS = '60000';
+    write('src/api.ts', 'export async function loadRepo(owner: string) {\n  // deferred edit\n  return fetch(`https://api.github.com/repos/${owner}`);\n}\n');
+    await cg.sync({ deferSynthesis: true });
+    expect(callees('loadRepo')).not.toContain('GET https://api.github.com/repos/${…}');
+    const result = await cg.sync({ requireComplete: true });
+    expect(result.filesAdded + result.filesModified + result.filesRemoved).toBe(0);
+    expect(callees('loadRepo')).toContain('GET https://api.github.com/repos/${…}');
+  });
+
+  it('rejects incomplete synthesis and retries it on the next complete reconciliation', async () => {
+    process.env.CODEGRAPH_SYNTH_REFRESH_MS = '60000';
+    write('src/api.ts', 'export async function loadRepo(owner: string) {\n  // deferred edit\n  return fetch(`https://api.github.com/repos/${owner}`);\n}\n');
+    await cg.sync({ deferSynthesis: true });
+    const resolver = (cg as unknown as { resolver: { resynthesize: (...args: unknown[]) => Promise<number> } }).resolver;
+    const failure = vi.spyOn(resolver, 'resynthesize').mockRejectedValueOnce(new Error('injected synthesis failure'));
+    await expect(cg.sync({ requireComplete: true })).rejects.toThrow('did not complete synthesized-edge refresh');
+    expect(callees('loadRepo')).not.toContain('GET https://api.github.com/repos/${…}');
+    failure.mockRestore();
+    await cg.sync({ requireComplete: true });
+    expect(callees('loadRepo')).toContain('GET https://api.github.com/repos/${…}');
+  });
+
+  it('rejects a failed pending-marker write and retains queued work for a no-change retry', async () => {
+    process.env.CODEGRAPH_SYNTH_REFRESH_MS = '60000';
+    write('src/api.ts', 'export async function loadRepo(owner: string) {\n  // pending marker failure\n  return fetch(`https://api.github.com/repos/${owner}`);\n}\n');
+    await cg.sync({ deferSynthesis: true });
+    const queries = (cg as unknown as { queries: { setSynthesisPending: (pending: boolean) => void;
+      isSynthesisPending: () => boolean } }).queries;
+    expect(queries.isSynthesisPending()).toBe(false);
+    const marker = vi.spyOn(queries, 'setSynthesisPending').mockImplementationOnce(() => {
+      throw new Error('injected pending-marker failure');
+    });
+    await expect(cg.sync({ requireComplete: true })).rejects.toThrow('did not complete synthesized-edge refresh');
+    expect(queries.isSynthesisPending()).toBe(false);
+    marker.mockRestore();
+    await cg.sync({ requireComplete: true });
+    expect(callees('loadRepo')).toContain('GET https://api.github.com/repos/${…}');
+  });
+
+  it('serializes a complete refresh behind watcher indexing and flushes its deferred edges', async () => {
+    process.env.CODEGRAPH_SYNTH_REFRESH_MS = '60000';
+    write('src/api.ts', 'export async function loadRepo(owner: string) {\n  // queued watcher edit\n  return fetch(`https://api.github.com/repos/${owner}`);\n}\n');
+    const orchestrator = (cg as unknown as { orchestrator: { sync: (...args: unknown[]) => Promise<unknown> } }).orchestrator;
+    const original = orchestrator.sync.bind(orchestrator);
+    let release!: () => void;
+    let entered!: () => void;
+    const held = new Promise<void>(resolve => { release = resolve; });
+    const started = new Promise<void>(resolve => { entered = resolve; });
+    const sync = vi.spyOn(orchestrator, 'sync').mockImplementationOnce(async (...args) => {
+      entered();
+      await held;
+      return original(...args);
+    });
+    const watcher = cg.sync({ deferSynthesis: true });
+    await started;
+    const refresh = cg.sync({ requireComplete: true });
+    try { expect(sync).toHaveBeenCalledTimes(1); }
+    finally { release(); await Promise.all([watcher, refresh]); }
+    expect(sync).toHaveBeenCalledTimes(2);
+    expect(callees('loadRepo')).toContain('GET https://api.github.com/repos/${…}');
+  });
+
+  it('rejects an unreadable changed file and indexes it when a later refresh can read it', async () => {
+    const source = path.join(dir, 'src/handlers.ts');
+    write('src/handlers.ts', 'export function readableAfterRetry() {}\n');
+    const mutableFs = require('fs') as typeof fs;
+    const original = mutableFs.openSync;
+    const read = vi.spyOn(mutableFs, 'openSync').mockImplementation((file, ...args) => {
+      if (String(file) === source) throw Object.assign(new Error('injected read failure'), { code: 'EACCES' });
+      return Reflect.apply(original, fs, [file, ...args]);
+    });
+    syncBuiltinESMExports();
+    await expect(cg.sync({ requireComplete: true })).rejects.toThrow('failed to index 1 file');
+    expect(cg.getNodesByName('onSaved')).not.toHaveLength(0);
+    read.mockRestore();
+    syncBuiltinESMExports();
+    await cg.sync({ requireComplete: true });
+    expect(cg.getNodesByName('readableAfterRetry')).not.toHaveLength(0);
+    expect(cg.getNodesByName('onSaved')).toHaveLength(0);
+  });
+
+  it('refuses complete reconciliation while synthesized-edge refresh is disabled', async () => {
+    process.env.CODEGRAPH_SYNC_RESYNTHESIS = '0';
+    await expect(cg.sync({ requireComplete: true })).rejects.toThrow('requires synthesized-edge refresh');
+  });
+
+  it('acknowledges current wiring across every refresh-ending sequence of up to three events', async () => {
+    process.env.CODEGRAPH_SYNTH_REFRESH_MS = '60000';
+    const attached = "import { bus } from './bus';\nimport { onSaved } from './handlers';\nexport function wire() { bus.on('saved', onSaved); }\n";
+    const detached = "import { bus } from './bus';\nexport function wire() { return bus; }\n";
+    const events = ['attach', 'detach', 'watcher', 'refresh'] as const;
+    const failures: string[] = [];
+    const prefixes = [[], ...events.map(a => [a]), ...events.flatMap(a => events.map(b => [a, b]))];
+    for (const prefix of prefixes) {
+      write('src/wire.ts', attached);
+      await cg.sync({ requireComplete: true });
+      let wired = true;
+      const sequence = [...prefix, 'refresh'];
+      for (const event of sequence) {
+        if (event === 'attach' || event === 'detach') {
+          wired = event === 'attach';
+          write('src/wire.ts', wired ? attached : detached);
+        } else if (event === 'watcher') {
+          await cg.sync({ paths: ['src/wire.ts'], deferSynthesis: true });
+        } else {
+          await cg.sync({ requireComplete: true });
+          if (callees('publish').includes('onSaved') !== wired) failures.push(sequence.join(' -> '));
+        }
+      }
+    }
+    expect(failures).toEqual([]);
+  }, 20000);
 });

site/src/content/docs/reference/api.md

@@ -82,25 +82,35 @@ validated build. The same functions are exported from the package entry point.
 | `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?, 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 |
+| `checkRuntimeReady(root, identity, timeoutMs?, options?)` | The expected identity after hello, MCP initialization and status verification; `{requireWatcher: true}` requires the exact active watcher and writer ownership; adding `refresh: true` waits for a daemon-owned fresh reconciliation |
+| `startRuntimeWatcher(root, options)` | `{pid, version, projectRoot, watching: true}` after guarded initialization or reuse, fresh reconciliation 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 module and selected CLI must match `expectedVersion` before initialization.
+The elected daemon acquires writer ownership before creating a missing index 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.
+Disabled, failed and initializing watchers cannot return ready. Every successful
+startup waits for a fresh reconciliation, including when it reuses a watcher.
+File indexing and queued synthesized edges must complete before acknowledgment.
+The daemon and caller both reject changed ownership records. A reconciliation
+failure, unsupported refresh protocol or timeout fails startup without stopping
+the shared daemon. Complete reconciliation also refuses
+`CODEGRAPH_SYNC_RESYNTHESIS=0`. These guards do not fence legacy CLI initialization that
+ignores writer ownership.
 
 `serve --mcp --path <root> --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
+clients. Adding `--initialize-index`, or `{initializeIndex: true}` to this
+constructor, permits guarded initialization and requires preservation plus an
+explicit root. Ordinary startup launches its detached candidate through this
 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.

site/src/content/docs/reference/cli.md

@@ -33,6 +33,18 @@ codegraph help [command]          # Show help, optionally for one command
 
 The MCP server (`codegraph serve --mcp`) is launched automatically by your agent — you don't run it by hand. See [MCP Server](/codegraph/reference/mcp-server/).
 
+## serve
+
+`codegraph serve --mcp --path <root> --preserve-existing` preserves existing daemons and requires an index at that exact project root.
+
+Adding `--initialize-index` permits the elected daemon to create a missing index after acquiring writer ownership. It requires `--preserve-existing` and an explicit `--path`; no ancestor or child index is adopted.
+
+```bash
+codegraph serve --mcp --path /path/to/project --preserve-existing --initialize-index
+```
+
+Embedding callers that need a verified watcher and fresh reconciliation use `startRuntimeWatcher`. See the [runtime control contract](/codegraph/reference/api/#installer-runtime-control).
+
 ## init, index, and sync
 
 `codegraph init` creates the local `.codegraph/` directory **and** builds the full graph in one step. (The old `-i`/`--index` flag is now a no-op, accepted only so existing scripts don't break.) After that the file watcher keeps the graph current automatically — `index` (a full rebuild from scratch) and `sync` (an incremental update) are only needed when the watcher is disabled or you're scripting against the index outside an agent session.

src/bin/codegraph.ts

@@ -2323,8 +2323,11 @@ program
   .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)')
   .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;
+  .option('--initialize-index', 'With --preserve-existing, initialize the exact project index under daemon ownership')
+  .action(async (options: { path?: string; mcp?: boolean; watch?: boolean; preserveExisting?: boolean; initializeIndex?: boolean }) => {
+    const projectPath = options.path
+      ? options.preserveExisting ? path.resolve(options.path) : resolveProjectPath(options.path)
+      : undefined;
 
     // Commander sets watch=false when --no-watch is passed. Route it through
     // the same env-var chokepoint the watcher and MCP server already honor.
@@ -2352,7 +2355,8 @@ program
         }
         // Start MCP server - it handles initialization lazily based on rootUri from client
         const { MCPServer } = await import('../mcp/index');
-        const server = new MCPServer(projectPath, { preserveExisting: options.preserveExisting });
+        const server = new MCPServer(projectPath, { preserveExisting: options.preserveExisting,
+          initializeIndex: options.initializeIndex });
         await server.start();
         // Server will run until terminated
       } else {