diff --git a/src/cli.ts b/src/cli.ts index a5ca466f7..a73f23839 100755 --- a/src/cli.ts +++ b/src/cli.ts @@ -235,6 +235,17 @@ async function main() { command = 'query'; } + // Local patch 2026-06-11 — mark one-shot CLI processes so the facts + // backstop routes absorb work to the durable jobs worker instead of the + // in-process queue that the exit teardown drains-then-aborts after ~1-2s + // (the `pipeline_error: [chat(...)] The operation was aborted.` class in + // ingest_log). Daemons keep the in-process queue: their event loop + // outlives the work. See src/core/facts/cli-process-mode.ts. + if (!['serve', 'jobs', 'autopilot'].includes(command)) { + const { markShortLivedCliProcess } = await import('./core/facts/cli-process-mode.ts'); + markShortLivedCliProcess(); + } + // T5 — `gbrain search modes|stats|tune` is the read-only config dashboard, // NOT a free-text search for the literal word "modes". Free-text // `gbrain search ""` falls through to the cheap-hybrid `search` op diff --git a/src/commands/jobs.ts b/src/commands/jobs.ts index 64aaa6a83..32902b55a 100644 --- a/src/commands/jobs.ts +++ b/src/commands/jobs.ts @@ -1612,6 +1612,43 @@ export async function registerBuiltinHandlers( return await runBacklinksCore({ action, dir, dryRun: !!job.data.dryRun }); }); + // Local patch 2026-06-11: durable facts:absorb. One-shot CLI processes + // (capture/put/sync) can't finish the extraction chat before their exit + // drain aborts it, so backstop.ts submits this job instead and the + // long-lived worker does the LLM work here. Inline mode: errors throw, + // so minion retry/backoff handles transient gateway failures and real + // failures stay visible in `gbrain jobs list --status failed`. + worker.register('facts-absorb', async (job) => { + const slug = typeof job.data.slug === 'string' ? job.data.slug : ''; + if (!slug) throw new Error('facts-absorb job requires data.slug'); + const sourceId = typeof job.data.sourceId === 'string' ? job.data.sourceId : 'default'; + const page = await engine.getPage(slug, { sourceId }); + if (!page) return { skipped: 'page_missing', slug, sourceId }; + const { runFactsBackstop } = await import('../core/facts/backstop.ts'); + const KNOWN_SOURCES = ['sync:import', 'mcp:put_page', 'mcp:extract_facts', 'file_upload', 'code_import'] as const; + const source = (KNOWN_SOURCES as readonly string[]).includes(job.data.source as string) + ? (job.data.source as typeof KNOWN_SOURCES[number]) + : 'mcp:put_page'; + return await runFactsBackstop( + { + slug: page.slug, + type: page.type, + compiled_truth: page.compiled_truth, + frontmatter: (page.frontmatter ?? {}) as Record, + }, + { + engine, + sourceId, + sessionId: typeof job.data.sessionId === 'string' ? job.data.sessionId : null, + source, + mode: 'inline', + notabilityFilter: job.data.notabilityFilter === 'high-only' ? 'high-only' : 'all', + visibility: job.data.visibility === 'world' ? 'world' : 'private', + ...(typeof job.data.model === 'string' && job.data.model ? { model: job.data.model } : {}), + }, + ); + }); + // Autopilot-cycle handler: delegates to runCycle. Shares the exact same // phase set and ordering as `gbrain dream` and autopilot's inline path — // one source of truth for what the brain does overnight. diff --git a/src/core/facts/backstop.ts b/src/core/facts/backstop.ts index eda17b993..59463d13c 100644 --- a/src/core/facts/backstop.ts +++ b/src/core/facts/backstop.ts @@ -157,6 +157,52 @@ export async function runFactsBackstop( // --- Mode dispatch --- if (mode === 'queue') { + // Local patch 2026-06-11: in a one-shot CLI process the in-process queue + // is doomed — cli.ts's exit drain aborts the in-flight chat after ~1-2s, + // so every CLI capture logged `pipeline_error: [chat(...)] The operation + // was aborted.` and extracted nothing. Submit a durable facts-absorb + // minion job for the long-lived jobs worker instead. Falls through to + // the in-process queue if durable submission fails (old schema, no + // minions infra), preserving prior behavior + absorb-log visibility. + const { isShortLivedCliProcess } = await import('./cli-process-mode.ts'); + if (isShortLivedCliProcess()) { + try { + const { MinionQueue } = await import('../minions/queue.ts'); + const { createHash } = await import('node:crypto'); + const contentHash = createHash('sha256') + .update(parsedPage.compiled_truth) + .digest('hex') + .slice(0, 16); + const minions = new MinionQueue(ctx.engine); + await minions.add( + 'facts-absorb', + { + slug: parsedPage.slug, + sourceId: ctx.sourceId, + source: ctx.source, + sessionId: ctx.sessionId, + notabilityFilter: ctx.notabilityFilter ?? 'all', + visibility: ctx.visibility ?? 'private', + ...(ctx.model ? { model: ctx.model } : {}), + }, + { + queue: 'default', + // Content-hash key: re-submits after edits, dedups rapid + // identical writes (idempotent ON CONFLICT returns existing row). + idempotency_key: `facts-absorb:${ctx.sourceId}:${parsedPage.slug}:${contentHash}`, + max_attempts: 3, + timeout_ms: 180_000, + }, + ); + return { mode: 'queue', enqueued: true, queueDepth: 0 }; + } catch (err) { + const msg = err instanceof Error ? err.message : String(err); + warnOnce( + 'facts-absorb-job-submit', + `[facts] durable facts-absorb submit failed (${msg}); falling back to in-process queue`, + ); + } + } const { getFactsQueue } = await import('./queue.ts'); const queue = getFactsQueue(); const enqueued = queue.enqueue(async (signal) => { diff --git a/src/core/facts/cli-process-mode.ts b/src/core/facts/cli-process-mode.ts new file mode 100644 index 000000000..f63a2adaa --- /dev/null +++ b/src/core/facts/cli-process-mode.ts @@ -0,0 +1,35 @@ +/** + * Local patch 2026-06-11 — short-lived-CLI marker for the facts backstop. + * + * Root cause: every `gbrain capture`/`put` from a one-shot CLI process + * enqueues the facts:absorb chat call into the in-process FactsQueue, then + * cli.ts's exit teardown drains background work for 1-2s and ABORTS the + * in-flight chat (the extraction call takes 5-30s). Result: a + * `pipeline_error: [chat(...)] The operation was aborted.` ingest_log row on + * every CLI-written eligible page since the v0.42.20.0 drain-then-abort + * teardown landed, and no facts ever extracted for those pages. + * + * Fix: cli.ts marks one-shot processes via markShortLivedCliProcess(); + * runFactsBackstop's queue mode checks isShortLivedCliProcess() and submits + * a durable `facts-absorb` minion job (processed by the long-lived + * `gbrain jobs work` daemon) instead of the doomed in-process enqueue. + * + * A marker (not argv heuristics) so tests and embedded/server callers are + * never affected: only cli.ts sets it, and never for daemon commands + * (serve / jobs / autopilot). + */ + +let _shortLivedCli = false; + +export function markShortLivedCliProcess(): void { + _shortLivedCli = true; +} + +export function isShortLivedCliProcess(): boolean { + return _shortLivedCli; +} + +/** @internal — test seam */ +export function __resetShortLivedCliForTests(): void { + _shortLivedCli = false; +} diff --git a/src/core/facts/fence-write.ts b/src/core/facts/fence-write.ts index febd2cb12..ae35769fb 100644 --- a/src/core/facts/fence-write.ts +++ b/src/core/facts/fence-write.ts @@ -34,9 +34,10 @@ */ import { existsSync, mkdirSync, readFileSync, writeFileSync, renameSync, appendFileSync } from 'node:fs'; -import { join, dirname } from 'node:path'; +import { dirname } from 'node:path'; import type { BrainEngine, NewFact, FactVisibility } from '../engine.ts'; +import { resolvePageFilePath } from '../markdown.ts'; import { withPageLock } from '../page-lock.ts'; import { gbrainPath } from '../config.ts'; import { upsertFactRow, parseFactsFence } from '../facts-fence.ts'; @@ -166,7 +167,12 @@ export async function writeFactsToFence( return { inserted: 0, ids: [] }; } - const filePath = join(target.localPath, `${target.slug}.md`); + // Local patch 2026-06-11: route through resolvePageFilePath so non-default + // sources fence into `/.sources//.md` — the same path + // the put_page write-through and dream-cycle reverse-render compute. The + // bare join wrote main-source fences to the repo ROOT (the default source's + // tree), polluting ~/brain with stray root-level fence files. + const filePath = resolvePageFilePath(target.localPath, target.slug, target.sourceId); const tmpPath = `${filePath}.tmp`; return withPageLock(