diff --git a/src/commands/agent.ts b/src/commands/agent.ts index deebbfc32..ad104ca60 100644 --- a/src/commands/agent.ts +++ b/src/commands/agent.ts @@ -18,6 +18,8 @@ import type { BrainEngine } from '../core/engine.ts'; import { MinionQueue } from '../core/minions/queue.ts'; import { waitForCompletion, TimeoutError } from '../core/minions/wait-for-completion.ts'; import type { MinionJobInput, SubagentHandlerData, AggregatorHandlerData } from '../core/minions/types.ts'; +import { resolveSourceId, ALL_SOURCES } from '../core/source-resolver.ts'; +import { fetchSource } from '../core/sources-load.ts'; import { runAgentLogs } from './agent-logs.ts'; // ── arg parsing helpers ──────────────────────────────────── @@ -72,6 +74,10 @@ SUBMITTING --max-turns Max assistant turns (default 20) --tools a,b,c Subset of registered tool names (comma list) --timeout-ms Per-job wall-clock timeout + --source Brain source the subagent's writes are scoped to. + Default: the standard resolution chain (GBRAIN_SOURCE, + .gbrain-source, sources.default, ...) — see + \`gbrain sources current\` --fanout-manifest JSON array of {prompt, input_vars?} — one child each --follow Tail status until terminal (default on TTY) --detach Submit + print job id, exit immediately @@ -116,6 +122,7 @@ interface RunFlags { maxTurns?: number; tools?: string[]; timeoutMs?: number; + source?: string; fanoutManifest?: string; follow: boolean; detach: boolean; @@ -181,6 +188,7 @@ function parseRunFlags(args: string[]): { flags: RunFlags; rest: string[] } { case '--max-turns': flags.maxTurns = parseIntFlagValue(requireFlagValue(args, ++i, a), a); break; case '--tools': flags.tools = requireFlagValue(args, ++i, a).split(',').map(s => s.trim()).filter(Boolean); break; case '--timeout-ms': flags.timeoutMs = parseIntFlagValue(requireFlagValue(args, ++i, a), a); break; + case '--source': flags.source = requireFlagValue(args, ++i, a); break; case '--fanout-manifest': flags.fanoutManifest = requireFlagValue(args, ++i, a); break; case '--follow': flags.follow = true; break; case '--no-follow': flags.follow = false; break; @@ -203,17 +211,86 @@ function parseRunFlags(args: string[]): { flags: RunFlags; rest: string[] } { return { flags, rest }; } +/** + * Predicate: is this error one of the source resolver's user-facing throws + * we want to surface as a clean stderr line + exit 1? Mirrors + * dream.ts:isResolverUserError — anything else (connection failures, + * genuine bugs) propagates with a stack trace. + */ +function isResolverUserError(e: unknown): boolean { + if (!(e instanceof Error)) return false; + const m = e.message; + return (m.startsWith('Source "') && m.includes(' not found.')) + || m.startsWith('Invalid --source value') + || m.startsWith('Invalid GBRAIN_SOURCE value'); +} + +/** + * #2922: resolve the brain source for a subagent submission via the + * canonical chain (explicit --source → GBRAIN_SOURCE → .gbrain-source → + * local_path match → sources.default → sole non-default → 'default'). + * Pre-fix, `gbrain agent run` never resolved a source, so every page an + * agent job wrote landed in the seed 'default' source even on brains with + * `gbrain sources default ` configured. + * + * The `__all__` sentinel is rejected here: subagent writes must target + * exactly one source (and `validateSourceId` at tool-registry build time + * would reject it anyway — better to fail at submit than at claim). + */ +async function resolveAgentSource(engine: BrainEngine, explicit: string | undefined): Promise { + // An empty `--source ""` must fail loudly, not silently degrade to the + // env/dotfile/default tiers (resolveSourceId's `if (explicit)` treats a + // falsy value as omitted — explicit-but-empty would slip through). + if (explicit !== undefined && explicit.trim() === '') { + console.error('gbrain agent run: --source requires a non-empty value. Run `gbrain agent run --help`.'); + process.exit(2); + } + let resolved: string; + try { + resolved = await resolveSourceId(engine, explicit ?? null); + } catch (e) { + if (isResolverUserError(e)) { + console.error(`gbrain agent run: ${(e as Error).message}`); + process.exit(1); + } + throw e; + } + if (resolved === ALL_SOURCES) { + console.error( + `gbrain agent run: --source ${ALL_SOURCES} is not supported — ` + + `subagent writes must target exactly one source. Pass a concrete --source .`, + ); + process.exit(2); + } + // Archived-source guard, mirroring dream.ts: writing subagent pages into + // an archived (normally invisible) source would mask them until restore. + const src = await fetchSource(engine, resolved); + if (src?.archived === true) { + console.error( + `gbrain agent run: source ${resolved} is archived; restore with ` + + `\`gbrain sources restore ${resolved}\` before submitting agent jobs`, + ); + process.exit(1); + } + return resolved; +} + export async function runAgentRun(engine: BrainEngine, args: string[]): Promise { const { flags, rest } = parseRunFlags(args); const queue = new MinionQueue(engine); + // #2922: resolve once at submit time; both the single-job and fan-out + // paths stamp it on SubagentHandlerData.source_id so buildOpContext + // scopes every tool call to it instead of the legacy 'default'. + const sourceId = await resolveAgentSource(engine, flags.source); + // Fan-out path: --fanout-manifest supplies explicit child inputs. The // aggregator submits first (so its id is available as parent for each // child); children submit with on_child_fail='continue' so mixed // outcomes don't cascade; aggregator waits in waiting-children until // Lane 1B's terminal-set check unblocks it. if (flags.fanoutManifest) { - await runFanout(engine, queue, flags, rest.join(' ')); + await runFanout(engine, queue, flags, rest.join(' '), sourceId); return; } @@ -223,7 +300,7 @@ export async function runAgentRun(engine: BrainEngine, args: string[]): Promise< process.exit(2); } - const data: SubagentHandlerData = { prompt }; + const data: SubagentHandlerData = { prompt, source_id: sourceId }; if (flags.subagentDef) data.subagent_def = flags.subagentDef; if (flags.model) data.model = flags.model; if (flags.maxTurns) data.max_turns = flags.maxTurns; @@ -248,7 +325,7 @@ export async function runAgentRun(engine: BrainEngine, args: string[]): Promise< // ── fan-out ─────────────────────────────────────────────── -async function runFanout(engine: BrainEngine, queue: MinionQueue, flags: RunFlags, promptTemplate: string): Promise { +async function runFanout(engine: BrainEngine, queue: MinionQueue, flags: RunFlags, promptTemplate: string, sourceId: string): Promise { const manifestPath = flags.fanoutManifest!; let manifest: Array<{ prompt?: string; input_vars?: Record }>; try { @@ -272,6 +349,7 @@ async function runFanout(engine: BrainEngine, queue: MinionQueue, flags: RunFlag const entry = manifest[0]!; const data: SubagentHandlerData = { prompt: entry.prompt ?? promptTemplate, + source_id: sourceId, ...(entry.input_vars ? { input_vars: entry.input_vars } : {}), ...(flags.subagentDef ? { subagent_def: flags.subagentDef } : {}), ...(flags.model ? { model: flags.model } : {}), @@ -303,6 +381,7 @@ async function runFanout(engine: BrainEngine, queue: MinionQueue, flags: RunFlag for (const entry of manifest) { const data: SubagentHandlerData = { prompt: entry.prompt ?? promptTemplate, + source_id: sourceId, ...(entry.input_vars ? { input_vars: entry.input_vars } : {}), ...(flags.subagentDef ? { subagent_def: flags.subagentDef } : {}), ...(flags.model ? { model: flags.model } : {}), diff --git a/test/agent-cli.test.ts b/test/agent-cli.test.ts index e953b2f58..fdb71922e 100644 --- a/test/agent-cli.test.ts +++ b/test/agent-cli.test.ts @@ -8,13 +8,14 @@ * glue. */ -import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test'; +import { describe, test, expect, beforeAll, afterAll, beforeEach, spyOn } from 'bun:test'; import * as fs from 'node:fs'; import * as path from 'node:path'; import * as os from 'node:os'; import { PGLiteEngine } from '../src/core/pglite-engine.ts'; import { MinionQueue } from '../src/core/minions/queue.ts'; -import { __testing as agentTesting } from '../src/commands/agent.ts'; +import { __testing as agentTesting, runAgentRun } from '../src/commands/agent.ts'; +import { withEnv } from './helpers/with-env.ts'; import { parseSince } from '../src/commands/agent-logs.ts'; import { isProtectedJobName, PROTECTED_JOB_NAMES } from '../src/core/minions/protected-names.ts'; @@ -137,6 +138,17 @@ describe('parseRunFlags', () => { const { flags } = agentTesting.parseRunFlags(['--fanout-manifest', '/tmp/m.json']); expect(flags.fanoutManifest).toBe('/tmp/m.json'); }); + + test('#2922: --source parsed as a leading value-flag', () => { + const { flags, rest } = agentTesting.parseRunFlags(['--source', 'corporate', 'do', 'x']); + expect(flags.source).toBe('corporate'); + expect(rest).toEqual(['do', 'x']); + }); + + test('#2922: --source missing its value throws a usage error', () => { + expect(() => agentTesting.parseRunFlags(['--source'])).toThrow(/requires a value/); + expect(() => agentTesting.parseRunFlags(['--source', '--detach', 'x'])).toThrow(/requires a value/); + }); }); describe('parseSince', () => { @@ -266,6 +278,188 @@ describe('queue.add trusted-submit gate for subagent', () => { }); }); +describe('#2922: submit-time source resolution', () => { + beforeEach(async () => { + await engine.executeRaw(`DELETE FROM sources WHERE id != 'default'`); + await engine.unsetConfig('sources.default'); + await engine.executeRaw( + `INSERT INTO sources (id, name) VALUES ('corporate', 'Corporate') ON CONFLICT (id) DO NOTHING`, + ); + }); + + async function jobData(jobId: number): Promise> { + const rows = await engine.executeRaw<{ data: unknown }>( + `SELECT data FROM minion_jobs WHERE id = $1`, [jobId], + ); + return typeof rows[0]!.data === 'string' + ? JSON.parse(rows[0]!.data as string) + : rows[0]!.data as Record; + } + + async function onlyJobData(): Promise> { + const rows = await engine.executeRaw<{ id: number }>( + `SELECT id FROM minion_jobs WHERE name = 'subagent' ORDER BY id`, + ); + expect(rows.length).toBe(1); + return jobData(rows[0]!.id); + } + + test('explicit --source lands on SubagentHandlerData.source_id', async () => { + await withEnv({ GBRAIN_SOURCE: undefined }, async () => { + await runAgentRun(engine, ['--detach', '--source', 'corporate', 'write', 'a', 'page']); + const data = await onlyJobData(); + expect(data.source_id).toBe('corporate'); + expect(data.prompt).toBe('write a page'); + }); + }); + + test('no --source: sources.default (tier 5) is honored instead of the seed default', async () => { + await withEnv({ GBRAIN_SOURCE: undefined }, async () => { + await engine.setConfig('sources.default', 'corporate'); + await runAgentRun(engine, ['--detach', 'write', 'a', 'page']); + const data = await onlyJobData(); + expect(data.source_id).toBe('corporate'); + }); + }); + + test('GBRAIN_SOURCE env (tier 2) is honored', async () => { + await withEnv({ GBRAIN_SOURCE: 'corporate' }, async () => { + await runAgentRun(engine, ['--detach', 'write', 'a', 'page']); + const data = await onlyJobData(); + expect(data.source_id).toBe('corporate'); + }); + }); + + test('no signal at all: resolves to the seed default (legacy behavior preserved)', async () => { + await withEnv({ GBRAIN_SOURCE: undefined }, async () => { + await runAgentRun(engine, ['--detach', 'write', 'a', 'page']); + const data = await onlyJobData(); + expect(data.source_id).toBe('default'); + }); + }); + + test('fan-out children all carry the resolved source_id', async () => { + const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'fanout-source-')); + try { + await withEnv({ GBRAIN_SOURCE: undefined }, async () => { + const manifestPath = path.join(tmp, 'm.json'); + fs.writeFileSync(manifestPath, JSON.stringify([ + { prompt: 'chunk 1' }, { prompt: 'chunk 2' }, + ])); + await runAgentRun(engine, [ + '--source', 'corporate', '--fanout-manifest', manifestPath, '--detach', + ]); + const rows = await engine.executeRaw<{ id: number }>( + `SELECT id FROM minion_jobs WHERE name = 'subagent' ORDER BY id`, + ); + expect(rows.length).toBe(2); + for (const r of rows) { + const data = await jobData(r.id); + expect(data.source_id).toBe('corporate'); + } + }); + } finally { + fs.rmSync(tmp, { recursive: true, force: true }); + } + }); + + test('--source "" (empty explicit value) exits 2 without silently falling back', async () => { + const spy = spyOn(process, 'exit').mockImplementation(() => { throw new Error('EXIT'); }); + const errSpy = spyOn(console, 'error').mockImplementation(() => {}); + try { + await withEnv({ GBRAIN_SOURCE: undefined }, async () => { + try { + await runAgentRun(engine, ['--detach', '--source', '', 'write', 'a', 'page']); + throw new Error('expected runAgentRun to exit'); + } catch (e: any) { + expect(e.message).toBe('EXIT'); + } + }); + expect(spy).toHaveBeenCalledWith(2); + const rows = await engine.executeRaw<{ id: number }>( + `SELECT id FROM minion_jobs WHERE name = 'subagent'`, + ); + expect(rows.length).toBe(0); + } finally { + spy.mockRestore(); + errSpy.mockRestore(); + } + }); + + test('--source __all__ is rejected (subagent writes must target exactly one source)', async () => { + const spy = spyOn(process, 'exit').mockImplementation(() => { throw new Error('EXIT'); }); + const errSpy = spyOn(console, 'error').mockImplementation(() => {}); + try { + await withEnv({ GBRAIN_SOURCE: undefined }, async () => { + try { + await runAgentRun(engine, ['--detach', '--source', '__all__', 'write', 'a', 'page']); + throw new Error('expected runAgentRun to exit'); + } catch (e: any) { + expect(e.message).toBe('EXIT'); + } + }); + expect(spy).toHaveBeenCalledWith(2); + const rows = await engine.executeRaw<{ id: number }>( + `SELECT id FROM minion_jobs WHERE name = 'subagent'`, + ); + expect(rows.length).toBe(0); + } finally { + spy.mockRestore(); + errSpy.mockRestore(); + } + }); + + test('--source pointing at a nonexistent id surfaces a clean error, not a stack trace', async () => { + const spy = spyOn(process, 'exit').mockImplementation(() => { throw new Error('EXIT'); }); + const errSpy = spyOn(console, 'error').mockImplementation(() => {}); + try { + await withEnv({ GBRAIN_SOURCE: undefined }, async () => { + try { + await runAgentRun(engine, ['--detach', '--source', 'does-not-exist', 'write', 'a', 'page']); + throw new Error('expected runAgentRun to exit'); + } catch (e: any) { + expect(e.message).toBe('EXIT'); + } + }); + expect(spy).toHaveBeenCalledWith(1); + expect(errSpy.mock.calls.some(call => String(call[0]).includes('not found'))).toBe(true); + const rows = await engine.executeRaw<{ id: number }>( + `SELECT id FROM minion_jobs WHERE name = 'subagent'`, + ); + expect(rows.length).toBe(0); + } finally { + spy.mockRestore(); + errSpy.mockRestore(); + } + }); + + test('--source pointing at an archived source is rejected with a restore hint', async () => { + const spy = spyOn(process, 'exit').mockImplementation(() => { throw new Error('EXIT'); }); + const errSpy = spyOn(console, 'error').mockImplementation(() => {}); + try { + await engine.executeRaw(`UPDATE sources SET archived = true WHERE id = 'corporate'`); + await withEnv({ GBRAIN_SOURCE: undefined }, async () => { + try { + await runAgentRun(engine, ['--detach', '--source', 'corporate', 'write', 'a', 'page']); + throw new Error('expected runAgentRun to exit'); + } catch (e: any) { + expect(e.message).toBe('EXIT'); + } + }); + expect(spy).toHaveBeenCalledWith(1); + expect(errSpy.mock.calls.some(call => String(call[0]).includes('archived'))).toBe(true); + const rows = await engine.executeRaw<{ id: number }>( + `SELECT id FROM minion_jobs WHERE name = 'subagent'`, + ); + expect(rows.length).toBe(0); + } finally { + await engine.executeRaw(`UPDATE sources SET archived = false WHERE id = 'corporate'`); + spy.mockRestore(); + errSpy.mockRestore(); + } + }); +}); + describe('fan-out manifest shape (integration)', () => { test('fanout-manifest with 3 entries creates 3 subagent children + 1 aggregator', async () => { // Manually replicate what runAgentRun does for --fanout-manifest > 1.