admin/gbrain
1
0
Fork 0
mirror of https://github.com/garrytan/gbrain.git synced 2026-07-29 19:01:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(autopilot): Minions dispatch + worker spawn supervisor + async shutdown

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
Garry Tan
2026-04-18 13:35:51 +08:00
co-authored by Claude Opus 4.7
parent 53d63b6380
commit ec0556a66e
2 changed files with 212 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/commands/autopilot.ts
+159 -30
View File
@@ -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 <path>] [--interval N] [--json]
* gbrain autopilot [--repo <path>] [--interval N] [--json] [--inline]
* gbrain autopilot --install [--repo <path>]
* 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 <path>] [--interval N] [--json]\n gbrain autopilot --install [--repo <path>]\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<void>(r => workerProc!.once('exit', () => r())),
new Promise<void>(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;
}
}
test/autopilot-resolve-cli.test.ts
+53
View File
@@ -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;
}
});
});