diff --git a/src/commands/lint.ts b/src/commands/lint.ts index 54b81c290..94a784b0e 100644 --- a/src/commands/lint.ts +++ b/src/commands/lint.ts @@ -383,15 +383,30 @@ async function resolveLintContentSanity( }; } +/** + * Directories never containing knowledge pages, skipped by default. + * Deliberately tiny: only vendored dependency trees qualify. Anything + * more opinionated (README.md, CHANGELOG.md, test/) is repo policy — + * callers opt in via `--exclude` / `LintOpts.exclude`. Dot- and + * underscore-prefixed entries are already skipped by the walk. + */ +const DEFAULT_LINT_EXCLUDE_DIRS = new Set(['node_modules']); + /** Collect markdown files from a directory */ -function collectPages(dir: string): string[] { +function collectPages(dir: string, extraExcludes: string[] = []): string[] { + const extra = new Set(extraExcludes); const pages: string[] = []; function walk(d: string) { for (const entry of readdirSync(d)) { if (entry.startsWith('.') || entry.startsWith('_')) continue; const full = join(d, entry); - if (lstatSync(full).isDirectory()) walk(full); - else if (entry.endsWith('.md')) pages.push(full); + if (lstatSync(full).isDirectory()) { + if (DEFAULT_LINT_EXCLUDE_DIRS.has(entry) || extra.has(entry)) continue; + walk(full); + } else if (entry.endsWith('.md')) { + if (extra.has(entry)) continue; + pages.push(full); + } } } walk(dir); @@ -419,6 +434,13 @@ export interface LintOpts { * yields + checks this every 200 pages. */ signal?: AbortSignal; + /** + * #2649: extra dir/file basenames to skip while collecting pages, in + * addition to node_modules and dot/underscore entries. For mixed-content + * repos (knowledge pages alongside software trees). Ignored for + * single-file targets. + */ + exclude?: string[]; } export interface LintResult { @@ -445,7 +467,7 @@ export async function runLintCore(opts: LintOpts): Promise { } const isSingleFile = statSync(opts.target).isFile(); - const pages = isSingleFile ? [opts.target] : collectPages(opts.target); + const pages = isSingleFile ? [opts.target] : collectPages(opts.target, opts.exclude ?? []); // Resolve content-sanity config once for this lint run (D1: lift DB // config when reachable). Caller can pre-pass via opts.contentSanity @@ -496,14 +518,27 @@ export async function runLintCore(opts: LintOpts): Promise { } export async function runLint(args: string[]) { - const target = args.find(a => !a.startsWith('--')); + // #2649: --exclude=a,b or --exclude a,b — extra basenames to skip. + const extraExcludes: string[] = []; + const skipIdx = new Set(); + for (let i = 0; i < args.length; i++) { + const a = args[i]; + if (a.startsWith('--exclude=')) { + extraExcludes.push(...a.slice('--exclude='.length).split(',').map(s => s.trim()).filter(Boolean)); + } else if (a === '--exclude' && i + 1 < args.length) { + extraExcludes.push(...args[i + 1].split(',').map(s => s.trim()).filter(Boolean)); + skipIdx.add(i + 1); + } + } + const target = args.find((a, i) => !a.startsWith('--') && !skipIdx.has(i)); const doFix = args.includes('--fix'); const dryRun = args.includes('--dry-run'); if (!target) { - console.error('Usage: gbrain lint [--fix] [--dry-run]'); + console.error('Usage: gbrain lint [--fix] [--dry-run] [--exclude a,b]'); console.error(' --fix Auto-fix fixable issues (LLM preambles, code fences)'); console.error(' --dry-run Preview fixes without writing'); + console.error(' --exclude Comma-separated dir/file basenames to skip (in addition to node_modules)'); process.exit(1); } @@ -515,7 +550,7 @@ export async function runLint(args: string[]) { // Single file or directory — print human detail as we go, then rely on // Core for the aggregate numbers at the end. const isSingleFile = statSync(target).isFile(); - const pages = isSingleFile ? [target] : collectPages(target); + const pages = isSingleFile ? [target] : collectPages(target, extraExcludes); // Progress on stderr. Stdout keeps the per-issue human output it always had. const { createProgress } = await import('../core/progress.ts'); @@ -562,7 +597,7 @@ export async function runLint(args: string[]) { // produces canonical numbers for the summary line). // Pass contentSanity through so runLintCore skips its own resolve // (we already resolved once for the human-detail loop above). - const result = await runLintCore({ target, fix: doFix, dryRun, contentSanity }); + const result = await runLintCore({ target, fix: doFix, dryRun, contentSanity, exclude: extraExcludes }); console.log(`\n${result.pages_scanned} pages scanned. ${result.total_issues} issue(s) in ${result.pages_with_issues} page(s).`); if (doFix) { console.log(`${dryRun ? '(dry run) ' : ''}${result.total_fixed} auto-fixed.`); diff --git a/src/core/cycle/synthesize.ts b/src/core/cycle/synthesize.ts index 813b06a45..dea310c63 100644 --- a/src/core/cycle/synthesize.ts +++ b/src/core/cycle/synthesize.ts @@ -28,6 +28,7 @@ import type Anthropic from '@anthropic-ai/sdk'; import { readFileSync, existsSync, writeFileSync, mkdirSync } from 'node:fs'; +import { randomUUID } from 'node:crypto'; import { chat as gatewayChat, validateModelId, type ChatResult } from '../ai/gateway.ts'; import { AIConfigError } from '../ai/errors.ts'; import { normalizeModelId } from '../model-id.ts'; @@ -37,7 +38,8 @@ import type { BrainEngine } from '../engine.ts'; import type { PhaseResult, PhaseError } from '../cycle.ts'; import { MinionQueue } from '../minions/queue.ts'; import { waitForCompletion, TimeoutError } from '../minions/wait-for-completion.ts'; -import type { MinionJobInput, SubagentHandlerData } from '../minions/types.ts'; +import { makeSubagentHandler } from '../minions/handlers/subagent.ts'; +import type { MinionJobInput, MinionJobContext, MinionHandler, SubagentHandlerData } from '../minions/types.ts'; import { discoverTranscripts, type DiscoveredTranscript } from './transcript-discovery.ts'; import { serializeMarkdown, serializePageToMarkdown } from '../markdown.ts'; import type { Page, PageType } from '../types.ts'; @@ -262,6 +264,121 @@ export interface SynthesizePhaseOpts { once?: boolean; } +const INLINE_PGLITE_LOCK_MS = 30_000; + +/** + * PGLite cannot be served by a separate Minions worker process: the embedded + * data-dir holds an exclusive file lock, so subagent children enqueued by the + * synth parent would sit in 'waiting' until waitForCompletion times out. + * Drive the same claim → run → complete/fail loop a worker would perform, + * inline, against this phase's private child queue. + * + * `yieldDuringPhase` is ticked on a 60s interval while a child runs so the + * 5-min cycle lock TTL keeps refreshing during long (up to 30-min) children. + */ +async function runPgliteSubagentsInline( + engine: BrainEngine, + queue: MinionQueue, + queueName: string, + yieldDuringPhase?: () => Promise, + handler: MinionHandler = makeSubagentHandler({ engine }), +): Promise { + if (engine.kind !== 'pglite') return; + + while (true) { + // Housekeeping a worker would normally perform, so child rows can reach + // terminal states (delayed retries promoted, timeouts dead-lettered) + // before the synth parent enters waitForCompletion polling. + await queue.promoteDelayed(); + await queue.handleStalled(); + await queue.handleTimeouts(); + await queue.handleWallClockTimeouts(INLINE_PGLITE_LOCK_MS); + + const lockToken = randomUUID(); + const job = await queue.claim(lockToken, INLINE_PGLITE_LOCK_MS, queueName, ['subagent']); + if (!job) return; + + const abort = new AbortController(); + const shutdown = new AbortController(); + const context: MinionJobContext = { + id: job.id, + name: job.name, + data: job.data, + attempts_made: job.attempts_made, + signal: abort.signal, + deadlineAtMs: job.timeout_at != null ? job.timeout_at.getTime() : null, + shutdownSignal: shutdown.signal, + updateProgress: async (progress: unknown) => { + await queue.updateProgress(job.id, lockToken, progress); + }, + updateTokens: async (tokens) => { + await queue.updateTokens(job.id, lockToken, tokens); + }, + log: async (message) => { + const value = typeof message === 'string' ? message : JSON.stringify(message); + await engine.executeRaw( + `UPDATE minion_jobs SET stacktrace = COALESCE(stacktrace, '[]'::jsonb) || to_jsonb($1::text), + updated_at = now() + WHERE id = $2 AND status = 'active' AND lock_token = $3`, + [value, job.id, lockToken], + ); + }, + isActive: async () => { + const rows = await engine.executeRaw<{ id: number }>( + `SELECT id FROM minion_jobs WHERE id = $1 AND status = 'active' AND lock_token = $2`, + [job.id, lockToken], + ); + return rows.length > 0; + }, + readInbox: async () => queue.readInbox(job.id, lockToken), + }; + + // Per-job deadline enforcement (worker.ts parity). While the drain loop + // awaits the handler, the handleTimeouts sweep above can't run, so nothing + // else can stop a child that blows past timeout_ms — the handler only + // stops when ctx.signal fires. Derive the delay from the claim-time + // timeout_at stamp so timer, DB sweeper, and deadlineAtMs agree. + let timeoutTimer: ReturnType | null = null; + if (job.timeout_ms != null) { + const delayMs = job.timeout_at != null + ? Math.max(0, job.timeout_at.getTime() - Date.now()) + : job.timeout_ms; + timeoutTimer = setTimeout(() => { + if (!abort.signal.aborted) abort.abort(new Error('timeout')); + }, delayMs); + } + + // Cycle-lock keepalive while the child runs (best-effort, never throws). + const keepalive = yieldDuringPhase + ? setInterval(() => { yieldDuringPhase().catch(() => { /* best-effort */ }); }, 60_000) + : null; + try { + const result = await handler(context); + await queue.completeJob( + job.id, + lockToken, + result != null ? (typeof result === 'object' ? result as Record : { value: result }) : undefined, + ); + } catch (e) { + // Timeout is terminal (handleTimeouts parity: stall → retry, + // timeout → dead), never a delayed retry. + const timedOut = abort.signal.aborted; + const errorText = timedOut ? 'timeout exceeded' : (e instanceof Error ? e.message : String(e)); + const attemptsExhausted = job.attempts_made + 1 >= job.max_attempts; + await queue.failJob( + job.id, + lockToken, + errorText, + timedOut || attemptsExhausted ? 'dead' : 'delayed', + 0, + ); + } finally { + if (timeoutTimer) clearTimeout(timeoutTimer); + if (keepalive) clearInterval(keepalive); + } + } +} + export async function runPhaseSynthesize( engine: BrainEngine, opts: SynthesizePhaseOpts, @@ -428,6 +545,12 @@ export async function runPhaseSynthesize( } const queue = new MinionQueue(engine); + // PGLite children drain inline (no separate worker can open the embedded + // data-dir), so give them a private per-run queue: the inline drain must + // never claim unrelated 'default'-queue jobs a Postgres worker owns. + const childQueueName = engine.kind === 'pglite' + ? `dream-inline-${Date.now()}-${randomUUID().slice(0, 8)}` + : 'default'; const childIds: number[] = []; /** Map child job_id → chunk metadata for D6 orchestrator-side slug rewrite. */ const chunkInfo = new Map(); @@ -506,6 +629,7 @@ export async function runPhaseSynthesize( on_child_fail: 'continue', idempotency_key, timeout_ms: config.subagentTimeoutMs, + queue: childQueueName, }; const child = await queue.add( 'subagent', @@ -520,6 +644,12 @@ export async function runPhaseSynthesize( } } + // PGLite cannot run a separate Minions worker because the embedded DB + // holds an exclusive file lock. Drain this phase's private child queue + // inline so the parent observes terminal child states instead of polling + // waiters until subagentWaitTimeoutMs expires. No-op on Postgres. + await runPgliteSubagentsInline(engine, queue, childQueueName, opts.yieldDuringPhase); + // Wait for every child to reach a terminal state. Tick yieldDuringPhase // every 5 min so the cycle lock TTL refreshes. const childOutcomes: Array<{ jobId: number; status: string }> = []; @@ -1382,4 +1512,5 @@ export const __testing = { buildSynthesisPrompt, stampDreamProvenance, reverseWriteRefs, + runPgliteSubagentsInline, }; diff --git a/test/e2e/dream-synthesize-pglite.test.ts b/test/e2e/dream-synthesize-pglite.test.ts index 7cb2082bf..a859f5af2 100644 --- a/test/e2e/dream-synthesize-pglite.test.ts +++ b/test/e2e/dream-synthesize-pglite.test.ts @@ -17,7 +17,7 @@ import { mkdtempSync, rmSync, writeFileSync } from 'fs'; import { tmpdir } from 'os'; import { join } from 'path'; import { PGLiteEngine } from '../../src/core/pglite-engine.ts'; -import { runPhaseSynthesize, renderPageToMarkdown } from '../../src/core/cycle/synthesize.ts'; +import { runPhaseSynthesize, renderPageToMarkdown, __testing as synthTesting } from '../../src/core/cycle/synthesize.ts'; interface TestRig { engine: PGLiteEngine; @@ -516,3 +516,115 @@ describe('E2E synthesize — verdict cache (Q-2)', () => { } }, 30_000); }); + +describe('E2E synthesize — PGLite inline subagent drain (takeover of #2699)', () => { + test('drains private subagent queue inline so the parent can observe completion', async () => { + const rig = await setupRig(); + try { + const { MinionQueue } = await import('../../src/core/minions/queue.ts'); + const queue = new MinionQueue(rig.engine); + const queueName = `dream-inline-test-${Date.now()}`; + const child = await queue.add( + 'subagent', + { prompt: 'test', model: 'anthropic:claude-sonnet-4-6', max_turns: 1 }, + { queue: queueName, max_attempts: 1 }, + { allowProtectedSubmit: true }, + ); + + let ticks = 0; + await synthTesting.runPgliteSubagentsInline( + rig.engine, + queue, + queueName, + async () => { ticks++; }, + async (ctx) => { + await ctx.log('inline child ran'); + await ctx.updateProgress({ step: 'done' }); + return { ok: true }; + }, + ); + expect(ticks).toBe(0); // 60s keepalive never fires for a fast child + + const final = await queue.getJob(child.id); + expect(final?.status).toBe('completed'); + expect(final?.result).toEqual({ ok: true }); + expect(final?.progress).toEqual({ step: 'done' }); + + const waiting = await rig.engine.executeRaw<{ count: string }>( + `SELECT COUNT(*)::text AS count FROM minion_jobs WHERE queue = $1 AND status = 'waiting'`, + [queueName], + ); + expect(waiting[0]?.count).toBe('0'); + } finally { + await rig.cleanup(); + } + }, 30_000); + + test('terminally marks failed inline children so synth parent will not hang', async () => { + const rig = await setupRig(); + try { + const { MinionQueue } = await import('../../src/core/minions/queue.ts'); + const queue = new MinionQueue(rig.engine); + const queueName = `dream-inline-test-fail-${Date.now()}`; + const child = await queue.add( + 'subagent', + { prompt: 'test', model: 'anthropic:claude-sonnet-4-6', max_turns: 1 }, + { queue: queueName, max_attempts: 1 }, + { allowProtectedSubmit: true }, + ); + + await synthTesting.runPgliteSubagentsInline( + rig.engine, + queue, + queueName, + undefined, + async () => { + throw new Error('synthetic child failure'); + }, + ); + + const final = await queue.getJob(child.id); + expect(final?.status).toBe('dead'); + expect(final?.error_text).toContain('synthetic child failure'); + } finally { + await rig.cleanup(); + } + }, 30_000); + + test('enforces per-job timeout_ms inline: aborts the child and dead-letters it', async () => { + const rig = await setupRig(); + try { + const { MinionQueue } = await import('../../src/core/minions/queue.ts'); + const queue = new MinionQueue(rig.engine); + const queueName = `dream-inline-test-timeout-${Date.now()}`; + const child = await queue.add( + 'subagent', + { prompt: 'test', model: 'anthropic:claude-sonnet-4-6', max_turns: 1 }, + { queue: queueName, max_attempts: 3, timeout_ms: 100 }, + { allowProtectedSubmit: true }, + ); + + // Handler only ends when ctx.signal fires — like the real subagent + // handler mid-LLM-call. Without the inline timeout timer this awaits + // forever and the drain (and the whole cycle) wedges. + await synthTesting.runPgliteSubagentsInline( + rig.engine, + queue, + queueName, + undefined, + async (ctx) => { + await new Promise((_, reject) => { + ctx.signal.addEventListener('abort', () => reject(new Error('aborted')), { once: true }); + }); + }, + ); + + // Timeout is terminal (dead), never a delayed retry, despite max_attempts: 3. + const final = await queue.getJob(child.id); + expect(final?.status).toBe('dead'); + expect(final?.error_text).toBe('timeout exceeded'); + } finally { + await rig.cleanup(); + } + }, 30_000); +}); diff --git a/test/lint.test.ts b/test/lint.test.ts index f843d5859..e6a8cb560 100644 --- a/test/lint.test.ts +++ b/test/lint.test.ts @@ -139,3 +139,48 @@ describe('fixContent', () => { expect(fixed).toContain('# Title'); }); }); + +describe('runLintCore exclude (takeover of #2649)', () => { + const { mkdtempSync, rmSync, mkdirSync, writeFileSync } = require('node:fs') as typeof import('node:fs'); + const { tmpdir } = require('node:os') as typeof import('node:os'); + const { join } = require('node:path') as typeof import('node:path'); + const { runLintCore } = require('../src/commands/lint.ts') as typeof import('../src/commands/lint.ts'); + + const PAGE = '---\ntitle: T\ntype: note\ncreated: 2026-04-11\n---\n\n# T\n\nBody.\n'; + + function makeRepo(): string { + const dir = mkdtempSync(join(tmpdir(), 'gbrain-lint-excl-')); + writeFileSync(join(dir, 'page.md'), PAGE); + writeFileSync(join(dir, 'README.md'), PAGE); + mkdirSync(join(dir, 'node_modules', 'dep'), { recursive: true }); + writeFileSync(join(dir, 'node_modules', 'dep', 'vendor.md'), PAGE); + mkdirSync(join(dir, 'software')); + writeFileSync(join(dir, 'software', 'notes.md'), PAGE); + return dir; + } + + test('node_modules is excluded by default; nothing else is', async () => { + const dir = makeRepo(); + try { + const result = await runLintCore({ target: dir, contentSanity: { disabled: true } }); + // page.md + README.md + software/notes.md — vendor.md skipped. + expect(result.pages_scanned).toBe(3); + } finally { + rmSync(dir, { recursive: true, force: true }); + } + }); + + test('--exclude basenames skip dirs and files', async () => { + const dir = makeRepo(); + try { + const result = await runLintCore({ + target: dir, + contentSanity: { disabled: true }, + exclude: ['software', 'README.md'], + }); + expect(result.pages_scanned).toBe(1); // only page.md + } finally { + rmSync(dir, { recursive: true, force: true }); + } + }); +});