From 14062504d94b3b224a469834945ba533109ee552 Mon Sep 17 00:00:00 2001 From: Garry Tan Date: Sat, 8 Aug 2026 23:59:51 -0700 Subject: [PATCH] =?UTF-8?q?fix(review):=20final-wave=20red-team=20batch=20?= =?UTF-8?q?=E2=80=94=20queue-aware=20worker=20probe,=20no=20cwd=20fallback?= =?UTF-8?q?s=20post-v127,=20config-set=20anchor=20normalization,=20honest?= =?UTF-8?q?=20abort=20detection?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The critical: the facts-backstop live-worker probe was queue-blind. The durable facts-absorb job goes to queue 'default', but readWorkers().length counted a worker on ANY queue (e.g. `gbrain jobs work --queue shell`) as proof the job would be drained — parking it forever AND skipping the in-process fallback, silently reintroducing the #2108 loss. The probe now requires a worker on the job's own queue (FACTS_ABSORB_QUEUE constant so submit + probe can't drift) and memoizes for 5s so readWorkers()'s synchronous per-entry `ps` exec stops running inside every remote put_page. Also in this batch: - No cwd fallbacks post-migration-v127: the extract + backlinks job handlers fell back to dir='.' (worker-daemon cwd, typically / under launchd) on a NULL anchor; doctor's image_assets fell back to process.cwd(); autopilot-cycle passed job.data.repoPath / the config anchor unvalidated. All four now fail loudly (doctor: WARN check) via requireAbsoluteStoredPath, mirroring the sync handler. - `gbrain config set sync.repo_path .` re-seeded a relative anchor around ingress normalization — now resolved absolute at set time (empty refused); isAnchorOwnedSyncPath no longer realpath-resolves a relative stored anchor against cwd (returns no-match). - requireAbsoluteStoredPath's remediation hint is parameterized per storage context (sources.local_path → per-source sync repoint; job payloads → cancel + resubmit; default unchanged). - Honest abort detection in the facts queue worker: /abort/i message sniffing matched Postgres's routine 'current transaction is aborted' — now err.name === 'AbortError' || signal.aborted only. - runImport's #2114 anchor decision (kept/seeded/repointed) now rides the JSON summary + return shape instead of stderr only. - TODOS: five follow-ups filed (orphaned-queue doctor check, wrapper-script scan, facts-absorb retention/PROTECTED posture, migration-version CI guard). Co-Authored-By: Claude Fable 5 --- TODOS.md | 31 ++++++++ src/commands/config.ts | 20 +++++- src/commands/doctor.ts | 88 +++++++++++++++-------- src/commands/import.ts | 42 ++++++++--- src/commands/jobs.ts | 54 +++++++++++--- src/commands/sync.ts | 18 ++++- src/core/facts/backstop.ts | 57 +++++++++++++-- src/core/repo-path.ts | 32 ++++++++- test/facts-backstop.test.ts | 140 +++++++++++++++++++++++++++++++++++- test/handlers.test.ts | 107 +++++++++++++++++++++++++++ test/repo-path.test.ts | 129 +++++++++++++++++++++++++++++++-- 11 files changed, 649 insertions(+), 69 deletions(-) diff --git a/TODOS.md b/TODOS.md index 34ff1362b..44702f2a7 100644 --- a/TODOS.md +++ b/TODOS.md @@ -1,5 +1,36 @@ # TODOS +## Sync/facts follow-ups (final wave, 2026-08) + +- [ ] **P2 — doctor check: waiting jobs on queues with zero live workers (red-team #1 + residual).** The facts-absorb probe is queue-aware now, but the general class remains: + any durable job submitted to a queue no live `gbrain jobs work` worker drains parks + silently. Add a doctor check that joins `minion_jobs` waiting rows against the worker + registry (`src/core/minions/worker-registry.ts:readWorkers`) by queue and WARNs with + the exact `gbrain jobs work --queue ` command per orphaned queue. +- [ ] **P2 — doctor scan for pre-fix launchd/cron wrapper scripts carrying a relative + `--repo`.** Migration v127 clears relative anchors in the DB but can't reach generated + wrapper scripts on disk (autopilot.ts:1373 generates absolute paths now; scripts baked + by older binaries may still carry `--repo .`). Scan the known install locations and + name the regenerate command (`gbrain autopilot --install`). +- [ ] **P3 — minion-jobs retention trade-off for facts-absorb rows.** Completed + facts-absorb rows double as idempotency memory: the content-hash `idempotency_key` + dedups re-submits of the same page content. Any future auto-prune of completed + `minion_jobs` rows re-opens duplicate extraction (cost, not correctness — the pipeline + dedups again at insert). Document the trade-off in the retention design before any + auto-prune lands. +- [ ] **P3 — CI guard: a new migration's version must exceed master's max.** The + v126/v127 land-order class: two in-flight branches claiming adjacent version numbers + rely on code comments today (a collision silently skips the second migration). A CI + check comparing the branch's `MIGRATIONS` max version against master's makes the + collision structural instead of social. +- [ ] **P3 — facts-absorb PROTECTED_JOB_NAMES posture note.** If `facts-absorb` is ever + added to PROTECTED_JOB_NAMES, the remote/MCP put_page backstop breaks: `minions.add` + from an untrusted context would be refused, and the durable path silently vanishes for + exactly the callers #2108 protects (falling back to the doomed in-process absorb). + Leave it unprotected deliberately (bounded LLM cost, no shell), or build a + trusted-submitter seam first — decide deliberately, don't drive-by. + ## WAL-repair wave follow-ups (#223/#1670/#2575) - [ ] **P2 — gate auto-repair on an unclean-shutdown marker (adversarial F7).** The classifier diff --git a/src/commands/config.ts b/src/commands/config.ts index 59be20056..5ee52d8cd 100644 --- a/src/commands/config.ts +++ b/src/commands/config.ts @@ -102,7 +102,7 @@ export async function runConfig(engine: BrainEngine, args: string[]) { } const key = args[1]; - const value = args[2]; + let value = args[2]; if (action === 'get' && key) { // #2120: `get` used to read only the DB plane, so a runtime-effective key @@ -204,6 +204,24 @@ export async function runConfig(engine: BrainEngine, args: string[]) { const coverageOverride = args.includes('--coverage-override') || args.includes('--yes'); + // repo-path.ts invariant (red-team): `gbrain config set sync.repo_path .` + // re-seeded a relative anchor, bypassing the ingress normalization every + // other writer applies (sync --repo, import, sources add, autopilot) — + // migration v127 would just clear it again, and until then every anchor + // reader hits the loud relative-path refusal. Resolve to absolute at set + // time; the confirmation line below prints the RESOLVED value so the user + // sees exactly what was persisted. + if (key === 'sync.repo_path') { + if (value.trim() === '') { + console.error( + `[config] sync.repo_path cannot be empty. To clear the anchor: gbrain config unset sync.repo_path`, + ); + process.exit(1); + } + const { resolveRepoArg } = await import('../core/repo-path.ts'); + value = resolveRepoArg(value); + } + // v0.42.42.0 (#2139): validate spend.posture at set time so a typo // ('tokenMax', 'max') doesn't silently fall back to gated. if (key === 'spend.posture') { diff --git a/src/commands/doctor.ts b/src/commands/doctor.ts index 4cfe2ad43..9f4b1c154 100644 --- a/src/commands/doctor.ts +++ b/src/commands/doctor.ts @@ -32,6 +32,7 @@ import { rankIssues, type RankedIssue } from '../core/doctor-cause-rank.ts'; import { getCliOptions, cliOptsToProgressOptions } from '../core/cli-options.ts'; import type { DbUrlSource } from '../core/config.ts'; import { gbrainPath, loadConfig } from '../core/config.ts'; +import { requireAbsoluteStoredPath } from '../core/repo-path.ts'; import { reflexEnabled } from '../core/context/reflex.ts'; import { resolveSocketPath } from '../core/context/resolve-ipc.ts'; import { resolveOwnerHolder } from '../core/owner-holder.ts'; @@ -7917,40 +7918,67 @@ export async function buildChecks( const { resolveAssetPath } = await import('./doctor-asset-paths.ts'); // storage_path is repo-relative for sync-ingested assets. Resolving // against cwd made this check a false-positive WARN whenever doctor - // ran outside the brain repo. - const repoRoot = (await engine.getConfig('sync.repo_path')) ?? process.cwd(); - for (const r of rows) { - // #1835: Windows drive paths (D:/…) translate to the WSL automount - // (/mnt/d/…) under WSL, and are SKIPPED (not "missing") on hosts - // where they cannot exist (macOS / plain Linux) — never joined onto - // repoRoot, which produced a false "restore from git" WARN. - const resolved = resolveAssetPath(r.storage_path, repoRoot); - if (resolved.abs === null) { - foreign++; - continue; - } + // ran outside the brain repo — and post-migration-v127 a cleared + // anchor must not silently become cwd either (repo-path.ts invariant). + // A missing/relative anchor is REPORTED (doctor's loud surface) and + // the presence scan skipped, never run against the wrong tree. + const anchorRaw = await engine.getConfig('sync.repo_path'); + let repoRoot: string | null = null; + let anchorProblem: string | null = null; + if (!anchorRaw) { + anchorProblem = 'no sync.repo_path anchor is configured'; + } else { try { - fs.statSync(resolved.abs); - } catch { - vanished++; - if (vanishedPaths.length < 5) vanishedPaths.push(r.storage_path); + repoRoot = requireAbsoluteStoredPath(anchorRaw, 'config sync.repo_path'); + } catch (err) { + anchorProblem = err instanceof Error ? err.message : String(err); } } - const checked = rows.length - foreign; - const foreignNote = foreign > 0 - ? ` (${foreign} Windows-drive path(s) skipped — not resolvable on this platform)` - : ''; - if (rows.length === 0) { - checks.push({ name: 'image_assets', status: 'ok', message: 'No image assets indexed yet' }); - } else if (vanished === 0) { - checks.push({ name: 'image_assets', status: 'ok', message: `${checked} image(s) all present on disk${foreignNote}` }); + if (repoRoot === null) { + if (rows.length === 0) { + checks.push({ name: 'image_assets', status: 'ok', message: 'No image assets indexed yet' }); + } else { + checks.push({ + name: 'image_assets', + status: 'warn', + message: `${rows.length} image asset(s) indexed but the on-disk presence scan was skipped: ${anchorProblem}. ` + + `Fix: gbrain sync --repo (or: gbrain config set sync.repo_path ).`, + }); + } } else { - checks.push({ - name: 'image_assets', - status: 'warn', - message: `${vanished} of ${checked} image(s) missing from disk (e.g. ${vanishedPaths.join(', ')})${foreignNote}. ` + - `Fix: restore from git, or \`gbrain sync --skip-failed\` to acknowledge.`, - }); + for (const r of rows) { + // #1835: Windows drive paths (D:/…) translate to the WSL automount + // (/mnt/d/…) under WSL, and are SKIPPED (not "missing") on hosts + // where they cannot exist (macOS / plain Linux) — never joined onto + // repoRoot, which produced a false "restore from git" WARN. + const resolved = resolveAssetPath(r.storage_path, repoRoot); + if (resolved.abs === null) { + foreign++; + continue; + } + try { + fs.statSync(resolved.abs); + } catch { + vanished++; + if (vanishedPaths.length < 5) vanishedPaths.push(r.storage_path); + } + } + const checked = rows.length - foreign; + const foreignNote = foreign > 0 + ? ` (${foreign} Windows-drive path(s) skipped — not resolvable on this platform)` + : ''; + if (rows.length === 0) { + checks.push({ name: 'image_assets', status: 'ok', message: 'No image assets indexed yet' }); + } else if (vanished === 0) { + checks.push({ name: 'image_assets', status: 'ok', message: `${checked} image(s) all present on disk${foreignNote}` }); + } else { + checks.push({ + name: 'image_assets', + status: 'warn', + message: `${vanished} of ${checked} image(s) missing from disk (e.g. ${vanishedPaths.join(', ')})${foreignNote}. ` + + `Fix: restore from git, or \`gbrain sync --skip-failed\` to acknowledge.`, + }); + } } } catch { // Pre-v36 brains may not have the files table on PGLite — quiet skip. diff --git a/src/commands/import.ts b/src/commands/import.ts index 99c095151..56a61fcbb 100644 --- a/src/commands/import.ts +++ b/src/commands/import.ts @@ -43,6 +43,15 @@ export interface RunImportResult { errors: number; chunksCreated: number; failures: Array<{ path: string; error: string }>; + /** + * #2114 anchor decision (red-team: was stderr-only, invisible to --json + * consumers). 'seeded' = sync.repo_path was unset and now points here; + * 'repointed' = --set-repo-path moved an existing anchor here; 'kept' = + * the existing anchor won (already here, or a different tree without the + * flag). Absent when no decision was made (non-git dir, managedBookmark, + * or import failures blocked the bookmark block). + */ + anchor?: 'kept' | 'seeded' | 'repointed'; } export async function runImport( @@ -490,13 +499,9 @@ export async function runImport( } const totalTime = ((Date.now() - startTime) / 1000).toFixed(1); - if (jsonOutput) { - console.log(JSON.stringify({ - status: 'success', duration_s: parseFloat(totalTime), - imported, skipped, errors, chunks: chunksCreated, - total_files: allFiles.length, - })); - } else { + // The --json summary is printed AFTER the anchor-decision block below so it + // can carry the `anchor` field (red-team: the decision was stderr-only). + if (!jsonOutput) { console.log(`\nImport complete (${totalTime}s):`); console.log(` ${imported} pages imported`); console.log(` ${skipped} pages skipped (${skipped - errors} unchanged, ${errors} errors)`); @@ -560,6 +565,7 @@ export async function runImport( // ledger + bookmark via the shared gate (applySyncFailureGate). Skipping the // internal handling here prevents double-recording (which would double-count // the auto-skip `attempts` streak) and a competing bookmark write. + let anchorDecision: 'kept' | 'seeded' | 'repointed' | undefined; if (gitHead && !opts.managedBookmark) { // Record failures into the central JSONL so doctor can surface them. // Use gitHead as the commit so a later sync can tell "same broken @@ -590,20 +596,38 @@ export async function runImport( // --set-repo-path flag; without it the existing anchor wins and we say // so on stderr. const existingAnchor = await engine.getConfig('sync.repo_path'); - if (!existingAnchor || existingAnchor === dir) { + if (!existingAnchor) { await engine.setConfig('sync.repo_path', dir); + anchorDecision = 'seeded'; + } else if (existingAnchor === dir) { + await engine.setConfig('sync.repo_path', dir); + anchorDecision = 'kept'; } else if (args.includes('--set-repo-path')) { console.error(`sync.repo_path repointed: ${existingAnchor} -> ${dir} (--set-repo-path)`); await engine.setConfig('sync.repo_path', dir); + anchorDecision = 'repointed'; } else { console.error( `sync.repo_path stays at ${existingAnchor} (this import ran in ${dir}). ` + `Pass --set-repo-path to repoint the sync anchor to this directory.`, ); + anchorDecision = 'kept'; } } - return { imported, skipped, errors, chunksCreated, failures }; + if (jsonOutput) { + console.log(JSON.stringify({ + status: 'success', duration_s: parseFloat(totalTime), + imported, skipped, errors, chunks: chunksCreated, + total_files: allFiles.length, + ...(anchorDecision ? { anchor: anchorDecision } : {}), + })); + } + + return { + imported, skipped, errors, chunksCreated, failures, + ...(anchorDecision ? { anchor: anchorDecision } : {}), + }; } /** diff --git a/src/commands/jobs.ts b/src/commands/jobs.ts index 60b0de0a6..0650c5fbd 100644 --- a/src/commands/jobs.ts +++ b/src/commands/jobs.ts @@ -10,7 +10,7 @@ import { WORKER_EXIT_RSS_WATCHDOG } from '../core/minions/worker-exit-codes.ts'; import type { MinionHandler, MinionJob, MinionJobStatus } from '../core/minions/types.ts'; import type { PaceKeyOverrides } from '../core/pace-mode.ts'; import { loadConfig, isThinClient } from '../core/config.ts'; -import { requireAbsoluteStoredPath } from '../core/repo-path.ts'; +import { requireAbsoluteStoredPath, JOB_PAYLOAD_REMEDIATION } from '../core/repo-path.ts'; import { callRemoteTool, unpackToolResult } from '../core/mcp-client.ts'; import { parseNiceValue, applyNiceness, getEffectiveNiceness, formatNice } from '../core/minions/niceness.ts'; @@ -60,6 +60,26 @@ const GATEWAY_REFRESH_JOB_NAMES = new Set([ 'embed-catch-up', ]); +/** + * repo-path.ts invariant, worker-daemon edition (red-team, post-migration + * v127): a job handler that needs a filesystem dir must NEVER fall back to + * '.' / cwd — the worker daemon's cwd (typically / under launchd) is exactly + * the wrong-tree footgun v127 clears anchors to avoid. NULL anchor → loud + * failure naming the fix; relative anchor → requireAbsoluteStoredPath's + * refusal. Used by the 'extract' and 'backlinks' handlers. + */ +function requireConfiguredRepoAnchor(anchor: string | null, jobName: string): string { + if (!anchor) { + throw new Error( + `${jobName} job has no data.dir and no sync.repo_path anchor is configured — ` + + `refusing to fall back to the worker's cwd (the wrong-tree footgun). ` + + `Resubmit the job with an absolute {"dir": "/abs/path"}, or seed the anchor: ` + + `gbrain sync --repo (or: gbrain config set sync.repo_path ).`, + ); + } + return requireAbsoluteStoredPath(anchor, 'config sync.repo_path'); +} + function registerBuiltinJob( worker: MinionWorker, engine: BrainEngine, @@ -1428,7 +1448,7 @@ export async function registerBuiltinHandlers( // enqueued by an older binary would otherwise resolve against the WORKER // daemon's cwd. Fail the job loudly instead. const repoPath = typeof job.data.repoPath === 'string' - ? requireAbsoluteStoredPath(job.data.repoPath, 'job.data.repoPath') + ? requireAbsoluteStoredPath(job.data.repoPath, 'job.data.repoPath', JOB_PAYLOAD_REMEDIATION) : undefined; const noPull = !resolveJobPull(job.data); // noEmbed defaults to true (embed is a separate job — submit `embed --stale` @@ -1704,9 +1724,14 @@ export async function registerBuiltinHandlers( const mode = (typeof job.data.mode === 'string' && ['links', 'timeline', 'all'].includes(job.data.mode)) ? (job.data.mode as 'links' | 'timeline' | 'all') : 'all'; + // repo-path.ts invariant: job payloads + the config anchor are STORAGE — + // a relative dir would resolve against the worker daemon's cwd (typically + // / under launchd), and post-migration-127 a cleared anchor must fail + // loudly instead of silently becoming '.' (the wrong-tree footgun the + // migration exists to close). Mirrors the sibling 'sync' handler. const dir = typeof job.data.dir === 'string' - ? job.data.dir - : (await engine.getConfig('sync.repo_path')) ?? '.'; + ? requireAbsoluteStoredPath(job.data.dir, 'job.data.dir', JOB_PAYLOAD_REMEDIATION) + : requireConfiguredRepoAnchor(await engine.getConfig('sync.repo_path'), 'extract'); return await runExtractCore(engine, { mode, dir, dryRun: !!job.data.dryRun }); }); @@ -1719,9 +1744,11 @@ export async function registerBuiltinHandlers( // (runPhaseBacklinks). The filesystem fixer stays available explicitly // via '{"action":"fix"}' or `gbrain check-backlinks fix`. const action: 'check' | 'fix' = job.data.action === 'fix' ? 'fix' : 'check'; + // repo-path.ts invariant: same guard as the 'extract' handler above — no + // '.' (worker-cwd) fallback for a NULL/relative stored anchor. const dir = typeof job.data.dir === 'string' - ? job.data.dir - : (await engine.getConfig('sync.repo_path')) ?? '.'; + ? requireAbsoluteStoredPath(job.data.dir, 'job.data.dir', JOB_PAYLOAD_REMEDIATION) + : requireConfiguredRepoAnchor(await engine.getConfig('sync.repo_path'), 'backlinks'); return await runBacklinksCore({ action, dir, dryRun: !!job.data.dryRun }); }); @@ -1795,9 +1822,18 @@ export async function registerBuiltinHandlers( // postgres brain should skip filesystem phases (no_brain_dir) and run the // DB-only phases (resolve_symbol_edges, embed, ...) — not silently lint/sync // against whatever directory the worker happens to be running in. - const repoPath: string | null = typeof job.data.repoPath === 'string' - ? job.data.repoPath - : (await engine.getConfig('sync.repo_path')) ?? null; + // + // repo-path.ts invariant (red-team): both the job payload and the config + // anchor are STORAGE — a relative value must fail loudly (mirroring the + // sibling 'sync' handler), never resolve against the worker daemon's cwd. + // Null stays null: checkout-less brains keep their skip-FS-phases contract. + let repoPath: string | null; + if (typeof job.data.repoPath === 'string') { + repoPath = requireAbsoluteStoredPath(job.data.repoPath, 'job.data.repoPath', JOB_PAYLOAD_REMEDIATION); + } else { + const anchor = await engine.getConfig('sync.repo_path'); + repoPath = anchor ? requireAbsoluteStoredPath(anchor, 'config sync.repo_path') : null; + } // v0.38 (codex r1 P1-2 + P1-5): per-source dispatch threading. // - source_id: when set, runCycle uses the per-source lock ID and diff --git a/src/commands/sync.ts b/src/commands/sync.ts index b9bbd4bd7..b7c8cedae 100644 --- a/src/commands/sync.ts +++ b/src/commands/sync.ts @@ -1,7 +1,7 @@ import { existsSync, readFileSync, writeFileSync, statSync, realpathSync } from 'fs'; import { execFileSync } from 'child_process'; import { isAbsolute, join, relative, sep } from 'path'; -import { resolveRepoArg, requireAbsoluteStoredPath } from '../core/repo-path.ts'; +import { resolveRepoArg, requireAbsoluteStoredPath, sourceLocalPathRemediation } from '../core/repo-path.ts'; import type { BrainEngine } from '../core/engine.ts'; import { DELETE_BATCH_SIZE } from '../core/engine-constants.ts'; import { importFile } from '../core/import-file.ts'; @@ -1294,7 +1294,7 @@ async function readSyncAnchor( * scope it to — collection happens after this point) — see the P2 review * finding on `createSyncBaselineCommit`'s callers. */ -async function isAnchorOwnedSyncPath( +export async function isAnchorOwnedSyncPath( engine: BrainEngine, opts: SyncOpts, repoPath: string, @@ -1303,6 +1303,13 @@ async function isAnchorOwnedSyncPath( if (opts.sourceId && opts.sourceId !== 'default') return false; const anchor = await readSyncAnchor(engine, opts.sourceId, 'repo_path'); if (anchor === null) return false; + // repo-path.ts invariant (red-team): realpathSync would resolve a relative + // stored anchor against THIS process's cwd — exactly the silent resolution + // the invariant forbids, and it could "prove" ownership of whatever tree + // the process happens to run in. A relative anchor can never vouch for a + // path; return no-match (self-heal doesn't fire — performSync's + // resolve_repo guard is the loud surface for the bad anchor itself). + if (!isAbsolute(anchor)) return false; try { return realpathSync(anchor) === realpathSync(repoPath); } catch { @@ -1841,6 +1848,7 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise boolean) | null): void { _liveWorkerProbeOverride = p; } +/** + * Queue the durable facts-absorb job is submitted to. The live-worker probe + * MUST check this same queue — a worker draining a different queue can never + * claim the job, so its presence proves nothing. + */ +const FACTS_ABSORB_QUEUE = 'default'; + +/** + * Short-TTL memo for the live-worker probe. readWorkers() execs `ps` + * synchronously per registry entry, and un-memoized that cost lands inside + * EVERY remote put_page. 5s of staleness is harmless here — worst case one + * page's facts run in-process (worker just started) or ride the in-process + * queue once more (worker just died; the durable submit would have parked + * anyway). + */ +const WORKER_PROBE_TTL_MS = 5_000; +let _workerProbeCache: { at: number; value: boolean } | null = null; +/** Test-only: reset the worker-probe TTL memo. */ +export function __resetWorkerProbeCacheForTests(): void { + _workerProbeCache = null; +} + /** * Is a live minion worker available to process a durable facts-absorb job - * for this brain? Cheap file-registry read (no DB). Fail-open to `false`: - * a probe error must never block the in-process fallback. + * for this brain — on the queue the job actually goes to? Cheap file-registry + * read (no DB), memoized for WORKER_PROBE_TTL_MS. Fail-open to `false`: a + * probe error must never block the in-process fallback. + * + * Red-team (#2108 residual): the probe is queue-aware. A queue-blind + * presence check let a worker on another queue (`gbrain jobs work --queue + * shell`) satisfy the probe while never claiming the 'default'-queue job — + * parking it forever AND skipping the in-process fallback: the exact silent + * loss #2108 closed. */ async function hasLiveMinionWorker(): Promise { if (_liveWorkerProbeOverride) return _liveWorkerProbeOverride(); + const now = Date.now(); + if (_workerProbeCache !== null && now - _workerProbeCache.at < WORKER_PROBE_TTL_MS) { + return _workerProbeCache.value; + } + let value = false; try { const { readWorkers } = await import('../minions/worker-registry.ts'); - return readWorkers().length > 0; + value = readWorkers().some((w) => w.queue === FACTS_ABSORB_QUEUE); } catch { - return false; + value = false; } + _workerProbeCache = { at: now, value }; + return value; } /** @@ -191,7 +227,7 @@ async function submitDurableFactsAbsorb( ...(ctx.model ? { model: ctx.model } : {}), }, { - queue: 'default', + queue: FACTS_ABSORB_QUEUE, idempotency_key: `facts-absorb:${ctx.sourceId}:${parsedPage.slug}:${contentHash}`, max_attempts: 3, timeout_ms: 180_000, @@ -317,9 +353,16 @@ export async function runFactsBackstop( ); } } catch (err) { + // Red-team: honest abort detection. The old `/abort/i` message sniff + // also matched Postgres's routine "current transaction is aborted, + // commands ignored until end of transaction block" — a plain DB + // failure that would spuriously requeue AND print the exit-drain + // stderr banner. The real drain path needs no message sniff: + // queue.shutdown() aborts `signal` before in-flight work observes + // it, and a gateway fetch abort throws with err.name === + // 'AbortError'. Errors are classified by NAME + signal state only. const aborted = - signal.aborted || - (err instanceof Error && (err.name === 'AbortError' || /abort/i.test(err.message))); + signal.aborted || (err instanceof Error && err.name === 'AbortError'); if (aborted) { await requeueAborted(); } diff --git a/src/core/repo-path.ts b/src/core/repo-path.ts index f00235120..07f99872b 100644 --- a/src/core/repo-path.ts +++ b/src/core/repo-path.ts @@ -24,18 +24,44 @@ export function resolveRepoArg(p: string): string { return resolve(p); } +/** + * Remediation hint for `sources.local_path` rows: the per-source sync repoint + * (matches the `sync --all` relative-path skip message shape). + */ +export function sourceLocalPathRemediation(sourceId: string): string { + return `Fix it once with an absolute path: gbrain sync --source ${sourceId} --repo .`; +} + +/** + * Remediation hint for minion job payloads: a queued row can't be edited in + * place — the only fix is a fresh submission. + */ +export const JOB_PAYLOAD_REMEDIATION = + 'Cancel this job and resubmit it with an absolute path.'; + /** * Guard for repo paths read back from storage. Refuses to resolve a relative * value against the current cwd — that is exactly the wrong-tree footgun: * whichever directory the next bare invocation happens to run from becomes * the sync source. + * + * `remediation` lets the caller name the fix command that matches WHERE the + * bad value is stored (sources.local_path → per-source sync repoint; job + * payload → resubmit; default → the config-anchor sync --repo / config set + * pair). Optional for backward compatibility. */ -export function requireAbsoluteStoredPath(value: string, storageDesc: string): string { +export function requireAbsoluteStoredPath( + value: string, + storageDesc: string, + remediation?: string, +): string { if (isAbsolute(value)) return value; + const fix = remediation ?? + 'Fix it once with an absolute path: gbrain sync --repo ' + + '(or: gbrain config set sync.repo_path ).'; throw new Error( `${storageDesc} holds a relative path "${value}". Refusing to resolve it against ` + `the current directory — a bare invocation from the wrong cwd would sync the wrong ` + - `tree. Fix it once with an absolute path: gbrain sync --repo ` + - `(or: gbrain config set sync.repo_path ).`, + `tree. ${fix}`, ); } diff --git a/test/facts-backstop.test.ts b/test/facts-backstop.test.ts index c0e0079c2..2ed5620f1 100644 --- a/test/facts-backstop.test.ts +++ b/test/facts-backstop.test.ts @@ -11,7 +11,11 @@ import { describe, test, expect, beforeAll, afterAll, afterEach } from 'bun:test'; import { PGLiteEngine } from '../src/core/pglite-engine.ts'; -import { runFactsBackstop, __setLiveWorkerProbeForTests } from '../src/core/facts/backstop.ts'; +import { + runFactsBackstop, + __setLiveWorkerProbeForTests, + __resetWorkerProbeCacheForTests, +} from '../src/core/facts/backstop.ts'; import type { FactsBackstopCtx } from '../src/core/facts/backstop.ts'; import { __setChatTransportForTests, @@ -324,6 +328,7 @@ describe('runFactsBackstop — durable facts-absorb (#2108)', () => { afterEach(async () => { __setLiveWorkerProbeForTests(null); + __resetWorkerProbeCacheForTests(); // eslint-disable-next-line @typescript-eslint/no-explicit-any await (engine as any).db.query(`DELETE FROM minion_jobs WHERE name = 'facts-absorb'`); }); @@ -363,6 +368,139 @@ describe('runFactsBackstop — durable facts-absorb (#2108)', () => { expect((await absorbJobsFor(page.slug)).length).toBe(0); }); + // --------------------------------------------------------------------- + // Red-team #1: the probe must be QUEUE-aware. The durable job goes to + // queue 'default'; a worker draining any other queue can never claim it, + // so its presence must not park the job (skipping the in-process + // fallback = the exact silent loss #2108 closed). These tests exercise + // the REAL registry probe (no override) via a temp GBRAIN_HOME. + // --------------------------------------------------------------------- + + const withWorkerRegistry = async ( + fn: (registerLiveWorker: (queue: string) => () => void) => Promise, + ): Promise => { + const { mkdtempSync, rmSync } = await import('node:fs'); + const { tmpdir } = await import('node:os'); + const { join } = await import('node:path'); + const { registerWorker } = await import('../src/core/minions/worker-registry.ts'); + const { withEnv } = await import('./helpers/with-env.ts'); + const home = mkdtempSync(join(tmpdir(), 'backstop-worker-probe-')); + const cleanups: Array<() => void> = []; + try { + await withEnv({ GBRAIN_HOME: home }, async () => { + __resetWorkerProbeCacheForTests(); + await fn((queue: string) => { + const cleanup = registerWorker({ + pid: process.pid, // this live test process — passes the liveness + PID-reuse guards + queue, + nice_requested: null, + nice_effective: null, + started_at: Date.now(), + }); + cleanups.push(cleanup); + return cleanup; + }); + }); + } finally { + for (const c of cleanups) c(); + __resetWorkerProbeCacheForTests(); + rmSync(home, { recursive: true, force: true }); + } + }; + + test('wrong-queue worker does NOT satisfy the probe → in-process fallback (red-team #1)', async () => { + await withWorkerRegistry(async (registerLiveWorker) => { + registerLiveWorker('shell'); // live worker, but on a queue the job never goes to + chatStub([]); + const page = meetingPage(); + const r = await runFactsBackstop(page, makeCtx({ preferDurableAbsorb: true })); + expect(r.mode).toBe('queue'); + if (r.mode === 'queue') expect(r.enqueued).toBe(true); + // No durable job parked on a queue nobody drains — the in-process + // queue owns the work. + expect((await absorbJobsFor(page.slug)).length).toBe(0); + }); + }); + + test('default-queue worker satisfies the probe → durable minion job', async () => { + await withWorkerRegistry(async (registerLiveWorker) => { + registerLiveWorker('default'); + const page = meetingPage(); + // NO chat stub: the durable path must not run the LLM in this process. + const r = await runFactsBackstop(page, makeCtx({ preferDurableAbsorb: true })); + expect(r.mode).toBe('queue'); + if (r.mode === 'queue') expect(r.enqueued).toBe(true); + const jobs = await absorbJobsFor(page.slug); + expect(jobs.length).toBe(1); + expect(jobs[0].status).toBe('waiting'); + }); + }); + + test('probe result is memoized (short TTL) — hot-path readWorkers/ps cost', async () => { + await withWorkerRegistry(async (registerLiveWorker) => { + // First probe: no workers → false, memoized. + chatStub([]); + const p1 = meetingPage(); + await runFactsBackstop(p1, makeCtx({ preferDurableAbsorb: true })); + expect((await absorbJobsFor(p1.slug)).length).toBe(0); + + // A default-queue worker appears, but WITHOUT a cache reset the + // memoized `false` still wins inside the TTL window. + registerLiveWorker('default'); + chatStub([]); + const p2 = meetingPage(); + await runFactsBackstop(p2, makeCtx({ preferDurableAbsorb: true })); + expect((await absorbJobsFor(p2.slug)).length).toBe(0); + + // After a reset (TTL-expiry stand-in) the live worker is seen. + __resetWorkerProbeCacheForTests(); + const p3 = meetingPage(); + await runFactsBackstop(p3, makeCtx({ preferDurableAbsorb: true })); + expect((await absorbJobsFor(p3.slug)).length).toBe(1); + }); + }); + + // --------------------------------------------------------------------- + // Red-team #5: honest abort detection. A Postgres-style "current + // transaction is aborted, commands ignored…" error is a plain pipeline + // failure — it must land in the absorb log WITHOUT triggering the + // durable requeue + exit-drain stderr banner (the old /abort/i message + // sniff matched it). + // --------------------------------------------------------------------- + + test("Postgres 'transaction is aborted' error does NOT trigger the durable requeue (red-team #5)", async () => { + const page = meetingPage(); + __setChatTransportForTests(async () => { + throw new Error('current transaction is aborted, commands ignored until end of transaction block'); + }); + + const r = await runFactsBackstop(page, makeCtx()); // default queue mode, in-process + expect(r.mode).toBe('queue'); + if (r.mode === 'queue') expect(r.enqueued).toBe(true); + + // Wait for the in-process worker to settle: the absorb-log row is + // written AFTER the requeue decision, so its presence proves the + // catch block completed. + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const logCount = async (): Promise => { + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const res = await (engine as any).db.query( + `SELECT COUNT(*)::int AS n FROM ingest_log WHERE source_type = 'facts:absorb' AND source_ref = $1`, + [page.slug], + ); + return res.rows[0].n as number; + }; + const deadline = Date.now() + 5000; + while ((await logCount()) === 0 && Date.now() < deadline) { + await new Promise((res) => setTimeout(res, 10)); + } + expect(await logCount()).toBeGreaterThanOrEqual(1); + + // The load-bearing assertion: NOT classified as an abort — no durable + // requeue happened. + expect((await absorbJobsFor(page.slug)).length).toBe(0); + }); + test('drain-abort → nothing stamped, durable retry job queued, next pass extracts', async () => { const page = meetingPage(); const before = await factsCount(); diff --git a/test/handlers.test.ts b/test/handlers.test.ts index 7a4ec3d4f..73bbaaa23 100644 --- a/test/handlers.test.ts +++ b/test/handlers.test.ts @@ -282,3 +282,110 @@ describe('autopilot-cycle handler — phase passthrough', () => { expect(phaseNames).toContain('embed'); }, 30_000); }); + +// --------------------------------------------------------------------------- +// Red-team (post-migration v127): worker handlers must never fall back to +// '.' / worker-daemon cwd when the stored anchor is NULL (v127 clears +// relative anchors) or relative (legacy row). The 'sync' handler already +// validated; these pin the same guard on extract / backlinks / +// autopilot-cycle. +// --------------------------------------------------------------------------- + +describe('extract/backlinks/autopilot-cycle handlers — no cwd fallback (repo-path invariant)', () => { + const clearAnchor = async () => { + await engine.executeRaw(`DELETE FROM config WHERE key = 'sync.repo_path'`); + }; + + test('extract: NULL anchor + no data.dir → loud failure naming the fix, not cwd', async () => { + await clearAnchor(); + const handler = (worker as any).handlers.get('extract'); + expect(handler).toBeDefined(); + await expect(handler({ + data: {}, + signal: { aborted: false } as any, + job: { id: 101, name: 'extract' } as any, + })).rejects.toThrow(/no sync\.repo_path anchor is configured.*gbrain sync --repo /s); + }); + + test('extract: relative stored anchor → refused, not resolved against cwd', async () => { + await engine.setConfig('sync.repo_path', '.'); + try { + const handler = (worker as any).handlers.get('extract'); + await expect(handler({ + data: {}, + signal: { aborted: false } as any, + job: { id: 102, name: 'extract' } as any, + })).rejects.toThrow(/config sync\.repo_path holds a relative path "\."/); + } finally { + await clearAnchor(); + } + }); + + test('extract: relative job.data.dir → refused with the resubmit hint', async () => { + const handler = (worker as any).handlers.get('extract'); + await expect(handler({ + data: { dir: 'relative/tree' }, + signal: { aborted: false } as any, + job: { id: 103, name: 'extract' } as any, + })).rejects.toThrow(/job\.data\.dir holds a relative path.*resubmit/s); + }); + + test('backlinks: NULL anchor + no data.dir → loud failure, not cwd', async () => { + await clearAnchor(); + const handler = (worker as any).handlers.get('backlinks'); + expect(handler).toBeDefined(); + await expect(handler({ + data: {}, + signal: { aborted: false } as any, + job: { id: 104, name: 'backlinks' } as any, + })).rejects.toThrow(/backlinks job has no data\.dir and no sync\.repo_path anchor/); + }); + + test('backlinks: relative stored anchor → refused', async () => { + await engine.setConfig('sync.repo_path', 'brain'); + try { + const handler = (worker as any).handlers.get('backlinks'); + await expect(handler({ + data: {}, + signal: { aborted: false } as any, + job: { id: 105, name: 'backlinks' } as any, + })).rejects.toThrow(/holds a relative path "brain"/); + } finally { + await clearAnchor(); + } + }); + + test('autopilot-cycle: relative job.data.repoPath → refused (mirrors sync handler)', async () => { + const handler = (worker as any).handlers.get('autopilot-cycle'); + await expect(handler({ + data: { repoPath: 'relative/tree' }, + signal: { aborted: false } as any, + job: { id: 106, name: 'autopilot-cycle' } as any, + })).rejects.toThrow(/job\.data\.repoPath holds a relative path.*[Cc]ancel this job and resubmit/s); + }); + + test('autopilot-cycle: relative stored anchor → refused; NULL anchor stays null (skip-FS contract)', async () => { + await engine.setConfig('sync.repo_path', '.'); + try { + const handler = (worker as any).handlers.get('autopilot-cycle'); + await expect(handler({ + data: {}, + signal: { aborted: false } as any, + job: { id: 107, name: 'autopilot-cycle' } as any, + })).rejects.toThrow(/config sync\.repo_path holds a relative path/); + } finally { + await clearAnchor(); + } + + // NULL anchor: the handler must NOT throw — checkout-less brains run + // DB-only phases (v0.41.30 T2 contract). Scope to a cheap phase. + const handler = (worker as any).handlers.get('autopilot-cycle'); + const result = await handler({ + data: { phases: ['orphans'], pull: false }, + signal: { aborted: false } as any, + job: { id: 108, name: 'autopilot-cycle' } as any, + }); + expect(result).toBeDefined(); + expect('report' in (result as any)).toBe(true); + }, 30_000); +}); diff --git a/test/repo-path.test.ts b/test/repo-path.test.ts index 7fa6a729c..778c4d9fc 100644 --- a/test/repo-path.test.ts +++ b/test/repo-path.test.ts @@ -10,17 +10,23 @@ * refuses at the resolve_repo phase instead of importing the foreign cwd. */ -import { test, expect, describe, beforeAll, afterAll, beforeEach } from 'bun:test'; +import { test, expect, describe, beforeAll, afterAll, beforeEach, spyOn } from 'bun:test'; import { resolve, isAbsolute, join } from 'path'; import { mkdtempSync, writeFileSync, rmSync, realpathSync } from 'fs'; import { execFileSync } from 'child_process'; import { tmpdir } from 'os'; import { PGLiteEngine } from '../src/core/pglite-engine.ts'; import { resetPgliteState } from './helpers/reset-pglite.ts'; -import { resolveRepoArg, requireAbsoluteStoredPath } from '../src/core/repo-path.ts'; +import { + resolveRepoArg, + requireAbsoluteStoredPath, + sourceLocalPathRemediation, + JOB_PAYLOAD_REMEDIATION, +} from '../src/core/repo-path.ts'; import { addSource } from '../src/core/sources-ops.ts'; -import { performSync } from '../src/commands/sync.ts'; +import { performSync, isAnchorOwnedSyncPath } from '../src/commands/sync.ts'; import { runImport } from '../src/commands/import.ts'; +import { runConfig } from '../src/commands/config.ts'; import { runMigrations } from '../src/core/migrate.ts'; let engine: PGLiteEngine; @@ -73,6 +79,28 @@ describe('requireAbsoluteStoredPath', () => { expect(() => requireAbsoluteStoredPath('brain', 'sources.local_path for "default"')) .toThrow(/gbrain sync --repo /); }); + + test('default remediation carries the sync --repo / config set pair', () => { + expect(() => requireAbsoluteStoredPath('.', 'config sync.repo_path')) + .toThrow(/gbrain sync --repo .*gbrain config set sync\.repo_path /s); + }); + + test('sources.local_path remediation names the per-source repoint command', () => { + expect(() => + requireAbsoluteStoredPath('.', 'sources.local_path for "wiki"', sourceLocalPathRemediation('wiki')), + ).toThrow(/gbrain sync --source wiki --repo /); + }); + + test('job-payload remediation says cancel + resubmit, not a config command', () => { + expect(() => + requireAbsoluteStoredPath('.', 'job.data.repoPath', JOB_PAYLOAD_REMEDIATION), + ).toThrow(/Cancel this job and resubmit it with an absolute path/); + // The parameterized hint REPLACES the default — the config-set command + // would be misleading remediation for a queued row. + expect(() => + requireAbsoluteStoredPath('.', 'job.data.repoPath', JOB_PAYLOAD_REMEDIATION), + ).not.toThrow(/config set sync\.repo_path/); + }); }); // --------------------------------------------------------------------------- @@ -131,6 +159,78 @@ describe('performSync stored-anchor guard', () => { }); }); +// --------------------------------------------------------------------------- +// isAnchorOwnedSyncPath — a relative stored anchor can never vouch for a path +// (red-team: realpathSync silently resolved it against cwd, so a "." anchor +// "proved" ownership of whatever tree the process ran in) +// --------------------------------------------------------------------------- + +describe('isAnchorOwnedSyncPath relative-anchor guard', () => { + test('relative anchor "." returns false even when cwd would realpath-match', async () => { + await engine.setConfig('sync.repo_path', '.'); + // Pre-fix: realpathSync('.') === realpathSync(process.cwd()) → true. + const owned = await isAnchorOwnedSyncPath(engine, {}, process.cwd()); + expect(owned).toBe(false); + }); + + test('absolute anchor still proves ownership by realpath identity', async () => { + await engine.setConfig('sync.repo_path', process.cwd()); + expect(await isAnchorOwnedSyncPath(engine, {}, process.cwd())).toBe(true); + // Different tree → no match. + expect(await isAnchorOwnedSyncPath(engine, {}, tmpdir())).toBe(false); + }); +}); + +// --------------------------------------------------------------------------- +// config set sync.repo_path — ingress normalization (red-team: `gbrain config +// set sync.repo_path .` re-seeded a relative anchor around every other +// writer's normalization) +// --------------------------------------------------------------------------- + +describe('config set sync.repo_path normalization', () => { + test('a relative value is resolved absolute before persisting (and echoed resolved)', async () => { + const logSpy = spyOn(console, 'log').mockImplementation(() => {}); + try { + await runConfig(engine, ['set', 'sync.repo_path', '.']); + expect(await engine.getConfig('sync.repo_path')).toBe(process.cwd()); + const confirmation = logSpy.mock.calls.map((c) => String(c[0])).find((l) => l.startsWith('Set ')); + expect(confirmation).toBe(`Set sync.repo_path = ${process.cwd()}`); + } finally { + logSpy.mockRestore(); + } + }); + + test('an absolute value is persisted unchanged', async () => { + const logSpy = spyOn(console, 'log').mockImplementation(() => {}); + const abs = resolve('/tmp', 'gbrain-config-set-abs'); + try { + await runConfig(engine, ['set', 'sync.repo_path', abs]); + expect(await engine.getConfig('sync.repo_path')).toBe(abs); + } finally { + logSpy.mockRestore(); + } + }); + + test('a whitespace-only value is refused (exit 1), anchor untouched', async () => { + const logSpy = spyOn(console, 'log').mockImplementation(() => {}); + const errSpy = spyOn(console, 'error').mockImplementation(() => {}); + const exitSpy = spyOn(process, 'exit').mockImplementation(((code?: number) => { + throw new Error(`process.exit(${code})`); + }) as never); + try { + await expect(runConfig(engine, ['set', 'sync.repo_path', ' '])) + .rejects.toThrow('process.exit(1)'); + expect(await engine.getConfig('sync.repo_path')).toBeFalsy(); + const errLines = errSpy.mock.calls.map((c) => String(c[0])).join('\n'); + expect(errLines).toContain('sync.repo_path cannot be empty'); + } finally { + exitSpy.mockRestore(); + errSpy.mockRestore(); + logSpy.mockRestore(); + } + }); +}); + // --------------------------------------------------------------------------- // import anchor guard (#2114) — set-if-unset, repoint only with --set-repo-path // --------------------------------------------------------------------------- @@ -153,19 +253,36 @@ describe('import sync.repo_path anchor guard (#2114)', () => { test('seeds when unset; refuses silent overwrite; repoints with --set-repo-path', async () => { const dirA = makeGitRepo('gbrain-anchor-a-'); const dirB = makeGitRepo('gbrain-anchor-b-'); + // Red-team FIX 6: the anchor decision was stderr-only — --json consumers + // couldn't see it. It now rides the return shape AND the JSON summary. + const logSpy = spyOn(console, 'log').mockImplementation(() => {}); try { // First import seeds the anchor. - await runImport(engine, [dirA, '--no-embed', '--json']); + const r1 = await runImport(engine, [dirA, '--no-embed', '--json']); expect(await engine.getConfig('sync.repo_path')).toBe(dirA); + expect(r1.anchor).toBe('seeded'); + // The --json stdout summary carries the same field. + const jsonLine = logSpy.mock.calls + .map((c) => String(c[0])) + .find((l) => l.startsWith('{') && l.includes('"anchor"')); + expect(jsonLine).toBeDefined(); + expect(JSON.parse(jsonLine!).anchor).toBe('seeded'); // Second import from a DIFFERENT tree must not silently repoint. - await runImport(engine, [dirB, '--no-embed', '--json']); + const r2 = await runImport(engine, [dirB, '--no-embed', '--json']); expect(await engine.getConfig('sync.repo_path')).toBe(dirA); + expect(r2.anchor).toBe('kept'); // Explicit flag repoints. - await runImport(engine, [dirB, '--no-embed', '--json', '--set-repo-path']); + const r3 = await runImport(engine, [dirB, '--no-embed', '--json', '--set-repo-path']); expect(await engine.getConfig('sync.repo_path')).toBe(dirB); + expect(r3.anchor).toBe('repointed'); + + // Re-import of the tree the anchor already points at → 'kept'. + const r4 = await runImport(engine, [dirB, '--no-embed', '--json']); + expect(r4.anchor).toBe('kept'); } finally { + logSpy.mockRestore(); rmSync(dirA, { recursive: true, force: true }); rmSync(dirB, { recursive: true, force: true }); }