admin/gbrain
1
0
Fork 0
mirror of https://github.com/garrytan/gbrain.git synced 2026-08-14 00:48:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(agent): resolve the brain source at submit time instead of hardcoding the seed default (#3647)

Browse Source 
Wave-assembled from PR #3647 by @Masashi-Ono0611.

Co-Authored-By: masashiono0611 <masashi.ono.0611@gmail.com>
This commit is contained in:
test
2026-08-13 12:18:13 -07:00
committed by Sina Matian
co-authored by masashiono0611
parent b92cc967df
commit 0cfedd026d
2 changed files with 278 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/commands/agent.ts
+82 -3
View File
@@ -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 <n> Max assistant turns (default 20)
--tools a,b,c Subset of registered tool names (comma list)
--timeout-ms <n> Per-job wall-clock timeout
--source <id> 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 <path> 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 <id>` 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<string> {
// 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 <id>.`,
);
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<void> {
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<void> {
async function runFanout(engine: BrainEngine, queue: MinionQueue, flags: RunFlags, promptTemplate: string, sourceId: string): Promise<void> {
const manifestPath = flags.fanoutManifest!;
let manifest: Array<{ prompt?: string; input_vars?: Record<string, unknown> }>;
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 } : {}),
test/agent-cli.test.ts
+196 -2
View File
@@ -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<Record<string, unknown>> {
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<string, unknown>;
}
async function onlyJobData(): Promise<Record<string, unknown>> {
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.