From ec0556a66ef197d696ec14ecd5fa1454a9ecfdb9 Mon Sep 17 00:00:00 2001 From: Garry Tan Date: Sat, 18 Apr 2026 13:35:51 +0800 Subject: [PATCH] feat(autopilot): Minions dispatch + worker spawn supervisor + async shutdown MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Autopilot now dispatches each cycle as a single `autopilot-cycle` Minion job (with idempotency_key on the cycle slot) instead of running steps inline. A forked `gbrain jobs work` child drains the queue durably, supervised by autopilot. The user runs ONE install step (`gbrain autopilot --install`) and gets sync + extract + embed + backlinks + durable job processing, with no separate worker daemon to manage. Mode selection: - minion_mode=always OR pain_triggered (default), engine=postgres → Minions dispatch. Spawn child, submit autopilot-cycle each interval. - minion_mode=off, OR engine=pglite, OR `--inline` flag → run steps inline in-process, same as pre-v0.11.1. PGLite has an exclusive file lock that blocks a second worker process, so the inline path is the only path that works there. Worker supervision: - spawn(resolveGbrainCliPath(), ['jobs', 'work'], { stdio: 'inherit' }). stdio:'inherit' avoids pipe-buffer blocking (Codex architecture #2). - On worker exit: 10s backoff + restart. Crash counter caps at 5 → autopilot stops with a clear error. - resolveGbrainCliPath() prefers argv[1] (cli.ts / /gbrain), then process.execPath (compiled binary suffix check), then `which gbrain` (installed to $PATH). NEVER blindly uses process.execPath, which on source installs is the Bun runtime, not `gbrain` (Codex architecture #1). Shutdown: - Async SIGTERM/SIGINT handler: sends SIGTERM to worker, awaits its exit for up to 35s (the worker's own drain is 30s; we add buffer for signal-delivery latency), then SIGKILL if still alive. - Drops the old `process.on('exit')` lock-cleanup handler — its callback runs synchronously and can't wait for the worker drain. Lock file cleanup moved inside the async shutdown. Lock-file mtime refresh every cycle (Codex C) so a long-lived autopilot doesn't get declared "stale" by the next cron-fired invocation after 10 minutes. Inline fallback path calls the new Core fns (runExtractCore, runEmbedCore) instead of the CLI wrappers. That way a bad arg from inside the loop can't process.exit() the autopilot itself (matches Codex #5). test/autopilot-resolve-cli.test.ts: 3 tests covering argv[1]-as-gbrain, argv[1]-as-cli.ts, and graceful error when no path resolves. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/commands/autopilot.ts | 189 ++++++++++++++++++++++++----- test/autopilot-resolve-cli.test.ts | 53 ++++++++ 2 files changed, 212 insertions(+), 30 deletions(-) create mode 100644 test/autopilot-resolve-cli.test.ts diff --git a/src/commands/autopilot.ts b/src/commands/autopilot.ts index f97a6afbb..a6fecaa30 100644 --- a/src/commands/autopilot.ts +++ b/src/commands/autopilot.ts @@ -1,20 +1,28 @@ /** * gbrain autopilot — Self-maintaining brain daemon. * - * Runs: sync → extract → embed → backlinks fix in a continuous loop. - * Health-based adaptive scheduling. Best-effort per step. + * v0.11.1 shape: + * - Default path (minion_mode != off AND engine == postgres): spawn a + * `gbrain jobs work` child process, submit ONE `autopilot-cycle` job + * per interval with an idempotency_key so slow cycles don't stack up. + * The forked worker drains the queue durably; restart with 10s backoff + * on crash (5-crash cap → autopilot stops with a clear error). + * - Fallback (minion_mode=off, PGLite, or `--inline`): run sync → + * extract → embed inline, same as pre-v0.11.1 behavior. * * Usage: - * gbrain autopilot [--repo ] [--interval N] [--json] + * gbrain autopilot [--repo ] [--interval N] [--json] [--inline] * gbrain autopilot --install [--repo ] * gbrain autopilot --uninstall * gbrain autopilot --status [--json] */ -import { existsSync, readFileSync, writeFileSync, mkdirSync, appendFileSync } from 'fs'; +import { existsSync, readFileSync, writeFileSync, mkdirSync, appendFileSync, utimesSync, unlinkSync } from 'fs'; import { join } from 'path'; -import { execSync } from 'child_process'; +import { execSync, spawn, type ChildProcess } from 'child_process'; import type { BrainEngine } from '../core/engine.ts'; +import { loadPreferences } from '../core/preferences.ts'; +import { loadConfig } from '../core/config.ts'; function parseArg(args: string[], flag: string): string | undefined { const idx = args.indexOf(flag); @@ -33,6 +41,35 @@ function logError(phase: string, e: unknown) { } catch { /* best-effort */ } } +/** + * Resolve the gbrain CLI entrypoint for spawning the worker child. + * + * Codex caught the bug in earlier plan drafts: `process.execPath` is the + * Bun (or Node) runtime binary on source installs, not `gbrain`. Blindly + * using it would spawn `bun jobs work`, which does not work. + * + * Order of resolution: + * 1. argv[1] if it clearly points at a gbrain entry (cli.ts or /gbrain). + * 2. process.execPath when running as the compiled binary. + * 3. `which gbrain` for installs where the binary is on $PATH. + * 4. Throw — nothing on $PATH, no way to supervise the worker. + */ +export function resolveGbrainCliPath(): string { + const arg1 = process.argv[1] ?? ''; + if (arg1.endsWith('/gbrain') || arg1.endsWith('/cli.ts') || arg1.endsWith('\\gbrain.exe')) { + return arg1; + } + const exec = process.execPath ?? ''; + if (exec.endsWith('/gbrain') || exec.endsWith('\\gbrain.exe')) { + return exec; + } + try { + const which = execSync('which gbrain', { encoding: 'utf-8', stdio: ['ignore', 'pipe', 'ignore'] }).trim(); + if (which) return which; + } catch { /* not on $PATH */ } + throw new Error('Could not resolve the gbrain CLI path. Install gbrain so it is on $PATH, or run autopilot from the compiled binary directly.'); +} + export async function runAutopilot(engine: BrainEngine, args: string[]) { if (args.includes('--help') || args.includes('-h')) { console.log('Usage: gbrain autopilot [--repo ] [--interval N] [--json]\n gbrain autopilot --install [--repo ]\n gbrain autopilot --uninstall\n gbrain autopilot --status [--json]\n\nSelf-maintaining brain daemon. Runs sync + extract + embed + backlinks in a loop.'); @@ -55,6 +92,7 @@ export async function runAutopilot(engine: BrainEngine, args: string[]) { const repoPath = parseArg(args, '--repo') || await engine.getConfig('sync.repo_path'); const baseInterval = parseInt(parseArg(args, '--interval') || '300', 10); const jsonMode = args.includes('--json'); + const forceInline = args.includes('--inline'); if (!repoPath) { console.error('No repo path. Use --repo or run gbrain sync --repo first.'); @@ -79,12 +117,69 @@ export async function runAutopilot(engine: BrainEngine, args: string[]) { console.log(`Autopilot starting. Repo: ${repoPath}, interval: ${baseInterval}s`); - // Signal handling + lock cleanup + // Mode resolution: Minions dispatch when the user has opted in AND the + // worker daemon can actually run (Postgres only; PGLite's exclusive file + // lock blocks a separate worker process). + const mode = loadPreferences().minion_mode ?? 'pain_triggered'; + const cfg = loadConfig(); + const engineType = cfg?.engine ?? 'pglite'; + const useMinionsDispatch = mode !== 'off' && engineType === 'postgres' && !forceInline; + let stopping = false; - const cleanup = () => { try { require('fs').unlinkSync(lockPath); } catch {} }; - process.on('exit', cleanup); - process.on('SIGTERM', () => { stopping = true; console.log('Autopilot stopping (SIGTERM).'); }); - process.on('SIGINT', () => { stopping = true; console.log('Autopilot stopping (SIGINT).'); }); + let workerProc: ChildProcess | null = null; + let crashCount = 0; + + if (useMinionsDispatch) { + const cliPath = resolveGbrainCliPath(); + const startWorker = () => { + const child = spawn(cliPath, ['jobs', 'work'], { stdio: 'inherit', env: process.env }); + workerProc = child; + console.log(`[autopilot] Minions worker spawned (pid: ${child.pid})`); + child.on('exit', (code) => { + workerProc = null; + if (stopping) return; + if (crashCount >= 5) { + console.error('[autopilot] 5 consecutive worker crashes, giving up.'); + process.exit(1); + } + crashCount++; + console.error(`[autopilot] worker exited code=${code}, restart #${crashCount} in 10s`); + setTimeout(startWorker, 10_000); + }); + }; + startWorker(); + } else { + const why = mode === 'off' ? 'minion_mode=off' + : (engineType !== 'postgres' ? 'engine=pglite' : 'flag=--inline'); + console.log(`[autopilot] running steps inline (${why})`); + } + + // Async shutdown with 35s drain window for the worker child. The worker + // has its own SIGTERM handler (minions/worker.ts:79-85) that drains + // in-flight jobs for up to 30s before exit. We give it 35s here to + // account for signal-delivery latency, then SIGKILL as a last resort. + // + // No `process.on('exit')` handler — its callback runs synchronously and + // cannot await the worker's drain. + const shutdown = async (sig: string) => { + if (stopping) return; + stopping = true; + console.log(`Autopilot stopping (${sig}).`); + if (workerProc) { + try { workerProc.kill('SIGTERM'); } catch { /* already dead */ } + await Promise.race([ + new Promise(r => workerProc!.once('exit', () => r())), + new Promise(r => setTimeout(() => r(), 35_000)), + ]); + if (workerProc && !workerProc.killed) { + try { workerProc.kill('SIGKILL'); } catch { /* already dead */ } + } + } + try { unlinkSync(lockPath); } catch { /* already gone */ } + process.exit(0); + }; + process.on('SIGTERM', () => { void shutdown('SIGTERM'); }); + process.on('SIGINT', () => { void shutdown('SIGINT'); }); let consecutiveErrors = 0; @@ -92,6 +187,10 @@ export async function runAutopilot(engine: BrainEngine, args: string[]) { const cycleStart = Date.now(); let cycleOk = true; + // Refresh the lock mtime so another cron-fired autopilot doesn't + // declare the instance stale after 10 minutes (Codex C). + try { utimesSync(lockPath, new Date(), new Date()); } catch { /* best-effort */ } + // DB health check (reconnect if needed) try { await engine.getConfig('version'); @@ -102,28 +201,57 @@ export async function runAutopilot(engine: BrainEngine, args: string[]) { } catch (e) { logError('reconnect', e); } } - // 1. Sync - try { - const { performSync } = await import('./sync.ts'); - const result = await performSync(engine, { repoPath, noEmbed: true }); - if (result.status === 'synced') { - console.log(`[sync] +${result.added} ~${result.modified} -${result.deleted}`); - } - } catch (e) { logError('sync', e); cycleOk = false; } + if (useMinionsDispatch) { + // Submit ONE autopilot-cycle job per cycle slot. The idempotency key + // dedupes overrun submissions — if a cycle's job runs longer than + // the interval, the next submission is a no-op at the DB layer + // (ON CONFLICT DO NOTHING on the unique partial index). + try { + const { MinionQueue } = await import('../core/minions/queue.ts'); + const queue = new MinionQueue(engine); + const slotMs = Math.floor(Date.now() / (baseInterval * 1000)) * baseInterval * 1000; + const slot = new Date(slotMs).toISOString(); + const timeoutMs = Math.max(baseInterval * 2 * 1000, 300_000); + const job = await queue.add('autopilot-cycle', + { repoPath }, + { + queue: 'default', + idempotency_key: `autopilot-cycle:${slot}`, + max_attempts: 2, + timeout_ms: timeoutMs, + }, + ); + if (jsonMode) { + process.stderr.write(JSON.stringify({ event: 'dispatched', job_id: job.id, slot }) + '\n'); + } else { + console.log(`[dispatch] job #${job.id} autopilot-cycle slot=${slot}`); + } + } catch (e) { logError('dispatch', e); cycleOk = false; } + } else { + // Inline fallback — same as pre-v0.11.1 behavior. + // 1. Sync + try { + const { performSync } = await import('./sync.ts'); + const result = await performSync(engine, { repoPath, noEmbed: true }); + if (result.status === 'synced') { + console.log(`[sync] +${result.added} ~${result.modified} -${result.deleted}`); + } + } catch (e) { logError('sync', e); cycleOk = false; } - // 2. Extract (full brain, incremental dedup handles repeats) - try { - const { runExtract } = await import('./extract.ts'); - await runExtract(engine, ['all', '--dir', repoPath]); - } catch (e) { logError('extract', e); cycleOk = false; } + // 2. Extract (full brain, incremental dedup handles repeats) + try { + const { runExtractCore } = await import('./extract.ts'); + await runExtractCore(engine, { mode: 'all', dir: repoPath }); + } catch (e) { logError('extract', e); cycleOk = false; } - // 3. Embed stale - try { - const { runEmbed } = await import('./embed.ts'); - await runEmbed(engine, ['--stale']); - } catch (e) { logError('embed', e); cycleOk = false; } + // 3. Embed stale + try { + const { runEmbedCore } = await import('./embed.ts'); + await runEmbedCore(engine, { stale: true }); + } catch (e) { logError('embed', e); cycleOk = false; } + } - // 4. Health check + adaptive interval + // 4. Health check + adaptive interval (same for both paths) let interval = baseInterval; try { const health = await engine.getHealth(); @@ -147,7 +275,8 @@ export async function runAutopilot(engine: BrainEngine, args: string[]) { consecutiveErrors++; if (consecutiveErrors >= 5) { console.error('5 consecutive cycle failures. Stopping autopilot.'); - process.exit(1); + void shutdown('cycle-failure-cap'); + break; } } diff --git a/test/autopilot-resolve-cli.test.ts b/test/autopilot-resolve-cli.test.ts new file mode 100644 index 000000000..175725496 --- /dev/null +++ b/test/autopilot-resolve-cli.test.ts @@ -0,0 +1,53 @@ +/** + * Tests for resolveGbrainCliPath() — picks the right executable to supervise + * as the Minions worker child. Codex caught that the earlier plan's use of + * process.execPath is wrong on source installs (points at the Bun runtime, + * not `gbrain`). + */ + +import { describe, test, expect } from 'bun:test'; +import { resolveGbrainCliPath } from '../src/commands/autopilot.ts'; + +describe('resolveGbrainCliPath', () => { + test('returns a non-empty string', () => { + // Whatever the test environment is (bun run ...), the resolver should + // find *something* — either argv[1] (cli.ts entry), execPath (compiled + // binary), or `which gbrain`. If none of those work, it throws; in + // test, argv[1] is the test runner path which usually ends in .ts, so + // the first branch or the `which` fallback catches it. + let path: string; + try { + path = resolveGbrainCliPath(); + } catch (e) { + // If we throw, that means neither argv[1] nor execPath nor $PATH has + // gbrain — on a machine without gbrain installed, this is expected. + expect((e as Error).message).toContain('resolve'); + return; + } + expect(typeof path).toBe('string'); + expect(path.length).toBeGreaterThan(0); + }); + + test('accepts /gbrain suffix (compiled binary)', () => { + // Simulate compiled-binary detection by setting argv[1] to /usr/local/bin/gbrain + const orig = process.argv[1]; + process.argv[1] = '/usr/local/bin/gbrain'; + try { + const path = resolveGbrainCliPath(); + expect(path).toBe('/usr/local/bin/gbrain'); + } finally { + process.argv[1] = orig; + } + }); + + test('accepts /cli.ts suffix (source install)', () => { + const orig = process.argv[1]; + process.argv[1] = '/some/path/src/cli.ts'; + try { + const path = resolveGbrainCliPath(); + expect(path).toBe('/some/path/src/cli.ts'); + } finally { + process.argv[1] = orig; + } + }); +});