mirror of
https://github.com/garrytan/gbrain.git
synced 2026-08-17 10:22:34 +00:00
Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
9391fb9317 |
+1
-1
@@ -2387,7 +2387,7 @@ JOBS (Minions)
|
||||
jobs get <id> Job details + history
|
||||
jobs cancel <id> Cancel job
|
||||
jobs retry <id> Re-queue failed/dead job
|
||||
jobs prune [--older-than 30d] [--status s,..] Clean old terminal jobs (0d = no age floor)
|
||||
jobs prune [--older-than 30d] Clean old jobs
|
||||
jobs stats Job health dashboard
|
||||
jobs work [--queue Q] Start worker daemon (Postgres only)
|
||||
|
||||
|
||||
+5
-33
@@ -106,21 +106,6 @@ export function parseMaxRssFlag(args: string[]): number | undefined {
|
||||
return parsed;
|
||||
}
|
||||
|
||||
/** Terminal statuses `jobs prune --status` accepts (PR #2282). Matches what
|
||||
* queue.prune can safely delete; anything else (waiting/active/…) is live. */
|
||||
export const PRUNE_STATUSES = ['completed', 'failed', 'dead', 'cancelled'] as const satisfies readonly MinionJobStatus[];
|
||||
|
||||
/** Parse a `--status a,b,c` value into prune statuses. Throws on any value
|
||||
* outside PRUNE_STATUSES (fail-fast, mirrors parseNiceValue). */
|
||||
export function parsePruneStatuses(raw: string): MinionJobStatus[] {
|
||||
const requested = raw.split(',').map(s => s.trim()).filter(Boolean);
|
||||
const invalid = requested.filter(s => !(PRUNE_STATUSES as readonly string[]).includes(s));
|
||||
if (requested.length === 0 || invalid.length > 0) {
|
||||
throw new Error(`--status accepts a comma-separated subset of [${PRUNE_STATUSES.join(', ')}]${invalid.length ? `. Invalid: ${invalid.join(', ')}` : ''}`);
|
||||
}
|
||||
return requested as MinionJobStatus[];
|
||||
}
|
||||
|
||||
/** Parse `--nice N` (then `GBRAIN_NICE` env). Returns:
|
||||
* - undefined if absent (no priority change — inherit)
|
||||
* - the validated integer in [-20, 19] otherwise
|
||||
@@ -223,9 +208,7 @@ USAGE
|
||||
gbrain jobs get <id>
|
||||
gbrain jobs cancel <id>
|
||||
gbrain jobs retry <id>
|
||||
gbrain jobs prune [--older-than 30d] [--status completed,failed,dead,cancelled]
|
||||
(--older-than 0d = no age floor: deletes ALL
|
||||
matching terminal jobs; pair with --status)
|
||||
gbrain jobs prune [--older-than 30d]
|
||||
gbrain jobs delete <id>
|
||||
gbrain jobs stats
|
||||
gbrain jobs smoke
|
||||
@@ -617,27 +600,16 @@ HANDLER TYPES (built in)
|
||||
case 'prune': {
|
||||
const olderThanStr = parseFlag(args, '--older-than') ?? '30d';
|
||||
const days = parseInt(olderThanStr, 10);
|
||||
if (isNaN(days) || days < 0) {
|
||||
console.error('Error: --older-than must be a non-negative number (days). Example: --older-than 30d; --older-than 0d removes the age floor (deletes ALL matching terminal jobs).');
|
||||
if (isNaN(days) || days <= 0) {
|
||||
console.error('Error: --older-than must be a positive number (days). Example: --older-than 30d');
|
||||
process.exit(1);
|
||||
}
|
||||
const statusFlag = parseFlag(args, '--status');
|
||||
let statuses: MinionJobStatus[] | undefined;
|
||||
if (statusFlag !== undefined) {
|
||||
try { statuses = parsePruneStatuses(statusFlag); }
|
||||
catch (e) { console.error(`Error: ${e instanceof Error ? e.message : String(e)}`); process.exit(1); }
|
||||
}
|
||||
|
||||
try { await queue.ensureSchema(); }
|
||||
catch (e) { console.error(e instanceof Error ? e.message : String(e)); process.exit(1); }
|
||||
|
||||
const count = await queue.prune({
|
||||
olderThan: new Date(Date.now() - days * 86400000),
|
||||
...(statuses ? { status: statuses } : {}),
|
||||
});
|
||||
const statusLabel = statuses ? statuses.join('+') : 'completed+dead+cancelled';
|
||||
const ageLabel = days === 0 ? 'regardless of age' : `older than ${days} days`;
|
||||
console.log(`Pruned ${count} ${statusLabel} jobs ${ageLabel}.`);
|
||||
const count = await queue.prune({ olderThan: new Date(Date.now() - days * 86400000) });
|
||||
console.log(`Pruned ${count} jobs older than ${days} days.`);
|
||||
break;
|
||||
}
|
||||
|
||||
|
||||
@@ -142,12 +142,16 @@ export async function runOnboard(engine: BrainEngine, args: string[]): Promise<v
|
||||
|
||||
// --auto path: runs through the T2 library orchestrator. Hooks emit CLI
|
||||
// progress to stderr; the final result lands as JSON on stdout (or human
|
||||
// summary).
|
||||
// summary). extraRemediations (gathered above from runAllOnboardChecks)
|
||||
// is threaded into the runner so the onboard-check remediations
|
||||
// (extract-ner, extract-timeline-from-meetings, etc.) reach the planner
|
||||
// — the same wiring the --check path uses above.
|
||||
const result = await runRemediation(
|
||||
engine,
|
||||
{
|
||||
targetScore,
|
||||
maxUsd,
|
||||
extraRemediations,
|
||||
// --auto --yes opts into the prompt_required tier too; library
|
||||
// doesn't distinguish auto_apply vs prompt_required, it just runs
|
||||
// every remediation in the plan. The plan-building side (T12 render)
|
||||
|
||||
@@ -4957,7 +4957,7 @@ const run_onboard: Operation = {
|
||||
// typo, the underlying queue.add would reject. Defense-in-depth.
|
||||
const result = await runRemediation(
|
||||
ctx.engine,
|
||||
{ targetScore, maxUsd },
|
||||
{ targetScore, maxUsd, extraRemediations: allowedExtras },
|
||||
{},
|
||||
);
|
||||
|
||||
|
||||
@@ -66,9 +66,10 @@ export async function runRemediation(
|
||||
} = await import('../remediation-checkpoint.ts');
|
||||
|
||||
const ctx = await loadRecommendationContext(engine);
|
||||
const extraRemediations = opts.extraRemediations ?? [];
|
||||
|
||||
// Pre-flight ceiling check via the shared plan computation.
|
||||
const initialPlan = await computeRemediationPlan(engine, { targetScore });
|
||||
const initialPlan = await computeRemediationPlan(engine, { targetScore, extraRemediations });
|
||||
if (initialPlan.target_unreachable) {
|
||||
hooks.onTargetUnreachable?.(targetScore, initialPlan.max_reachable_score);
|
||||
return {
|
||||
@@ -87,7 +88,7 @@ export async function runRemediation(
|
||||
}
|
||||
|
||||
const initialHealth = await engine.getHealth();
|
||||
let recs: RemediationStep[] = computeRecommendations(initialHealth, ctx)
|
||||
let recs: RemediationStep[] = computeRecommendations(initialHealth, ctx, extraRemediations)
|
||||
.filter((r) => r.status === 'remediable');
|
||||
if (recs.length === 0) {
|
||||
hooks.onNothingToDo?.(initialHealth.brain_score, targetScore);
|
||||
@@ -305,7 +306,13 @@ export async function runRemediation(
|
||||
// steps with bumped retry suffix (D1).
|
||||
if (recs.length === 0 || stepCount >= maxJobs) break;
|
||||
const freshHealth = await engine.getHealth();
|
||||
recs = computeRecommendations(freshHealth, ctx).filter((r) => r.status === 'remediable');
|
||||
// Extras carry a static status:'remediable' — a fresh health snapshot
|
||||
// never ages them out the way health-derived steps drop. Filter out
|
||||
// ids this run already processed (any terminal status), or the recheck
|
||||
// would resubmit completed extras every iteration, forever.
|
||||
const processedIds = new Set(submitted.map((s) => s.id));
|
||||
const pendingExtras = extraRemediations.filter((r) => !processedIds.has(r.id));
|
||||
recs = computeRecommendations(freshHealth, ctx, pendingExtras).filter((r) => r.status === 'remediable');
|
||||
}
|
||||
};
|
||||
|
||||
|
||||
@@ -63,6 +63,16 @@ export interface RemediationOpts {
|
||||
resumePlanHash?: string;
|
||||
/** Whether to attempt resume at all (default false). */
|
||||
resume?: boolean;
|
||||
/**
|
||||
* Caller-supplied RemediationStep entries threaded into the planner.
|
||||
* Mirrors RemediationPlanOpts.extraRemediations so onboard's --apply
|
||||
* --auto path (and MCP run_onboard auto modes) forward the same
|
||||
* onboard-check remediations the --check path already passes through
|
||||
* computeRemediationPlan. Without this the runner saw only generic
|
||||
* brain_score remediations and reported "Nothing to do" whenever the
|
||||
* only applicable work was an extra (e.g. extract-ner).
|
||||
*/
|
||||
extraRemediations?: RemediationStep[];
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -1,30 +0,0 @@
|
||||
/**
|
||||
* Unit tests for parsePruneStatuses (PR #2282) — `jobs prune --status` parsing.
|
||||
*/
|
||||
|
||||
import { describe, test, expect } from 'bun:test';
|
||||
import { parsePruneStatuses, PRUNE_STATUSES } from '../src/commands/jobs.ts';
|
||||
|
||||
describe('parsePruneStatuses', () => {
|
||||
test('parses a single status', () => {
|
||||
expect(parsePruneStatuses('failed')).toEqual(['failed']);
|
||||
});
|
||||
|
||||
test('parses a comma-separated list with whitespace', () => {
|
||||
expect(parsePruneStatuses(' completed, dead ')).toEqual(['completed', 'dead']);
|
||||
});
|
||||
|
||||
test('accepts every documented terminal status', () => {
|
||||
expect(parsePruneStatuses(PRUNE_STATUSES.join(','))).toEqual([...PRUNE_STATUSES]);
|
||||
});
|
||||
|
||||
test('throws on non-terminal statuses', () => {
|
||||
expect(() => parsePruneStatuses('waiting')).toThrow(/Invalid: waiting/);
|
||||
expect(() => parsePruneStatuses('completed,active')).toThrow(/Invalid: active/);
|
||||
});
|
||||
|
||||
test('throws on empty value', () => {
|
||||
expect(() => parsePruneStatuses('')).toThrow(/comma-separated subset/);
|
||||
expect(() => parsePruneStatuses(',')).toThrow(/comma-separated subset/);
|
||||
});
|
||||
});
|
||||
@@ -702,32 +702,6 @@ describe('MinionQueue: Prune', () => {
|
||||
const count = await queue.prune({ olderThan: new Date(Date.now() + 86400000) }); // future date = prune everything old enough
|
||||
expect(count).toBe(1); // only the cancelled one
|
||||
});
|
||||
|
||||
// PR #2282: `jobs prune --status` passes an explicit status subset through.
|
||||
test('status filter prunes only the requested terminal statuses', async () => {
|
||||
const cancelled = await queue.add('sync', {});
|
||||
await queue.cancelJob(cancelled.id);
|
||||
const dead = await queue.add('embed', {}, { max_attempts: 1 });
|
||||
await queue.claim('tok1', 30000, 'default', ['embed']);
|
||||
await queue.failJob(dead.id, 'tok1', 'boom', 'dead');
|
||||
|
||||
const count = await queue.prune({ olderThan: new Date(Date.now() + 86400000), status: ['dead'] });
|
||||
expect(count).toBe(1); // only the dead one
|
||||
|
||||
const remaining = await queue.getJobs({ status: 'cancelled' });
|
||||
expect(remaining.length).toBe(1);
|
||||
});
|
||||
|
||||
// PR #2282: `--older-than 0d` = no age floor — olderThan of "now" deletes
|
||||
// terminal jobs that finished moments ago.
|
||||
test('olderThan now (0d semantics) prunes just-terminated jobs', async () => {
|
||||
const job = await queue.add('sync', {});
|
||||
await queue.cancelJob(job.id);
|
||||
await new Promise(r => setTimeout(r, 5)); // ensure updated_at < now
|
||||
|
||||
const count = await queue.prune({ olderThan: new Date() });
|
||||
expect(count).toBe(1);
|
||||
});
|
||||
});
|
||||
|
||||
// --- Stats (1 test) ---
|
||||
|
||||
@@ -0,0 +1,108 @@
|
||||
// test/remediation-run-extras.serial.test.ts
|
||||
// Regression for PR #2161 takeover: `gbrain onboard --apply --auto` dropped
|
||||
// onboard-check extraRemediations. Two distinct halves of the bug:
|
||||
// 1. runRemediation built the pre-flight plan + initial recs WITHOUT the
|
||||
// extras, so an extras-only plan reported "Nothing to do".
|
||||
// 2. The D7 mid-run recheck rebuilt recs WITHOUT the extras after every
|
||||
// completed step, so with 2+ plannable steps all remaining extras were
|
||||
// dropped after step 1. The recheck must also filter out extras this
|
||||
// run already processed — extras carry static status:'remediable', so
|
||||
// unfiltered threading would resubmit completed extras forever.
|
||||
//
|
||||
// SERIAL: mock.module (queue + wait-for-completion stubs, R2) + GBRAIN_HOME
|
||||
// env mutation so checkpoint files land in a tmpdir, not ~/.gbrain.
|
||||
|
||||
import { describe, expect, test, beforeAll, afterAll, mock } from 'bun:test';
|
||||
import { mkdtempSync, rmSync } from 'node:fs';
|
||||
import { tmpdir } from 'node:os';
|
||||
import { join } from 'node:path';
|
||||
import { PGLiteEngine } from '../src/core/pglite-engine.ts';
|
||||
import { makeRemediationStep } from '../src/core/remediation-step.ts';
|
||||
|
||||
// Stub the Minion queue: every submitted job is immediately 'completed'.
|
||||
// runRemediation only calls queue.add + waitForCompletion(queue, id).
|
||||
let nextJobId = 1;
|
||||
const submittedJobs: Array<{ name: string }> = [];
|
||||
mock.module('../src/core/minions/queue.ts', () => ({
|
||||
MinionQueue: class {
|
||||
async add(name: string) {
|
||||
submittedJobs.push({ name });
|
||||
return { id: nextJobId++, status: 'completed' };
|
||||
}
|
||||
},
|
||||
}));
|
||||
mock.module('../src/core/minions/wait-for-completion.ts', () => ({
|
||||
waitForCompletion: async () => ({ status: 'completed' }),
|
||||
}));
|
||||
|
||||
let engine: PGLiteEngine;
|
||||
let home: string;
|
||||
const prevHome = process.env.GBRAIN_HOME;
|
||||
|
||||
beforeAll(async () => {
|
||||
home = mkdtempSync(join(tmpdir(), 'gbrain-remextras-'));
|
||||
process.env.GBRAIN_HOME = home;
|
||||
engine = new PGLiteEngine();
|
||||
await engine.connect({});
|
||||
await engine.initSchema();
|
||||
}, 120_000);
|
||||
|
||||
afterAll(async () => {
|
||||
await engine.disconnect();
|
||||
if (prevHome === undefined) delete process.env.GBRAIN_HOME;
|
||||
else process.env.GBRAIN_HOME = prevHome;
|
||||
rmSync(home, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
function extra(id: string, job: string) {
|
||||
return makeRemediationStep({
|
||||
id,
|
||||
job,
|
||||
params: {},
|
||||
severity: 'medium',
|
||||
est_seconds: 5,
|
||||
est_usd_cost: 0,
|
||||
rationale: 'synthetic onboard-check extra',
|
||||
status: 'remediable',
|
||||
});
|
||||
}
|
||||
|
||||
describe('runRemediation extraRemediations threading', () => {
|
||||
test('extras-only plan runs BOTH extras and terminates (no Nothing-to-do, no resubmit loop)', async () => {
|
||||
// Empty PGLite brain → zero health-derived recommendations. Without the
|
||||
// fix, half 1 makes this run return submitted: [] via onNothingToDo.
|
||||
// With only half 1 (the original PR #2161 diff), the mid-run recheck
|
||||
// drops the second extra after step 1 — submitted has 1 entry, not 2.
|
||||
const { runRemediation } = await import('../src/core/remediation/run.ts');
|
||||
let nothingToDo = false;
|
||||
const result = await runRemediation(
|
||||
engine,
|
||||
{
|
||||
targetScore: 1,
|
||||
extraRemediations: [
|
||||
extra('onboard.extract_ner', 'extract-ner'),
|
||||
extra('onboard.extract_timeline', 'extract-timeline-from-meetings'),
|
||||
],
|
||||
// Safety bound: an unfiltered recheck would resubmit completed
|
||||
// extras forever; maxJobs turns that regression into a fast fail
|
||||
// (extra count > 1 below) instead of a hung test.
|
||||
maxJobs: 5,
|
||||
},
|
||||
{ onNothingToDo: () => { nothingToDo = true; } },
|
||||
);
|
||||
|
||||
expect(nothingToDo).toBe(false);
|
||||
const ids = result.submitted.map((s) => s.id);
|
||||
expect(ids).toContain('onboard.extract_ner');
|
||||
expect(ids).toContain('onboard.extract_timeline');
|
||||
// Each extra ran exactly once — the recheck must not re-plan extras the
|
||||
// run already processed.
|
||||
expect(ids.filter((i) => i === 'onboard.extract_ner').length).toBe(1);
|
||||
expect(ids.filter((i) => i === 'onboard.extract_timeline').length).toBe(1);
|
||||
expect(result.submitted.every((s) => s.status === 'completed')).toBe(true);
|
||||
expect(submittedJobs.map((j) => j.name).sort()).toEqual([
|
||||
'extract-ner',
|
||||
'extract-timeline-from-meetings',
|
||||
]);
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user