fix(cycle,lint): PGLite inline synth subagent drain + lint --exclude (takeover of #2699, #2649) (#3162)

* fix(cycle): drain PGLite synth subagents inline (takeover of #2699) PGLite holds an exclusive file lock on its embedded data-dir, so no separate Minions worker can serve the subagent children the synthesize phase enqueues — they sat in 'waiting' until waitForCompletion timed out. Drain a private per-run child queue inline (claim → run → complete/fail, plus the promote/stall/timeout housekeeping a worker would perform). No-op on Postgres, where children stay on the shared 'default' queue. Rebased onto the reworked synthesize (config.subagentTimeoutMs, #1586 source scoping): the inline job context now carries deadlineAtMs from the claim-time timeout_at stamp, and opts.yieldDuringPhase is ticked on a 60s keepalive while each child runs so the 5-min cycle lock TTL refreshes across long (up to 30-min) children. Co-authored-by: TheRealMrSystem <TheRealMrSystem@users.noreply.github.com> Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * feat(lint): --exclude flag for mixed-content repos (takeover of #2649) Adds --exclude=a,b (and LintOpts.exclude) so mixed-content repos can skip software trees and repo metafiles by basename when collecting pages. The only built-in default is node_modules — vendored dependency trees are never knowledge pages; dot/underscore entries were already skipped by the walk. Diverges from #2649 deliberately: the original hardcoded an opinionated default list (README.md, CHANGELOG.md, CLAUDE.md, test/ dirs at any depth, plus fork-specific filenames), which silently changed lint counts for every existing repo. Those are repo policy — pass --exclude. Co-authored-by: ryangu00 <ryangu00@users.noreply.github.com> Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(cycle): enforce per-job timeout_ms in the PGLite inline subagent drain The inline drain claimed children with deadlineAtMs derived from timeout_at but never armed the worker's timeout timer — and the handleTimeouts sweep only runs between jobs, so nothing could stop a child that blew past its 30-min timeout_ms. A hung LLM call wedged the drain loop (and the whole cycle) indefinitely, with the 60s keepalive refreshing the cycle lock forever. Worker.ts parity: arm a timer from the claim-time timeout_at stamp, abort ctx.signal on fire, and dead-letter (never delayed-retry) timed-out children, mirroring handleTimeouts' stall→retry / timeout→dead split. Regression test: a child with timeout_ms=100 whose handler only ends on ctx.signal abort is dead-lettered with 'timeout exceeded'; pre-fix the test hangs to its 30s timeout. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: Garry Tan <garrytan@gmail.com> Co-authored-by: TheRealMrSystem <TheRealMrSystem@users.noreply.github.com> Co-authored-by: Claude Fable 5 <noreply@anthropic.com> Co-authored-by: ryangu00 <ryangu00@users.noreply.github.com>
2026-07-27 22:15:33 +00:00 · 2026-07-23 13:09:42 -07:00
parent 080b64e052
commit 04e6b3af14
4 changed files with 333 additions and 10 deletions
@@ -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<LintResult> {
  }

  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<LintResult> {
 }

 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<number>();
+  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 <dir|file.md> [--fix] [--dry-run]');
+    console.error('Usage: gbrain lint <dir|file.md> [--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.`);
@@ -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<void>,
+  handler: MinionHandler = makeSubagentHandler({ engine }),
+): Promise<void> {
+  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<typeof setTimeout> | 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<string, unknown> : { 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<number, { idx: number; hash6: string }>();
@@ -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,
 };
@@ -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);
+});
@@ -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 });
+    }
+  });
+});