diff --git a/src/commands/jobs.ts b/src/commands/jobs.ts index 599a934ff..4f7c6a240 100644 --- a/src/commands/jobs.ts +++ b/src/commands/jobs.ts @@ -1885,6 +1885,7 @@ export async function registerBuiltinHandlers( signal: job.signal, deadlineAtMs: job.deadlineAtMs, // #2781: phases budget sub-work from remaining time phases, + forceGlobalOrphans: true, yieldBetweenPhases: async () => { await new Promise((r) => setImmediate(r)); }, }); diff --git a/src/core/cycle.ts b/src/core/cycle.ts index a6a15fbc1..b43d3ad24 100644 --- a/src/core/cycle.ts +++ b/src/core/cycle.ts @@ -195,8 +195,10 @@ export const ALL_PHASES: CyclePhase[] = [ * - `source`: safe to parallelize per source. Sync reads/writes the * one source's rows; extract walks changed slugs. * - `global`: must serialize across the brain. Embed walks all stale - * chunks; orphans/purge sweep brain-wide; grade_takes + calibration - * aggregate across sources; resolve_symbol_edges walks every chunk. + * chunks; purge sweeps brain-wide; orphans can report a single + * resolved source but still belongs in the serialized global lane; + * grade_takes + calibration aggregate across sources; + * resolve_symbol_edges walks every chunk. * - `mixed`: per-phase decomposition needed before parallelizing. * Synthesize reads the brain-global transcripts dir but writes to * per-source slugs (via subagent allowlist). Patterns reads @@ -453,6 +455,12 @@ export interface CycleOpts { * loop bug (codex finding #3). */ synthBypassDreamGuard?: boolean; + /** + * Force the orphans phase to scan brain-wide even when `brainDir` resolves to + * a source. Used by autopilot global maintenance, whose phase set is + * intentionally brain-wide. + */ + forceGlobalOrphans?: boolean; /** * AbortSignal from the Minions worker (v0.22.1, #403). When aborted * (timeout, cancel, lock-loss), runCycle bails between phases and @@ -469,12 +477,14 @@ export interface CycleOpts { * + every existing caller). * * **Note for follow-up waves:** this only scopes the LOCK. Several - * cycle phases (`embed`, `orphans`, `purge`, `resolve_symbol_edges`, - * `grade_takes`, `calibration_profile`) still operate brain-wide - * regardless of sourceId — see the `PHASE_SCOPE` taxonomy. Per-source - * cycle locks let two cycles RUN, but the global-scoped phases - * inside each will still touch the same rows. Genuine per-source - * fan-out requires the deferred TODOs in the plan. + * cycle phases (`embed`, `purge`, `resolve_symbol_edges`, `grade_takes`, + * `calibration_profile`) still operate brain-wide regardless of sourceId + * — see the `PHASE_SCOPE` taxonomy. `orphans` uses the resolved source + * for its candidate set when one exists, but it remains in the serialized + * global lane for autopilot scheduling. Per-source cycle locks let two + * cycles RUN, but the global-scoped phases inside each will still touch + * the same rows. Genuine per-source fan-out requires the deferred TODOs + * in the plan. * * Validated via `assertValidSourceId` in `cycleLockIdFor` (defense-in-depth). */ @@ -1410,10 +1420,10 @@ async function runPhasePurge(engine: BrainEngine, dryRun: boolean): Promise { +async function runPhaseOrphans(engine: BrainEngine, sourceId?: string): Promise { try { const { findOrphans } = await import('../commands/orphans.ts'); - const result = await findOrphans(engine); + const result = await findOrphans(engine, sourceId !== undefined ? { sourceId } : {}); const count = result.total_orphans; // Orphans are a code-smell signal, not a fatal condition. The // original `count > 20` cutoff was tuned for small dev brains; on @@ -1432,8 +1442,10 @@ async function runPhaseOrphans(engine: BrainEngine): Promise { summary: `${count} orphan page(s) out of ${result.total_pages} total`, details: { total_orphans: count, + total_linkable: result.total_linkable, total_pages: result.total_pages, excluded: result.excluded, + ...(sourceId !== undefined ? { source_id: sourceId } : {}), }, }; } catch (e) { @@ -1497,6 +1509,7 @@ export async function runCycle( const cycleSourceId: string | undefined = engine ? (opts.sourceId ?? (await resolveSourceForDir(engine, brainDir))) : opts.sourceId; + const orphansSourceId = opts.forceGlobalOrphans ? undefined : cycleSourceId; const progress = createProgress(cliOptsToProgressOptions(getCliOptions())); @@ -2268,7 +2281,7 @@ export async function runCycle( }); } else { progress.start('cycle.orphans'); - const { result, duration_ms } = await timePhase(() => runPhaseOrphans(engine)); + const { result, duration_ms } = await timePhase(() => runPhaseOrphans(engine, orphansSourceId)); result.duration_ms = duration_ms; phaseResults.push(result); progress.finish(); diff --git a/test/autopilot-global-maintenance.test.ts b/test/autopilot-global-maintenance.test.ts index 0ec76f290..c5c500d88 100644 --- a/test/autopilot-global-maintenance.test.ts +++ b/test/autopilot-global-maintenance.test.ts @@ -11,6 +11,9 @@ */ import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test'; +import { mkdtempSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; import { PGLiteEngine } from '../src/core/pglite-engine.ts'; import { resetPgliteState } from './helpers/reset-pglite.ts'; import { registerBuiltinHandlers } from '../src/commands/jobs.ts'; @@ -130,13 +133,23 @@ describe('autopilot-global-maintenance handler stamps last_global_at (PGLite)', 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 repoPath = mkdtempSync(join(tmpdir(), 'gbrain-global-maintenance-')); + await engine.executeRaw( + `INSERT INTO sources (id, name, local_path) VALUES ($1, $2, $3)`, + ['repo-a', 'repo-a', repoPath], + ); const handlers = await captureHandlers(); const handler = handlers.get('autopilot-global-maintenance'); expect(handler).toBeTruthy(); - const result = await handler!({ data: { phases: ['orphans', 'embed'] }, signal: undefined }); + const result = await handler!({ + data: { phases: ['orphans', 'embed'], repoPath }, + 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); + const orphans = result.report.phases.find((p: any) => p.phase === 'orphans'); + expect(orphans).toBeTruthy(); + expect(orphans.details.source_id).toBeUndefined(); 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); diff --git a/test/core/cycle.serial.test.ts b/test/core/cycle.serial.test.ts index 247124d58..e4f6ff3dc 100644 --- a/test/core/cycle.serial.test.ts +++ b/test/core/cycle.serial.test.ts @@ -21,6 +21,7 @@ let syncCalls: Array<{ dryRun: boolean | undefined; noPull: boolean | undefined; let extractCalls: Array<{ mode: string; dir: string; slugs: string[] | undefined }> = []; let embedCalls: Array<{ stale: boolean | undefined; dryRun: boolean | undefined }> = []; let orphansCalls: number = 0; +let orphansOpts: Array<{ sourceId?: string } | undefined> = []; // Mock lint mock.module('../../src/commands/lint.ts', () => ({ @@ -98,8 +99,9 @@ mock.module('../../src/commands/embed.ts', () => ({ // Mock orphans mock.module('../../src/commands/orphans.ts', () => ({ - findOrphans: async () => { + findOrphans: async (_engine: any, opts?: { sourceId?: string }) => { orphansCalls++; + orphansOpts.push(opts); return { orphans: [], total_orphans: 1, @@ -148,6 +150,7 @@ beforeEach(() => { extractCalls = []; embedCalls = []; orphansCalls = 0; + orphansOpts = []; }); // ─── dryRun propagation (regression guards) ──────────────────────── @@ -215,6 +218,11 @@ describe('runCycle — phase selection', () => { expect(orphansCalls).toBe(1); expect(syncCalls.length).toBe(0); }); + + test('--phase orphans preserves explicit source scope', async () => { + await runCycle(sharedEngine, { brainDir: '/tmp/brain', phases: ['orphans'], sourceId: 'source-a' }); + expect(orphansOpts.at(-1)).toEqual({ sourceId: 'source-a' }); + }); }); // ─── Lock-skip for non-DB-write phase selections ────────────────── @@ -499,6 +507,28 @@ describe('runCycle — sourceId resolution (regression #475)', () => { expect(syncCalls.at(-1)?.sourceId).toBe('default'); }); + test('seeded sources row → orphans phase receives matching sourceId', async () => { + await (sharedEngine as any).db.query( + `INSERT INTO sources (id, name, local_path) VALUES ($1, $2, $3)`, + ['alpha', 'alpha', '/tmp/brain-2349-alpha'], + ); + await runCycle(sharedEngine, { brainDir: '/tmp/brain-2349-alpha', phases: ['orphans'] }); + expect(orphansOpts.at(-1)).toEqual({ sourceId: 'alpha' }); + }); + + test('forceGlobalOrphans keeps orphans brain-wide even when brainDir maps to a source', async () => { + await (sharedEngine as any).db.query( + `INSERT INTO sources (id, name, local_path) VALUES ($1, $2, $3)`, + ['global-source', 'global-source', '/tmp/brain-2349-global'], + ); + await runCycle(sharedEngine, { + brainDir: '/tmp/brain-2349-global', + phases: ['embed', 'orphans', 'purge'], + forceGlobalOrphans: true, + }); + expect(orphansOpts.at(-1)).toEqual({}); + }); + test('no matching sources row → performSync receives sourceId=undefined', async () => { await runCycle(sharedEngine, { brainDir: '/tmp/brain-475-b' }); expect(syncCalls.at(-1)?.sourceId).toBeUndefined();