diff --git a/src/commands/autopilot-fanout.ts b/src/commands/autopilot-fanout.ts index 5b186238c..9c558ad57 100644 --- a/src/commands/autopilot-fanout.ts +++ b/src/commands/autopilot-fanout.ts @@ -32,6 +32,7 @@ import type { BrainEngine, SourceRow } from '../core/engine.ts'; import type { MinionQueue } from '../core/minions/queue.ts'; +import { NON_GLOBAL_PHASES, GLOBAL_PHASES, LAST_GLOBAL_AT_KEY } from '../core/cycle.ts'; const FULL_CYCLE_FLOOR_MIN = 60; @@ -436,6 +437,11 @@ export async function dispatchPerSource( repoPath: opts.repoPath, source_id: src.id, pull: !!remoteUrl, + // #2194 fix #3 (cycle split): per-source cycles run ONLY source-scoped + // (+ mixed) phases. The brain-wide global phases (embed, orphans, + // purge, …) run once in autopilot-global-maintenance, not N times + // concurrently here — the fix for the 4→10GB RSS blowout. + phases: NON_GLOBAL_PHASES, }, { queue: 'default', @@ -504,3 +510,62 @@ export async function dispatchPerSource( legacy_fallback: false, }; } + +const GLOBAL_FLOOR_MIN = 60; + +/** Is the brain-wide maintenance overdue? Null/unparseable → overdue. */ +export function isGlobalMaintenanceStale(lastGlobalAtIso: string | null, now = Date.now(), floorMin = GLOBAL_FLOOR_MIN): boolean { + if (!lastGlobalAtIso) return true; + const d = new Date(lastGlobalAtIso); + if (!Number.isFinite(d.getTime())) return true; + return (now - d.getTime()) / 60_000 >= floorMin; +} + +/** + * #2194 fix #3 / #2227 bug #3 — dispatch the single brain-wide maintenance job + * that runs the `global` cycle phases (embed, orphans, purge, …) ONCE per + * window, instead of N per-source cycles each running them concurrently (the + * RSS blowout). Single-flight is structural: one `idempotency_key` + + * `maxWaiting:1`, so a slow run never stacks. Gated on `autopilot.last_global_at` + * (stamped by the handler on success). Postgres-only fan-out concern; on PGLite + * the file lock already serializes, but the job is still correct there. + */ +export async function dispatchGlobalMaintenance( + engine: BrainEngine, + queue: MinionQueue, + opts: { repoPath: string; slot: string; timeoutMs: number; jsonMode: boolean; emit?: (l: string) => void; log?: (l: string) => void }, +): Promise<{ dispatched: boolean; reason: 'stale' | 'fresh' }> { + const emit = opts.emit ?? ((line) => process.stderr.write(line + '\n')); + const log = opts.log ?? ((line) => console.log(line)); + + let floorMin = GLOBAL_FLOOR_MIN; + const floorCfg = await engine.getConfig('autopilot.global_floor_min'); + if (floorCfg) { + const n = parseInt(floorCfg, 10); + if (Number.isFinite(n) && n >= 1) floorMin = n; + } + const lastGlobalAt = await engine.getConfig(LAST_GLOBAL_AT_KEY); + if (!isGlobalMaintenanceStale(lastGlobalAt, Date.now(), floorMin)) { + return { dispatched: false, reason: 'fresh' }; + } + + const job = await queue.add( + 'autopilot-global-maintenance', + { repoPath: opts.repoPath, phases: GLOBAL_PHASES }, + { + queue: 'default', + // Structural single-flight: one global job per slot; maxWaiting:1 coalesces + // any surplus so a slow brain-wide pass never stacks duplicates. + idempotency_key: `autopilot-global:${opts.slot}`, + max_attempts: 2, + timeout_ms: opts.timeoutMs, + maxWaiting: 1, + }, + ); + if (opts.jsonMode) { + emit(JSON.stringify({ event: 'dispatched', job_id: job.id, mode: 'global_maintenance', slot: opts.slot })); + } else { + log(`[dispatch] job #${job.id} autopilot-global-maintenance (brain-wide phases)`); + } + return { dispatched: true, reason: 'stale' }; +} diff --git a/src/commands/autopilot.ts b/src/commands/autopilot.ts index bbfc3eb79..927b571e5 100644 --- a/src/commands/autopilot.ts +++ b/src/commands/autopilot.ts @@ -871,7 +871,7 @@ export async function runAutopilot(engine: BrainEngine, args: string[]) { // codex P1-3). Fresh-install brains with no sources rows fall // back to the legacy single autopilot-cycle so existing // behavior is preserved. - const { dispatchPerSource, resolveEffectiveFanoutMax } = await import('./autopilot-fanout.ts'); + const { dispatchPerSource, dispatchGlobalMaintenance, resolveEffectiveFanoutMax } = await import('./autopilot-fanout.ts'); // #2194 fix #1: clamp fan-out to the worker's effective concurrency // (reserve ≥1 slot), gated on a LIVE supervisor so a stale audit row // can't shrink throughput (codex #9/D5). autopilot-cycle jobs run on @@ -884,6 +884,18 @@ export async function runAutopilot(engine: BrainEngine, args: string[]) { fanoutMax, jsonMode, }); + // #2194 fix #3 / #2227 bug #3: dispatch the single brain-wide + // maintenance job (embed/orphans/purge/…) once per window — the per- + // source cycles above no longer run global phases, so this is where + // the brain-wide work happens (single-flight, no RSS blowout). Only on + // the per-source path (legacy single-source still runs everything). + if (!result.legacy_fallback) { + try { + await dispatchGlobalMaintenance(engine, queue, { repoPath, slot, timeoutMs, jsonMode }); + } catch (e) { + if (jsonMode) process.stderr.write(JSON.stringify({ event: 'global_maintenance_dispatch_failed', error: e instanceof Error ? e.message : String(e) }) + '\n'); + } + } if (result.dispatched.length > 0 || result.legacy_fallback) { lastFullCycleAt = Date.now(); } diff --git a/src/commands/jobs.ts b/src/commands/jobs.ts index 14270cbd1..ac5de27ca 100644 --- a/src/commands/jobs.ts +++ b/src/commands/jobs.ts @@ -1739,6 +1739,49 @@ export async function registerBuiltinHandlers( }; }); + // #2194 fix #3 / #2227 bug #3 — brain-wide maintenance. Runs the `global` + // cycle phases (embed, orphans, purge, resolve_symbol_edges, grade_takes, + // calibration_profile, synthesize_concepts, skillopt) ONCE per window instead + // of N times concurrently across per-source cycles (the 4→10GB RSS blowout). + // No source_id → uses the legacy global cycle lock; stamps autopilot.last_global_at + // on success so the dispatch gate backs off. + worker.register('autopilot-global-maintenance', async (job) => { + const { runCycle, GLOBAL_PHASES, LAST_GLOBAL_AT_KEY, ALL_PHASES } = await import('../core/cycle.ts'); + const repoPath: string | null = typeof job.data.repoPath === 'string' + ? job.data.repoPath + : (await engine.getConfig('sync.repo_path')) ?? null; + + const validPhases = new Set(ALL_PHASES); + const requested = Array.isArray(job.data.phases) + ? (job.data.phases as string[]).filter((p) => validPhases.has(p as never)) + : GLOBAL_PHASES; + const phases = (requested.length > 0 ? requested : GLOBAL_PHASES) as typeof GLOBAL_PHASES; + + const report = await runCycle(engine, { + brainDir: repoPath, + pull: false, // brain-wide DB/maintenance work never git-pulls + signal: job.signal, + phases, + yieldBetweenPhases: async () => { await new Promise((r) => setImmediate(r)); }, + }); + + // Stamp last_global_at only on a non-failed run so a failed pass stays stale + // and re-dispatches next tick (self-healing retry). + if (report.status === 'ok' || report.status === 'clean' || report.status === 'partial') { + try { + await engine.setConfig(LAST_GLOBAL_AT_KEY, new Date().toISOString()); + } catch (e) { + console.warn(`[autopilot-global-maintenance] failed to stamp last_global_at: ${e instanceof Error ? e.message : String(e)}`); + } + } + + return { + partial: report.status === 'partial' || report.status === 'failed', + status: report.status, + report, + }; + }); + // Shell handler is always registered. Runtime env guard lives inside the // handler so claimed jobs emit a clear rejection log on workers missing // GBRAIN_ALLOW_SHELL_JOBS=1. diff --git a/src/core/cycle.ts b/src/core/cycle.ts index 17a31bccb..8219b3b77 100644 --- a/src/core/cycle.ts +++ b/src/core/cycle.ts @@ -242,6 +242,25 @@ export const PHASE_SCOPE: Record = { skillopt: 'global', }; +/** + * #2194 fix #3 / #2227 bug #3 — the cycle split. + * + * Per-source autopilot cycles run ONLY the source-scoped (and mixed) phases; + * the brain-wide `global` phases (embed, orphans, purge, resolve_symbol_edges, + * grade_takes, calibration_profile, synthesize_concepts, skillopt) run ONCE in + * a separate `autopilot-global-maintenance` job instead of N times concurrently + * across per-source cycles (the 4→10GB RSS blowout). Single-flight is + * structural: one global job, not a skip-and-pretend-fresh hack (codex #1/#2). + * + * GLOBAL_PHASES ∪ NON_GLOBAL_PHASES == ALL_PHASES, with no overlap — pinned by + * test/autopilot-global-maintenance.test.ts. + */ +export const GLOBAL_PHASES: CyclePhase[] = ALL_PHASES.filter((p) => PHASE_SCOPE[p] === 'global'); +export const NON_GLOBAL_PHASES: CyclePhase[] = ALL_PHASES.filter((p) => PHASE_SCOPE[p] !== 'global'); + +/** Config key holding the ISO timestamp of the last successful global-maintenance run. */ +export const LAST_GLOBAL_AT_KEY = 'autopilot.last_global_at'; + /** * Phases that mutate state (filesystem or DB) and therefore should * coordinate via the cycle lock. Only orphans is truly read-only @@ -2305,12 +2324,21 @@ export async function runCycle( // the cost of missing a successful write (next cycle will redo work). if (opts.sourceId && engine && !dryRun && !aborted && (status === 'ok' || status === 'clean' || status === 'partial')) { try { + const nowIso = new Date().toISOString(); + // #2194 fix #3 (the cycle split): `last_source_cycle_at` is the NEW gate + // for per-source dispatch (source-scoped phases done). We ALSO keep + // `last_full_cycle_at` current so doctor's cycle-freshness check and any + // legacy reader stay valid — it's no longer a *gate* for the brain-wide + // phases (those gate on autopilot.last_global_at), so writing it on a + // source-only cycle does not re-introduce the freshness poisoning codex + // flagged in the rejected skip-based design. await engine.updateSourceConfig(opts.sourceId, { - last_full_cycle_at: new Date().toISOString(), + last_source_cycle_at: nowIso, + last_full_cycle_at: nowIso, }); } catch (e) { // Best-effort; cycle already succeeded by the time we get here. - console.warn(`[cycle] failed to write last_full_cycle_at for source ${opts.sourceId}: ${e instanceof Error ? e.message : String(e)}`); + console.warn(`[cycle] failed to write last_source_cycle_at for source ${opts.sourceId}: ${e instanceof Error ? e.message : String(e)}`); } } diff --git a/src/core/minions/handler-timeouts.ts b/src/core/minions/handler-timeouts.ts index 9cfa6919b..365e51160 100644 --- a/src/core/minions/handler-timeouts.ts +++ b/src/core/minions/handler-timeouts.ts @@ -34,6 +34,9 @@ export const HANDLER_DEFAULT_TIMEOUT_MS: Readonly> = { subagent_aggregator: THIRTY_MIN_MS, 'embed-backfill': THIRTY_MIN_MS, 'autopilot-cycle': THIRTY_MIN_MS, + // #2194 fix #3: brain-wide maintenance (embed-all/orphans/purge/…) can run + // longer than a single source cycle; give it the same 30-min budget. + 'autopilot-global-maintenance': THIRTY_MIN_MS, }; /** diff --git a/test/autopilot-global-maintenance.test.ts b/test/autopilot-global-maintenance.test.ts new file mode 100644 index 000000000..0ec76f290 --- /dev/null +++ b/test/autopilot-global-maintenance.test.ts @@ -0,0 +1,146 @@ +/** + * #2194 fix #3 / #2227 bug #3 — the cycle split. + * + * Per-source autopilot cycles run ONLY source-scoped (+ mixed) phases; the + * brain-wide `global` phases run ONCE in a separate autopilot-global-maintenance + * job. This replaces the rejected skip-and-stamp-fresh design (codex #1/#2): the + * split makes single-flight structural (one global job, not N concurrent embeds) + * and never marks a source "fresh" for global work it didn't do. These tests pin + * the phase partition, the dispatch gate, the per-source phase set, and the + * global handler stamping autopilot.last_global_at. + */ + +import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test'; +import { PGLiteEngine } from '../src/core/pglite-engine.ts'; +import { resetPgliteState } from './helpers/reset-pglite.ts'; +import { registerBuiltinHandlers } from '../src/commands/jobs.ts'; +import { + ALL_PHASES, + GLOBAL_PHASES, + NON_GLOBAL_PHASES, + PHASE_SCOPE, + LAST_GLOBAL_AT_KEY, +} from '../src/core/cycle.ts'; +import { + dispatchGlobalMaintenance, + isGlobalMaintenanceStale, + dispatchPerSource, +} from '../src/commands/autopilot-fanout.ts'; +import type { BrainEngine } from '../src/core/engine.ts'; + +describe('cycle phase partition (#2194 fix #3)', () => { + test('GLOBAL ∪ NON_GLOBAL == ALL_PHASES, no overlap', () => { + const union = new Set([...GLOBAL_PHASES, ...NON_GLOBAL_PHASES]); + expect(union.size).toBe(ALL_PHASES.length); + for (const p of ALL_PHASES) expect(union.has(p)).toBe(true); + // No phase in both. + const overlap = GLOBAL_PHASES.filter((p) => NON_GLOBAL_PHASES.includes(p)); + expect(overlap).toEqual([]); + }); + + test('every GLOBAL phase is PHASE_SCOPE==="global"; embed is global, lint is not', () => { + for (const p of GLOBAL_PHASES) expect(PHASE_SCOPE[p]).toBe('global'); + expect(GLOBAL_PHASES).toContain('embed'); + expect(GLOBAL_PHASES).toContain('orphans'); + expect(GLOBAL_PHASES).toContain('purge'); + expect(NON_GLOBAL_PHASES).toContain('lint'); + expect(NON_GLOBAL_PHASES).toContain('sync'); + expect(NON_GLOBAL_PHASES).not.toContain('embed'); + }); +}); + +describe('isGlobalMaintenanceStale', () => { + const now = Date.UTC(2026, 5, 16, 12, 0, 0); + test('null/unparseable → stale (must run)', () => { + expect(isGlobalMaintenanceStale(null, now)).toBe(true); + expect(isGlobalMaintenanceStale('not-a-date', now)).toBe(true); + }); + test('older than floor → stale; within floor → fresh', () => { + expect(isGlobalMaintenanceStale(new Date(now - 61 * 60_000).toISOString(), now, 60)).toBe(true); + expect(isGlobalMaintenanceStale(new Date(now - 10 * 60_000).toISOString(), now, 60)).toBe(false); + }); +}); + +describe('dispatchGlobalMaintenance — single-flight gate', () => { + function stubs(lastGlobalAt: string | null) { + const added: Array<{ name: string; data: any; opts: any }> = []; + const engine = { + kind: 'postgres' as const, + getConfig: async (k: string) => (k === LAST_GLOBAL_AT_KEY ? lastGlobalAt : null), + } as unknown as BrainEngine; + const queue = { + add: async (name: string, data: unknown, opts: Record) => { + added.push({ name, data, opts }); return { id: 1 }; + }, + } as any; + return { engine, queue, added }; + } + + test('stale (never run) → dispatches one global job with single-flight opts', async () => { + const { engine, queue, added } = stubs(null); + const r = await dispatchGlobalMaintenance(engine, queue, { repoPath: '/tmp', slot: 's1', timeoutMs: 1, jsonMode: true, emit: () => {} }); + expect(r.dispatched).toBe(true); + expect(added.length).toBe(1); + expect(added[0].name).toBe('autopilot-global-maintenance'); + expect(added[0].opts.idempotency_key).toBe('autopilot-global:s1'); + expect(added[0].opts.maxWaiting).toBe(1); // structural single-flight + expect(added[0].data.phases).toEqual(GLOBAL_PHASES); + }); + + test('fresh → does NOT dispatch', async () => { + const { engine, queue, added } = stubs(new Date().toISOString()); + const r = await dispatchGlobalMaintenance(engine, queue, { repoPath: '/tmp', slot: 's1', timeoutMs: 1, jsonMode: true, emit: () => {} }); + expect(r.dispatched).toBe(false); + expect(added.length).toBe(0); + }); +}); + +describe('dispatchPerSource — per-source jobs carry NON_GLOBAL phases (no embed)', () => { + test('each per-source job sets phases = NON_GLOBAL_PHASES', async () => { + const sources = [{ id: 'repo-a', name: 'a', config: {} }, { id: 'repo-b', name: 'b', config: {} }]; + const added: any[] = []; + const engine = { + kind: 'postgres' as const, + listAllSources: async () => sources, + getConfig: async () => null, + executeRaw: async () => [], + } as unknown as BrainEngine; + const queue = { add: async (name: string, data: unknown, opts: unknown) => { added.push({ name, data, opts }); return { id: added.length }; } } as any; + await dispatchPerSource(engine, queue, { repoPath: '/tmp', slot: 's', timeoutMs: 1, fanoutMax: 4, jsonMode: true, emit: () => {}, log: () => {} }); + expect(added.length).toBe(2); + for (const j of added) { + expect(j.data.phases).toEqual(NON_GLOBAL_PHASES); + expect(j.data.phases).not.toContain('embed'); + } + }); +}); + +describe('autopilot-global-maintenance handler stamps last_global_at (PGLite)', () => { + let engine: PGLiteEngine; + beforeAll(async () => { engine = new PGLiteEngine(); await engine.connect({}); await engine.initSchema(); }, 30000); + afterAll(async () => { await engine.disconnect(); }); + beforeEach(async () => { await resetPgliteState(engine); }); + + async function captureHandlers() { + const handlers = new Map Promise>(); + const fakeWorker = { register(name: string, fn: (job: any) => Promise) { handlers.set(name, fn); } }; + await registerBuiltinHandlers(fakeWorker as never, engine); + return handlers; + } + + test('runs global phases (no source_id) and stamps autopilot.last_global_at on success', async () => { + expect(await engine.getConfig(LAST_GLOBAL_AT_KEY)).toBeNull(); + const handlers = await captureHandlers(); + const handler = handlers.get('autopilot-global-maintenance'); + expect(handler).toBeTruthy(); + + const result = await handler!({ data: { phases: ['orphans', 'embed'] }, signal: undefined }); + // The cycle ran the requested global phases (DB-only on an empty brain). + expect(result.report.phases.some((p: any) => p.phase === 'orphans')).toBe(true); + expect(['ok', 'clean', 'partial']).toContain(result.report.status); + // Freshness stamped so the dispatch gate backs off. + const stamped = await engine.getConfig(LAST_GLOBAL_AT_KEY); + expect(stamped).not.toBeNull(); + expect(Number.isFinite(new Date(stamped!).getTime())).toBe(true); + }); +});