feat(autopilot): split the cycle — per-source phases + one global-maintenance job (#2194 #2227)

N per-source cycles each ran the brain-wide global phases (embed-all/orphans/
purge/…) concurrently, thrashing the same rows and taking the worker 4→10GB in
<60s → RSS-kill → orphaned stalls. Split them: per-source jobs now run only
source-scoped (+ mixed) phases and stamp last_source_cycle_at; a new
autopilot-global-maintenance job runs the global phases ONCE per window
(idempotency_key + maxWaiting:1 = structural single-flight) and stamps
autopilot.last_global_at. This is the codex-endorsed design that replaced the
rejected skip-and-stamp-fresh approach (codex #1/#2): no freshness poisoning, no
starvation — global work always runs as its own job, never marked done when it
wasn't. PHASE_SCOPE is now a runtime partition (GLOBAL ∪ NON_GLOBAL == ALL).
last_full_cycle_at still written for doctor/legacy (no longer a global gate).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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' };
+}

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();
           }

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<void>((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.

src/core/cycle.ts

@@ -242,6 +242,25 @@ export const PHASE_SCOPE: Record<CyclePhase, PhaseScope> = {
   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)}`);
     }
   }
 

src/core/minions/handler-timeouts.ts

@@ -34,6 +34,9 @@ export const HANDLER_DEFAULT_TIMEOUT_MS: Readonly<Record<string, number>> = {
   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,
 };
 
 /**

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<string, unknown>) => {
+        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<string, (job: any) => Promise<any>>();
+    const fakeWorker = { register(name: string, fn: (job: any) => Promise<any>) { 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);
+  });
+});