triton6564685/gbrain
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: shell job type + worker abort-path fix (v0.13.0) (#217)

Browse Source 
* feat(minions): add protected-name constant + ctx.shutdownSignal

Introduce PROTECTED_JOB_NAMES ('shell') in a side-effect-free core module
so queue.ts can check it without importing from handlers/. MinionJobContext
gains shutdownSignal (distinct from signal) — handlers that need to run
SIGTERM-triggered cleanup subscribe to both; most handlers ignore shutdown
and run through the worker's 30s cleanup race to natural completion.

* fix(minions): MinionQueue.add gains trusted 4th arg + trim-normalized guard

Adds allowProtectedSubmit opt-in as a separate 4th parameter (NOT folded into
opts) so callers spreading user-provided opts ({...userOpts}) can't accidentally
carry the trust flag. PROTECTED_JOB_NAMES check runs on the trimmed name BEFORE
insert, closing the queue.add(' shell ', ...) whitespace bypass that would have
evaded a has(name) check.

* fix(minions): worker calls failJob on abort + wires ctx.shutdownSignal

Pre-v0.13.0 worker returned silently when ctx.signal.aborted fired, leaving
jobs in 'active' until stall sweep. Handlers using cooperative cancel had
no deterministic status flip — timeout/cancel/lock-loss all looked the same
from downstream callers (gbrain jobs get, --follow loops).

Fix: derive abort reason from abort.signal.reason ('timeout' | 'cancel' |
'lock-lost' | 'shutdown') and call failJob with 'aborted: <reason>' text.
failJob is idempotent via token+status match, so no-op when another path
already flipped status (handleTimeouts, cancelJob, stall).

Also: new shutdownAbort (instance-level AbortController) fires on process
SIGTERM/SIGINT and propagates to every handler's ctx.shutdownSignal.
Shell handler listens to both signals and runs SIGTERM→5s→SIGKILL on its
child on either; other handlers only listen to ctx.signal so deploy
restarts don't cancel them mid-flight.

* feat(minions): add shell job handler + submission audit log

New 'shell' job type spawns arbitrary commands under the Minions worker.
Deterministic cron scripts (API fetch, token refresh, scrape+write) can
move off the LLM gateway — zero Opus tokens per fire.

Handler contract:
- cmd or argv (exactly one required). cmd spawns via /bin/sh -c (absolute
  path, not 'sh', to block PATH-override shell substitution). argv spawns
  direct with no shell.
- cwd required, must be absolute. Operator-trust boundary.
- env defaults to SHELL_ENV_ALLOWLIST ({PATH, HOME, USER, LANG, TZ,
  NODE_ENV}) picked from process.env, with caller overrides merged on top.
  Prevents accidental $OPENAI_API_KEY interpolation into scripts.
- stdout/stderr retained as UTF-8-safe tails (64KB/16KB) via
  string_decoder.StringDecoder. Prepends [truncated N bytes] marker.
- Abort (either ctx.signal or ctx.shutdownSignal) fires SIGTERM → 5s grace
  → SIGKILL on child. Timer NOT .unref'd so worker's 30s race waits for
  the child to actually die.

shell-audit.ts writes a JSONL line per submission to
~/.gbrain/audit/shell-jobs-YYYY-Www.jsonl (ISO-week rotated, override via
GBRAIN_AUDIT_DIR). argv logged as JSON array (not space-joined, which would
flatten args with spaces). Never logs env values. Best-effort writes:
failures log to stderr but don't block submission.

* feat(jobs): submit_job MCP guard + CLI --timeout-ms + starvation warning

submit_job operation gains timeout_ms param (was missing — couldn't plumb
the existing MinionJobInput field through from either CLI or MCP). When
ctx.remote=true and name is in PROTECTED_JOB_NAMES, throws
OperationError('permission_denied'). Combined with the queue.add trusted
guard, MCP callers can never submit shell jobs even if the env flag is on.

CLI submit: new --timeout-ms N flag. Passes {allowProtectedSubmit:true}
as the 4th arg to queue.add only when the submitted name is protected
(not blanket-set for every job). Prints a starvation-warning block to
stderr when a shell job is submitted without --follow, pointing at both
--follow and 'gbrain jobs work' remediation. Fires for every shell submit
regardless of the submitter's env — the submitter env is a weak proxy for
the worker env.

Worker handler registration: conditional on GBRAIN_ALLOW_SHELL_JOBS=1.
Default: off. 'gbrain jobs submit --help' now lists handler types with a
pointer to docs/guides/minions-shell-jobs.md for shell.

* test(minions): 40 unit + 4 E2E cases for shell handler

Unit (test/minions-shell.test.ts):
- Protected names: trim-normalized, case-sensitive, whitespace bypass defense
- MinionQueue.add: trusted opt-in, whitespace bypass, non-protected untouched
- Handler validation: cmd|argv exclusive, cwd required/absolute, env strings
- Spawn: cmd/argv happy paths, non-zero exit, ENOENT, result shape
- Env allowlist: leaked-secret blocked, PATH inherited, caller override
- Abort: ctx.signal, ctx.shutdownSignal, pre-aborted signal
- Audit: ISO-week year boundary (2027-01-01 → W53 2026), mid-year W52/W53,
  GBRAIN_AUDIT_DIR override, argv as JSON array, env never logged, EACCES
  non-blocking
- Output truncation: 100KB → last 64KB with [truncated N bytes] marker

E2E (test/e2e/minions-shell.test.ts):
- Full lifecycle: submit → worker claim → spawn → complete
- MinionQueue.add without trusted arg throws (including whitespace bypass)
- submit_job with ctx.remote=true rejects shell (MCP guard)
- submit_job with ctx.remote=false allows shell (CLI path)

* chore: bump version and changelog (v0.13.0)

Move gateway crons to Minions. Zero LLM tokens per cron fire.
Worker abort path finally marks aborted jobs dead.

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

* docs: reframe v0.13.0 copy for OpenClaw operators (not Wintermute-specific)

gbrain is an open-source product for any OpenClaw/Hermes operator, not
Garry's personal Wintermute deployment. Rewords the v0.13.0 CHANGELOG
entry, the minions-shell-jobs guide, and the deferred TODOS entries to
speak to "your OpenClaw" / "OpenClaw operators" instead.

Replaces /data/wintermute cwd examples with the canonical
/data/.openclaw/workspace path. Pre-existing Wintermute references in
older CHANGELOG entries (v0.11/v0.10.3) left unchanged.

* feat(migrations): add v0.13.0 adoption playbook for shell jobs

Adding the migration file the CEO review originally scoped out. Without
it, operators upgrade to v0.13.0 and the capability ships but adoption
doesn't happen — the 60% gateway CPU reduction only lands if someone
actually rewrites their crontab.

skills/migrations/v0.13.0.md is the instruction manual the host agent
reads on gbrain upgrade:

- Enable worker: GBRAIN_ALLOW_SHELL_JOBS=1 gbrain jobs work (Postgres)
  or per-tick --follow (PGLite)
- Audit cron manifest: classify LLM-requiring vs deterministic
- Propose per-cron rewrites with diffs, approved one at a time
- Env allowlist guidance for scripts that need API keys
- Verification playbook: run one fire, compare pre/post, only then
  approve the next batch
- Starvation sanity-check runbook item

Iron rules: never auto-rewrite the operator's crontab (host-specific
code per CLAUDE.md). LLM-requiring crons stay on the gateway. Ambiguous
cases ask the operator.

No mechanical orchestrator ships with this migration — every rewrite
is operator judgment. A future gbrain crontab-to-minions helper is
tracked in TODOS.md as P1.

* docs: sync UPGRADING + SKILLPACK with v0.13.0 shell jobs

UPGRADING_DOWNSTREAM_AGENTS.md: append v0.13.0 section per the file's
convention (each release appends). No skill edits required, feature is
off-by-default, optional adoption via skills/migrations/v0.13.0.md.
Lists typical LLM-vs-deterministic classifications so operators know
which of their crons are candidates for migration.

GBRAIN_SKILLPACK.md: add shell-jobs guide row to the cron/Minions guide
table so it's discoverable alongside existing Cron via Minions, Plugin
Handlers, and Minions fix guides.

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Garry Tan
2026-04-20 10:54:31 +08:00
committed by GitHub
parent c89aa909c7
commit 5fd9cd2644
19 changed files with 1580 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

76
src/commands/jobs.ts
View File

@@ -57,8 +57,8 @@ export async function runJobs(engine: BrainEngine, args: string[]): Promise<void
USAGE
gbrain jobs submit <name> [--params JSON] [--follow] [--priority N]
[--delay Nms] [--max-attempts N] [--queue Q]
[--dry-run]
[--delay Nms] [--timeout-ms Nms] [--max-attempts N]
[--queue Q] [--dry-run]
gbrain jobs list [--status S] [--queue Q] [--limit N]
gbrain jobs get <id>
gbrain jobs cancel <id>
@@ -68,6 +68,18 @@ USAGE
gbrain jobs stats
gbrain jobs smoke
gbrain jobs work [--queue Q] [--concurrency N]
HANDLER TYPES (built in)
sync Pull and embed new pages from the repo
embed (Re-)embed pages; --params '{"slug":...}' or '{"all":true}'
lint Run page linter; --params '{"dir":"...","fix":true}'
import Bulk import markdown; --params '{"dir":"..."}'
extract Extract links + timeline entries; '{"mode":"all"}'
backlinks Check or fix back-links; '{"action":"fix"}'
autopilot-cycle One autopilot pass (sync+extract+embed+backlinks)
shell Run a command or argv. Requires GBRAIN_ALLOW_SHELL_JOBS=1
on the worker. Params: {cmd?, argv?, cwd, env?}.
See: docs/guides/minions-shell-jobs.md
`);
return;
}
@@ -93,6 +105,12 @@ USAGE
const delay = parseInt(parseFlag(args, '--delay') ?? '0', 10);
const maxAttempts = parseInt(parseFlag(args, '--max-attempts') ?? '3', 10);
const queueName = parseFlag(args, '--queue') ?? 'default';
const timeoutMsRaw = parseFlag(args, '--timeout-ms');
const timeoutMs = timeoutMsRaw !== undefined ? parseInt(timeoutMsRaw, 10) : undefined;
if (timeoutMsRaw !== undefined && (isNaN(timeoutMs!) || timeoutMs! <= 0)) {
console.error('Error: --timeout-ms must be a positive integer (milliseconds)');
process.exit(1);
}
const dryRun = hasFlag(args, '--dry-run');
const follow = hasFlag(args, '--follow');
@@ -103,6 +121,7 @@ USAGE
console.log(` Priority: ${priority}`);
console.log(` Max attempts: ${maxAttempts}`);
if (delay > 0) console.log(` Delay: ${delay}ms`);
if (timeoutMs) console.log(` Timeout: ${timeoutMs}ms`);
console.log(` Data: ${JSON.stringify(data)}`);
return;
}
@@ -114,12 +133,51 @@ USAGE
process.exit(1);
}
// The CLI path is a trusted submitter. Pass {allowProtectedSubmit: true}
// ONLY for protected names, not blanket-set for every submission, so any
// future protected name forces explicit opt-in at the call site.
const { isProtectedJobName } = await import('../core/minions/protected-names.ts');
const trusted = isProtectedJobName(name) ? { allowProtectedSubmit: true } : undefined;
const job = await queue.add(name, data, {
priority,
delay: delay > 0 ? delay : undefined,
max_attempts: maxAttempts,
queue: queueName,
});
timeout_ms: timeoutMs,
}, trusted);
// Submission audit log (operational trace, not forensic insurance).
try {
const { logShellSubmission } = await import('../core/minions/handlers/shell-audit.ts');
if (name.trim() === 'shell') {
logShellSubmission({
caller: 'cli',
remote: false,
job_id: job.id,
cwd: typeof data.cwd === 'string' ? data.cwd : '',
cmd_display: typeof data.cmd === 'string' ? data.cmd.slice(0, 80) : undefined,
argv_display: Array.isArray(data.argv)
? (data.argv as unknown[]).filter((a): a is string => typeof a === 'string').map((a) => a.slice(0, 80))
: undefined,
});
}
} catch { /* audit failures never block submission */ }
// Starvation warning (DX polish). Fire for every non-`--follow` shell submit
// regardless of the submitter's own `GBRAIN_ALLOW_SHELL_JOBS` — the submitter
// env is a weak proxy for the worker env (they may run on different machines),
// so the warning remains useful any time the job might sit in 'waiting'.
if (!follow && name.trim() === 'shell') {
process.stderr.write(
`\n⚠ Shell jobs require GBRAIN_ALLOW_SHELL_JOBS=1 on the worker process.\n` +
` Your job was queued (id=${job.id}) but will sit in 'waiting' until a\n` +
` worker with the env flag starts. To run now:\n\n` +
` GBRAIN_ALLOW_SHELL_JOBS=1 gbrain jobs submit shell \\\n` +
` --params '...' --follow\n\n` +
` Or start a persistent worker (Postgres only — PGLite uses --follow):\n\n` +
` GBRAIN_ALLOW_SHELL_JOBS=1 gbrain jobs work\n\n`,
);
}
if (follow) {
console.log(`Job #${job.id} submitted (${name}). Executing inline...`);
@@ -470,4 +528,16 @@ export async function registerBuiltinHandlers(worker: MinionWorker, engine: Brai
}
return { partial: false, steps };
});
// Shell handler: registered ONLY when GBRAIN_ALLOW_SHELL_JOBS=1 is set on the
// worker process. Default-closed; opt-in per-host. Without the flag, shell
// jobs submitted via CLI insert rows but no worker claims them (they sit in
// 'waiting' — the CLI prints a starvation warning for that case).
if (process.env.GBRAIN_ALLOW_SHELL_JOBS === '1') {
const { shellHandler } = await import('../core/minions/handlers/shell.ts');
worker.register('shell', shellHandler);
process.stderr.write('[minion worker] shell handler enabled (GBRAIN_ALLOW_SHELL_JOBS=1)\n');
} else {
process.stderr.write('[minion worker] shell handler disabled (set GBRAIN_ALLOW_SHELL_JOBS=1 to enable)\n');
}
}

75
src/core/minions/handlers/shell-audit.ts Normal file
View File

@@ -0,0 +1,75 @@
/**
* Shell-job submission audit log (operational trace, NOT forensic insurance).
*
* Writes a JSONL line per shell-job submission to `~/.gbrain/audit/shell-jobs-YYYY-Www.jsonl`
* (ISO week rotation, override via `GBRAIN_AUDIT_DIR`). Best-effort: write failures go
* to stderr and never block submission, which means a disk-full attacker could silently
* disable the trail. CHANGELOG calls this out honestly: it's for debugging "what did
* this cron submit last Tuesday?", not for security-critical forensics.
*
* Never logs `env` values (may contain secrets). Does log `cmd` and `argv` truncated to
* 80 chars for cmd / stored as JSON array for argv — the command text itself can contain
* inline tokens (`curl -H 'Authorization: Bearer ...'`) and the guide explicitly tells
* operators to put secrets in `env:` instead of embedding them in the command line.
*/
import * as fs from 'node:fs';
import * as path from 'node:path';
import * as os from 'node:os';
export interface ShellAuditEvent {
ts: string;
caller: 'cli' | 'mcp';
remote: boolean;
job_id: number;
cwd: string;
cmd_display?: string; // first 80 chars of cmd; may contain inline tokens
argv_display?: string[]; // each arg truncated individually to preserve separation
}
/** Compute `shell-jobs-YYYY-Www.jsonl` using ISO-8601 week numbering.
*
* Year-boundary edge: 2027-01-01 is ISO week 53 of year 2026, so the correct
* filename is `shell-jobs-2026-W53.jsonl`. This matches the ISO week standard
* (week containing the first Thursday of the year is W1; week containing Dec 28
* is always W52 or W53 of that year).
*/
export function computeAuditFilename(now: Date = new Date()): string {
// Copy date and move to nearest Thursday (ISO week anchor).
const d = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate()));
const dayNum = (d.getUTCDay() + 6) % 7; // Mon=0, Sun=6
d.setUTCDate(d.getUTCDate() - dayNum + 3); // shift to Thursday
const isoYear = d.getUTCFullYear();
const firstThursday = new Date(Date.UTC(isoYear, 0, 4));
const firstThursdayDayNum = (firstThursday.getUTCDay() + 6) % 7;
firstThursday.setUTCDate(firstThursday.getUTCDate() - firstThursdayDayNum + 3);
const weekNum = Math.round((d.getTime() - firstThursday.getTime()) / (7 * 86400000)) + 1;
const ww = String(weekNum).padStart(2, '0');
return `shell-jobs-${isoYear}-W${ww}.jsonl`;
}
/** Resolve the audit dir. Honors `GBRAIN_AUDIT_DIR` for container/sandbox deployments
* where `$HOME` is read-only. Defaults to `~/.gbrain/audit/`. */
export function resolveAuditDir(): string {
const override = process.env.GBRAIN_AUDIT_DIR;
if (override && override.trim().length > 0) return override;
return path.join(os.homedir(), '.gbrain', 'audit');
}
export function logShellSubmission(event: Omit<ShellAuditEvent, 'ts'>): void {
const dir = resolveAuditDir();
const filename = computeAuditFilename();
const fullPath = path.join(dir, filename);
const line = JSON.stringify({ ...event, ts: new Date().toISOString() }) + '\n';
try {
fs.mkdirSync(dir, { recursive: true });
fs.appendFileSync(fullPath, line, { encoding: 'utf8' });
} catch (err) {
// Best-effort: log to stderr and keep going. A disk-full or EACCES attacker
// can silently disable this trail, which is why CHANGELOG calls it an
// operational trace, not forensic insurance.
const msg = err instanceof Error ? err.message : String(err);
process.stderr.write(`[shell-audit] write failed (${msg}); submission continues\n`);
}
}

311
src/core/minions/handlers/shell.ts Normal file
View File

@@ -0,0 +1,311 @@
/**
* `shell` job handler.
*
* Runs an arbitrary shell command or argv vector as a child process under the
* Minions worker. Purpose: move deterministic cron scripts (API fetch, token
* refresh, scrape + write) off the LLM gateway so they don't consume an Opus
* session each time.
*
* Security (both gates must pass):
* 1. `MinionQueue.add()` rejects name='shell' unless the caller explicitly
* opts in via `trusted.allowProtectedSubmit`. CLI path and the `submit_job`
* operation (when `ctx.remote === false`) set the flag. MCP callers don't.
* 2. This handler only registers when `process.env.GBRAIN_ALLOW_SHELL_JOBS === '1'`.
* Default: off. Without the flag the worker's `registeredNames` excludes
* shell and queued jobs stay in 'waiting'.
*
* Env model (honest): the child process receives a small allowlist (PATH, HOME,
* USER, LANG, TZ, NODE_ENV) merged with caller-supplied `job.data.env`. This
* prevents the accidental `$OPENAI_API_KEY` interpolation footgun. It does NOT
* sandbox filesystem reads — a shell script can `cat ~/.env` or any file the
* worker can read. The operator picks a safe `cwd`; that's the trust boundary.
*
* Shutdown: the handler listens to BOTH `ctx.signal` (timeout/cancel/lock-loss)
* and `ctx.shutdownSignal` (worker process SIGTERM). Either triggers the same
* kill sequence: SIGTERM → 5s grace → SIGKILL. Non-shell handlers ignore
* `shutdownSignal` so deploy restarts don't interrupt them mid-flight.
*/
import { spawn, type ChildProcess } from 'node:child_process';
import { StringDecoder } from 'node:string_decoder';
import * as path from 'node:path';
import type { MinionJobContext } from '../types.ts';
import { UnrecoverableError } from '../types.ts';
/** Environment variables passed through to shell children by default. Callers
* that need additional keys (e.g. a specific API token for a cron) must name
* them explicitly in `job.data.env`. Named keys override this allowlist. */
const SHELL_ENV_ALLOWLIST = ['PATH', 'HOME', 'USER', 'LANG', 'TZ', 'NODE_ENV'] as const;
/** Max bytes retained from stdout/stderr. Output exceeding these caps is
* truncated with a `[truncated N bytes]` marker. UTF-8-safe via StringDecoder. */
const STDOUT_TAIL_MAX_BYTES = 64 * 1024;
const STDERR_TAIL_MAX_BYTES = 16 * 1024;
/** Grace period between SIGTERM and SIGKILL. Well-behaved scripts catch SIGTERM,
* flush state, exit cleanly; non-behaving scripts get reaped. */
const KILL_GRACE_MS = 5000;
export interface ShellJobParams {
/** Shell command. Spawned via `/bin/sh -c cmd`. Exactly one of cmd or argv is required. */
cmd?: string;
/** Argv vector. Spawned directly without a shell. Exactly one of cmd or argv is required. */
argv?: string[];
/** Working directory. REQUIRED, must be an absolute path. The operator chooses
* this; it's the trust boundary for what files the script can read/write. */
cwd: string;
/** Additional env vars to pass to the child. Merged on top of SHELL_ENV_ALLOWLIST. */
env?: Record<string, string>;
}
export interface ShellJobResult {
exit_code: number;
stdout_tail: string;
stderr_tail: string;
duration_ms: number;
pid: number;
}
/** Validate and narrow `job.data` to ShellJobParams. Throws UnrecoverableError
* for misshapen input — validation failures are not retry-worthy. */
function validateParams(data: Record<string, unknown>): ShellJobParams {
const hasCmd = typeof data.cmd === 'string' && data.cmd.length > 0;
const hasArgv = Array.isArray(data.argv) && data.argv.length > 0;
if (hasCmd && hasArgv) {
throw new UnrecoverableError(
'shell: specify exactly one of cmd or argv (see: docs/guides/minions-shell-jobs.md#errors)',
);
}
if (!hasCmd && !hasArgv) {
throw new UnrecoverableError(
'shell: specify exactly one of cmd or argv (see: docs/guides/minions-shell-jobs.md#errors)',
);
}
if (hasArgv) {
const argvOk = (data.argv as unknown[]).every((a) => typeof a === 'string');
if (!argvOk) {
throw new UnrecoverableError(
'shell: argv must be an array of strings (see: docs/guides/minions-shell-jobs.md#errors)',
);
}
}
if (typeof data.cwd !== 'string' || data.cwd.length === 0) {
throw new UnrecoverableError(
'shell: cwd is required and must be an absolute path (see: docs/guides/minions-shell-jobs.md#errors)',
);
}
if (!path.isAbsolute(data.cwd)) {
throw new UnrecoverableError(
'shell: cwd is required and must be an absolute path (see: docs/guides/minions-shell-jobs.md#errors)',
);
}
if (data.env !== undefined) {
if (typeof data.env !== 'object' || data.env === null || Array.isArray(data.env)) {
throw new UnrecoverableError(
'shell: env must be an object of string values (see: docs/guides/minions-shell-jobs.md#errors)',
);
}
for (const v of Object.values(data.env as Record<string, unknown>)) {
if (typeof v !== 'string') {
throw new UnrecoverableError(
'shell: env values must all be strings (see: docs/guides/minions-shell-jobs.md#errors)',
);
}
}
}
return {
cmd: hasCmd ? (data.cmd as string) : undefined,
argv: hasArgv ? (data.argv as string[]) : undefined,
cwd: data.cwd,
env: (data.env as Record<string, string> | undefined),
};
}
/** Build the child process env: SHELL_ENV_ALLOWLIST picked from process.env,
* overlaid with caller-supplied `job.data.env`. Prevents accidental leak of
* OPENAI_API_KEY / DATABASE_URL / etc. into user-authored scripts. */
function buildChildEnv(override: Record<string, string> | undefined): Record<string, string> {
const env: Record<string, string> = {};
for (const key of SHELL_ENV_ALLOWLIST) {
const v = process.env[key];
if (typeof v === 'string') env[key] = v;
}
if (override) {
for (const [k, v] of Object.entries(override)) env[k] = v;
}
return env;
}
/** Bounded-length UTF-8-safe tail buffer. Accumulates bytes via StringDecoder
* so the last `maxBytes` of output is character-safe (no split multibyte chars).
* On truncation, the emitted string is prefixed with `[truncated N bytes]`. */
class TailBuffer {
private decoder = new StringDecoder('utf8');
private body = '';
private bodyBytes = 0;
private truncatedBytes = 0;
constructor(private readonly maxBytes: number) {}
append(chunk: Buffer): void {
const str = this.decoder.write(chunk);
if (str.length === 0) return;
this.body += str;
this.bodyBytes = Buffer.byteLength(this.body, 'utf8');
this.compactIfOver();
}
private compactIfOver(): void {
if (this.bodyBytes <= this.maxBytes) return;
// We need to keep only the trailing maxBytes. Byte-slicing mid-character is
// unsafe; instead, find the highest character offset whose byte length from
// that point is <= maxBytes. Linear-scan from the end over grapheme-safe
// codepoints is good enough at 64KB scales.
const targetByteSize = this.maxBytes;
// Fast path: if body is all ASCII (1 byte per char), byteLength === length.
if (this.body.length === this.bodyBytes) {
const drop = this.bodyBytes - targetByteSize;
this.truncatedBytes += drop;
this.body = this.body.slice(drop);
this.bodyBytes = targetByteSize;
return;
}
// Slow path: find a character boundary that lands just under maxBytes.
// Scan from the end; accumulate bytes per codepoint.
let tailBytes = 0;
let cut = this.body.length;
for (let i = this.body.length - 1; i >= 0; i--) {
const code = this.body.codePointAt(i);
const cpBytes = code === undefined ? 0
: code < 0x80 ? 1
: code < 0x800 ? 2
: code < 0x10000 ? 3
: 4;
if (tailBytes + cpBytes > targetByteSize) break;
tailBytes += cpBytes;
cut = i;
}
const droppedBytes = this.bodyBytes - tailBytes;
this.truncatedBytes += droppedBytes;
this.body = this.body.slice(cut);
this.bodyBytes = tailBytes;
}
done(): string {
const tail = this.decoder.end();
if (tail.length > 0) {
this.body += tail;
this.bodyBytes = Buffer.byteLength(this.body, 'utf8');
this.compactIfOver();
}
if (this.truncatedBytes === 0) return this.body;
return `[truncated ${this.truncatedBytes} bytes]\n${this.body}`;
}
}
/** The shell handler itself. */
export async function shellHandler(ctx: MinionJobContext): Promise<ShellJobResult> {
const params = validateParams(ctx.data);
const env = buildChildEnv(params.env);
const startedAt = Date.now();
let proc: ChildProcess;
try {
if (params.cmd) {
// Absolute /bin/sh — not 'sh' — so a caller-supplied env with a poisoned
// PATH can't redirect to a different shell binary.
proc = spawn('/bin/sh', ['-c', params.cmd], {
cwd: params.cwd,
env,
stdio: ['ignore', 'pipe', 'pipe'],
});
} else {
const argv = params.argv!;
proc = spawn(argv[0], argv.slice(1), {
cwd: params.cwd,
env,
stdio: ['ignore', 'pipe', 'pipe'],
});
}
} catch (err) {
// Spawn-phase failure (e.g. cwd doesn't exist when using '/bin/sh' directly).
// Retryable.
throw err instanceof Error ? err : new Error(String(err));
}
const pid = proc.pid ?? -1;
const stdoutTail = new TailBuffer(STDOUT_TAIL_MAX_BYTES);
const stderrTail = new TailBuffer(STDERR_TAIL_MAX_BYTES);
proc.stdout?.on('data', (c: Buffer) => stdoutTail.append(c));
proc.stderr?.on('data', (c: Buffer) => stderrTail.append(c));
// Wire BOTH signals to the kill sequence. `ctx.signal` fires on timeout /
// cancel / lock-loss; `ctx.shutdownSignal` fires only on worker SIGTERM/SIGINT.
// Shell handler needs both — a deploy restart shouldn't leave children running
// past the 30s worker cleanup race.
let killTimer: ReturnType<typeof setTimeout> | null = null;
let killReason = '';
const onAbort = (label: string) => () => {
if (killTimer !== null) return; // already started
killReason = label;
if (!proc.killed) {
try { proc.kill('SIGTERM'); } catch { /* proc already exited */ }
}
killTimer = setTimeout(() => {
if (!proc.killed) {
try { proc.kill('SIGKILL'); } catch { /* already exited */ }
}
}, KILL_GRACE_MS);
};
const sigAbort = onAbort('signal');
const shutdownAbort = onAbort('shutdown');
ctx.signal.addEventListener('abort', sigAbort);
ctx.shutdownSignal.addEventListener('abort', shutdownAbort);
// Fire immediately if either already aborted before wiring
if (ctx.signal.aborted) sigAbort();
if (ctx.shutdownSignal.aborted) shutdownAbort();
const exitCode: number = await new Promise((resolve, reject) => {
proc.on('error', (err) => {
reject(err);
});
proc.on('exit', (code, signal) => {
// Node maps signal-terminated exits to a 128+N code convention; we use
// whichever is defined.
if (code !== null) resolve(code);
else if (signal === 'SIGTERM') resolve(143);
else if (signal === 'SIGKILL') resolve(137);
else resolve(-1);
});
}).finally(() => {
if (killTimer !== null) clearTimeout(killTimer);
ctx.signal.removeEventListener('abort', sigAbort);
ctx.shutdownSignal.removeEventListener('abort', shutdownAbort);
});
const duration_ms = Date.now() - startedAt;
const stdout_tail = stdoutTail.done();
const stderr_tail = stderrTail.done();
// If we sent SIGTERM/SIGKILL in response to an abort, surface that as the
// error rather than the exit code — clearer for debugging. Worker catch
// handles retry/dead classification.
if (killReason === 'signal' || killReason === 'shutdown') {
const err = new Error(
`aborted: ${killReason === 'shutdown' ? 'shutdown' : (ctx.signal.reason as Error)?.message || 'signal'}`,
);
throw err;
}
if (exitCode !== 0) {
throw new Error(
`exit ${exitCode}: ${stderr_tail.slice(-500)}`,
);
}
return { exit_code: exitCode, stdout_tail, stderr_tail, duration_ms, pid };
}

20
src/core/minions/protected-names.ts Normal file
View File

@@ -0,0 +1,20 @@
/**
* Protected job names — side-effect-free constant module.
*
* Names in this set require an explicit `trusted.allowProtectedSubmit: true` opt-in
* when passed to `MinionQueue.add()`. The CLI path and the `submit_job` operation
* (when `ctx.remote === false`) set the flag; MCP callers never do. Defense-in-depth
* against in-process handlers that programmatically submit a shell child via
* `queue.add('shell', ...)`.
*
* This file must stay pure — no imports from handlers, no filesystem, no env reads.
* Queue core imports it; if this module grew side effects, every queue user would
* pay them at module load.
*/
export const PROTECTED_JOB_NAMES: ReadonlySet<string> = new Set(['shell']);
/** Check a job name against the protected set. Normalizes whitespace first. */
export function isProtectedJobName(name: string): boolean {
return PROTECTED_JOB_NAMES.has(name.trim());
}

31
src/core/minions/queue.ts
View File

@@ -15,6 +15,16 @@ import type {
} from './types.ts';
import { rowToMinionJob, rowToInboxMessage, rowToAttachment } from './types.ts';
import { validateAttachment } from './attachments.ts';
import { isProtectedJobName } from './protected-names.ts';
/** Options for opting into protected-job-name submission. Passed as a separate
* 4th arg to `MinionQueue.add()` (NOT folded into `opts`) so user-spread
* `{...userOpts}` payloads can't accidentally carry the trust flag. */
export interface TrustedSubmitOpts {
/** When true, allow submission of names in PROTECTED_JOB_NAMES (currently 'shell').
* Set only by the CLI path and by `submit_job` when `ctx.remote === false`. */
allowProtectedSubmit?: boolean;
}
const MIGRATION_VERSION = 7;
@@ -55,10 +65,25 @@ export class MinionQueue {
* to 'waiting-children' atomically. Idempotency_key dedups via PG unique
* partial index; same key returns the existing row (no second insert).
*/
async add(name: string, data?: Record<string, unknown>, opts?: Partial<MinionJobInput>): Promise<MinionJob> {
if (!name || name.trim().length === 0) {
async add(
name: string,
data?: Record<string, unknown>,
opts?: Partial<MinionJobInput>,
trusted?: TrustedSubmitOpts,
): Promise<MinionJob> {
// Normalize first so the protected-name check and the insert use the same
// canonical form. Without the trim-before-check, `queue.add(' shell ', ...)`
// would evade the guard and insert a job literally named 'shell'.
const jobName = (name || '').trim();
if (jobName.length === 0) {
throw new Error('Job name cannot be empty');
}
if (isProtectedJobName(jobName) && !trusted?.allowProtectedSubmit) {
throw new Error(
`protected job name '${jobName}' requires CLI or operation-local submitter ` +
`(pass {allowProtectedSubmit: true} as the 4th arg to MinionQueue.add)`,
);
}
await this.ensureSchema();
const childStatus: MinionJobStatus = opts?.delay ? 'delayed' : 'waiting';
@@ -126,7 +151,7 @@ export class MinionQueue {
RETURNING *`;
const params = [
name.trim(),
jobName,
opts?.queue ?? 'default',
childStatus,
opts?.priority ?? 0,

7
src/core/minions/types.ts
View File

@@ -159,8 +159,13 @@ export interface MinionJobContext {
name: string;
data: Record<string, unknown>;
attempts_made: number;
/** AbortSignal for cooperative cancellation (fires on pause or lock loss). */
/** AbortSignal for cooperative cancellation (fires on timeout, cancel, pause, or lock loss). */
signal: AbortSignal;
/** AbortSignal that fires only on worker process SIGTERM/SIGINT. Handlers sensitive
* to deploy restarts (e.g. the shell handler, which must run a SIGTERM → 5s → SIGKILL
* sequence on its child) listen to this in addition to `signal`. Most handlers can
* ignore it — workers give them the full 30s cleanup race to finish naturally. */
shutdownSignal: AbortSignal;
/** Update structured progress (not just 0-100). */
updateProgress(progress: unknown): Promise<void>;
/** Accumulate token usage for this job. */

44
src/core/minions/worker.ts
View File

@@ -49,6 +49,13 @@ export class MinionWorker {
private inFlight = new Map<number, InFlightJob>();
private workerId = randomUUID();
/** Fires only on worker process SIGTERM/SIGINT. Handlers that need to run
* shutdown-specific cleanup (e.g. shell handler's SIGTERM→SIGKILL sequence on
* its child) subscribe via `ctx.shutdownSignal`. Separated from the per-job
* abort controller so non-shell handlers don't get cancelled mid-flight on
* deploy restart — they still get the full 30s cleanup race instead. */
private shutdownAbort = new AbortController();
private opts: Required<MinionWorkerOpts>;
constructor(
@@ -88,10 +95,16 @@ export class MinionWorker {
await this.queue.ensureSchema();
this.running = true;
// Graceful shutdown
// Graceful shutdown. Fires shutdownAbort so handlers subscribed to
// `ctx.shutdownSignal` (currently: shell handler) can run their own cleanup
// BEFORE the 30s cleanup race expires. Non-shell handlers ignore shutdown
// and keep running — they get the full 30s window.
const shutdown = () => {
console.log('Minion worker shutting down...');
this.running = false;
if (!this.shutdownAbort.signal.aborted) {
this.shutdownAbort.abort(new Error('shutdown'));
}
};
process.on('SIGTERM', shutdown);
process.on('SIGINT', shutdown);
@@ -246,7 +259,7 @@ export class MinionWorker {
if (!renewed) {
console.warn(`Lock lost for job ${job.id}, aborting execution`);
clearInterval(lockTimer);
abort.abort();
abort.abort(new Error('lock-lost'));
}
}, this.opts.lockDuration / 2);
@@ -260,7 +273,7 @@ export class MinionWorker {
timeoutTimer = setTimeout(() => {
if (!abort.signal.aborted) {
console.warn(`Job ${job.id} (${job.name}) hit per-job timeout (${job.timeout_ms}ms), aborting`);
abort.abort();
abort.abort(new Error('timeout'));
}
}, job.timeout_ms);
}
@@ -287,13 +300,18 @@ export class MinionWorker {
return;
}
// Build job context with per-job AbortSignal
// Build job context with per-job AbortSignal + shared shutdown signal.
// Most handlers only care about `signal` (timeout / cancel / lock-loss).
// `shutdownSignal` is separate: fires only on worker process SIGTERM/SIGINT.
// Handlers that need to run cleanup before worker exit (shell handler's
// SIGTERM→5s→SIGKILL on its child) subscribe to shutdownSignal too.
const context: MinionJobContext = {
id: job.id,
name: job.name,
data: job.data,
attempts_made: job.attempts_made,
signal: abort.signal,
shutdownSignal: this.shutdownAbort.signal,
updateProgress: async (progress: unknown) => {
await this.queue.updateProgress(job.id, lockToken, progress);
},
@@ -343,13 +361,23 @@ export class MinionWorker {
} catch (err) {
clearInterval(lockTimer);
// If aborted (paused or lock lost), don't try to fail the job
// If the per-job abort fired, derive the reason from signal.reason (set
// by whichever site aborted: 'timeout' / 'cancel' / 'lock-lost'). We call
// failJob unconditionally — the DB match on status='active' + lock_token
// makes it idempotent: if another path (handleTimeouts, cancelJob, stall)
// already flipped status, our call no-ops cleanly. The prior silent-return
// left jobs stranded in 'active' until a secondary sweep, breaking
// timeout/cancel contracts downstream callers rely on.
let errorText: string;
if (abort.signal.aborted) {
console.log(`Job ${job.id} (${job.name}) aborted (paused or lock lost)`);
return;
const reason = abort.signal.reason instanceof Error
? abort.signal.reason.message
: String(abort.signal.reason || 'aborted');
errorText = `aborted: ${reason}`;
} else {
errorText = err instanceof Error ? err.message : String(err);
}
const errorText = err instanceof Error ? err.message : String(err);
const isUnrecoverable = err instanceof UnrecoverableError;
const attemptsExhausted = job.attempts_made + 1 >= job.max_attempts;

28
src/core/operations.ts
View File

@@ -1047,26 +1047,44 @@ const file_url: Operation = {
const submit_job: Operation = {
name: 'submit_job',
description: 'Submit a background job to the Minions queue',
description: 'Submit a background job to the Minions queue. Built-in types: sync, embed, lint, import, extract, backlinks, autopilot-cycle. The `shell` type is CLI-only and rejected over MCP.',
params: {
name: { type: 'string', required: true, description: 'Job type (sync, embed, lint, import)' },
name: { type: 'string', required: true, description: 'Job type (sync, embed, lint, import, extract, backlinks, autopilot-cycle; shell is CLI-only)' },
data: { type: 'object', description: 'Job payload (JSON)' },
queue: { type: 'string', description: 'Queue name (default: "default")' },
priority: { type: 'number', description: 'Priority (0 = highest, default: 0)' },
max_attempts: { type: 'number', description: 'Max retry attempts (default: 3)' },
delay: { type: 'number', description: 'Delay in ms before eligible' },
timeout_ms: { type: 'number', description: 'Per-job wall-clock timeout in ms; aborted job goes to dead' },
},
mutating: true,
handler: async (ctx, p) => {
if (ctx.dryRun) return { dry_run: true, action: 'submit_job', name: p.name };
const name = typeof p.name === 'string' ? p.name.trim() : '';
if (ctx.dryRun) return { dry_run: true, action: 'submit_job', name };
// Submit-side MCP guard: reject protected job names from untrusted callers
// BEFORE we touch the DB. This is the first of the two security layers
// (the second is MinionQueue.add's check). Independent of the worker-side
// GBRAIN_ALLOW_SHELL_JOBS env flag — even if that flag is on, MCP callers
// cannot submit protected-type jobs.
const { isProtectedJobName } = await import('./minions/protected-names.ts');
if (ctx.remote && isProtectedJobName(name)) {
throw new OperationError('permission_denied', `'${name}' jobs cannot be submitted over MCP (CLI-only for security)`);
}
const { MinionQueue } = await import('./minions/queue.ts');
const queue = new MinionQueue(ctx.engine);
return queue.add(p.name as string, (p.data as Record<string, unknown>) || {}, {
// Trusted flag set only when this is a local (non-remote) submission. When
// remote=true, the guard above has already thrown for protected names, so
// passing undefined here is safe for any non-protected name that slips by.
const trusted = !ctx.remote && isProtectedJobName(name) ? { allowProtectedSubmit: true } : undefined;
return queue.add(name, (p.data as Record<string, unknown>) || {}, {
queue: (p.queue as string) || 'default',
priority: (p.priority as number) || 0,
max_attempts: (p.max_attempts as number) || 3,
delay: (p.delay as number) || undefined,
});
timeout_ms: (p.timeout_ms as number) || undefined,
}, trusted);
},
};