diff --git a/docs/architecture/KEY_FILES.md b/docs/architecture/KEY_FILES.md index acbaacf62..277d92b25 100644 --- a/docs/architecture/KEY_FILES.md +++ b/docs/architecture/KEY_FILES.md @@ -194,7 +194,7 @@ per-release `**vX.Y.Z:**` narration — CI enforces this - `src/commands/embed.ts` — `gbrain embed [--stale|--all] [--slugs ...]`. `--stale` starts with `engine.countStaleChunks()` (single SELECT count(*) WHERE embedding IS NULL, ~50 bytes wire) so a fully-embedded brain short-circuits with no further reads. When stale chunks exist, `engine.listStaleChunks()` returns just the chunks needing embeddings (slug + chunk_index + chunk_text + metadata, no `vector(1536)` payload); caller groups by slug, embeds, re-upserts via `upsertChunks`. All `console.log`/`console.error` call sites use `slog`/`serr` from `src/core/console-prefix.ts` so when `runEmbedCore` runs inside a per-source `withSourcePrefix` scope (installed by the `gbrain sync --all` worker pool) every line carries the `[] ` prefix; standalone callers see identical output because slog/serr fall through to bare console fns outside the wrap. Every embed-write path stamps `pages.embedding_signature` via `engine.setPageEmbeddingSignature(slug, {sourceId, signature: currentEmbeddingSignature()})` so a later model/dims swap is detectable as stale. The per-slug path (`embedPage`, used by `gbrain embed ` AND sync's post-import embed step) and the full-re-embed path (`embedAll`) stamp unconditionally per page. The stale path (`embedAllStale`) first calls `invalidateStaleSignatureEmbeddings` on a live run so signature-drifted pages flow through the NULL cursor, then stamps each page — but ONLY when EVERY chunk was stale this pass (a partially-stale page keeps preserved chunks of unknown provenance, so it stays unstamped rather than falsely marked current; `embed --all` fully re-embeds + stamps those). dry-run never mutates: it counts signature-drift via the widened `countStaleChunks({signature})` predicate without NULLing anything. - `src/core/conversation-parser/` — 17-pattern built-in chat-format registry + opt-in LLM polish/fallback. Modules: `types.ts` (PatternEntry + ParseResult + DateContext + CaptureMap + TimezonePolicy), `builtins.ts` (17 hand-vetted patterns sourced from public format docs — iMessage/Slack, Telegram bracket + text-export, bold-paren-time, bold-time-dash, bold-name-no-time, Discord classic + export, WhatsApp ISO + US, Signal, Matrix/Element, IRC classic + weechat, Teams export; module-load validation runs every `test_positive[]` + `test_negative[]` sample at startup so a typo in any built-in regex makes gbrain refuse to start; `DEFAULT_SPEAKER_CLEAN` exported as a module-level default), `parse.ts` (orchestrator with pattern-priority scoring across the first 10 lines + date derivation chain `explicit > frontmatter.date > effective_date > '1970-01-01'` + multi-line continuation + timezone warning), `llm-base.ts` (shared `runLlmCall` with content-hash cache in-process + DB-persistent via migration v97 + 4-strategy JSON repair + Anthropic-key probe; polish and fallback are thin wrappers), `llm-polish.ts` (opt-IN; headroom guard skips when tracker within $0.10 of cap; pure `applyPolish` for merge/drop/edit ops), `llm-fallback.ts` (opt-IN; NO regex inference + NO persistence), `eval.ts` (`scoreFixture` + `aggregateScores` + `parseFixtureJsonl` for the fixture-corpus CI gate), `nightly-probe.ts` (DI-stubbed; mode-gated default tokenmax=ON, conservative/balanced opt-in; adversarial false-positive detection). Pattern `bold-name-no-time` (regex `/^\*\*(?!\[)(.+?):\*\*\s*(.*)$/`, ordered after the time-bearing bold patterns) parses `**Speaker:** text` with NO per-line timestamp (Circleback/Granola/Zoom), anchoring every message at `T00:00:00Z` of the frontmatter date (line order preserves sequence, same no-time convention as `irc-classic`); the `(?!\[)` lookahead rejects telegram-bracket `**[18:37] Name:**`; non-shadow is the colon-INSIDE-bold regex (NOT declaration order — `parse.ts` scores every candidate independently, order is only the tie-break). Because `**Label:** text` is a common prose idiom, the pattern sets optional `PatternEntry.score_full_body: true` so `parse.ts` recomputes the winner's acceptance score over the FULL body before the `SCORING_MIN_ACCEPTANCE` floor, keeping a bold-label notes page at `no_match`. Pattern `bold-paren-time` parses `**Speaker** (HH:MM): text` and `(HH:MM:SS)` (date_source: frontmatter). Fallback gates: `SCORING_HEAD_TRIGGER_THRESHOLD = 0.3` triggers a full-body re-score when the head pass scores below that; `SCORING_MIN_ACCEPTANCE = 0.05` blocks essay false-positives. Exported `scorePatternFull(body, entry)`; private `getNonBlankLines(body, headCap?)` + `scoreFromLines(lines, entry)` DRY the quick_reject+regex loop. CLI surfaces at `src/commands/eval-conversation-parser.ts` (`gbrain eval conversation-parser ` exit 0/1/2, wired into `bun run verify` via `check:conversation-parser`) and `src/commands/conversation-parser.ts` (`scan ` debug, `list-builtins`, `validate `). Doctor checks: `conversation_format_coverage`, `progressive_batch_audit_health`, `conversation_parser_probe_health`. Pinned by `test/conversation-parser/{parse,llm-base,llm-fallback,llm-polish,nightly-probe}.test.ts` + the 27-case baseline at `test/extract-conversation-facts.test.ts` (back-compat invariant). Migration v97 (`conversation_parser_llm_cache_table`). Fixtures at `test/fixtures/conversation-formats/{imessage,imessage-time-only-12h,telegram-bracket,whatsapp-iso,whatsapp-us,signal-export,irc-classic,irc-weechat,matrix-element,teams-export,all,adversarial,bold-name-no-time,bold-time-dash}.jsonl` with `scripts/check-fixture-privacy.sh` banning real-name leaks. Maintainer guidance: [conversation parser patterns](conversation-parser-patterns.md). - `src/core/progressive-batch/` — shared ramp-up + cost-cap + verification primitive (trial 10 → ramp 100 → ramp 500 → full, with verification at each stage), with verifier+policy injection (callers describe HOW TO MEASURE SUCCESS, not WHEN TO WAIT FOR CTRL-C). Modules: `types.ts` (Stage, StageVerdict, AbortReason, discriminated `Verifier` union `OutputCountVerifier | IdempotentMutationVerifier | NoopVerifier`, Policy, StageReport), `orchestrator.ts` (`runProgressiveBatch(items, verifier, policy, runner)` — reads `getCurrentBudgetTracker()` ahead of `Policy.maxCostUsd` fail-closed; null both ways triggers `abort_cost_cap reason='no_budget_safety_net'`), `audit.ts` (ISO-week JSONL at `~/.gbrain/audit/progressive-batch-YYYY-Www.jsonl` via the shared `audit-writer` primitive), `stage-report.ts` (ASCII formatter for the default `Policy.onStageReport`). Env knobs: `GBRAIN_PROGRESSIVE_BATCH_DISABLED=1`, `GBRAIN_PROGRESSIVE_BATCH_AUTO=1` (skip Ctrl-C grace), `GBRAIN_PROGRESSIVE_BATCH_STAGES=10,100,500`. Sites that "jump straight to full" stay that way by default; ramp is opt-in per-site via `Policy.interactiveAbortMs > 0`. Pinned by `test/progressive-batch/orchestrator.test.ts` (35 cases, every verdict path). -- `src/commands/extract-conversation-facts.ts` + `src/core/cycle/conversation-facts-backfill.ts` — bulk fact extraction for long-form conversation pages. Walks conversation/meeting/slack/email/imessage/imessage-daily pages, splits into time-windowed segments (30-min gap or 30-msg cap), prepends a topical/temporal header, and runs through `extractFactsFromTurn()` so anchor-rich facts surface in `gbrain search`. Invariants: **strict per-source core** (`runExtractConversationFactsCore({sourceId, ...})` always takes one sourceId; CLI + cycle phase each do their own multi-source iteration because PHASE_SCOPE='source' is taxonomy-only); **two-phase enumeration** (paginated `listPages({type, sourceId, limit:10})`; per-page body cap MAX_PAGE_BODY_BYTES=25MB with `pages_skipped_too_large` counter surfaced in doctor); **page-global row_num accumulator** (facts unique index is `(source_id, source_markdown_slug, row_num)` per migration v51 — per-segment row_num would collide); **page-level TERMINAL audit row** to facts table after all segments commit (source='cli:extract-conversation-facts:terminal'; doctor's NOT EXISTS matches the terminal row so partial-extraction pages stay in backlog); **optional `opts.budgetTracker?`** (when present, used as-is — nested `withBudgetTracker` REPLACES; when absent, core auto-wraps with `BudgetTracker({maxCostUsd})`); **body read covers compiled_truth + timeline**; **honors `facts.extraction_enabled` kill-switch** with `--override-disabled` escape; **--types LIST allowlist** (`conversation,meeting,slack,email,imessage,imessage-daily`) with CLI default reading `cycle.conversation_facts_backfill.types`; **fingerprint on sourceId only**; **string-encoded op-checkpoint** entries `"||"` for resume (durable audit is the facts terminal row); **`--background` via maybeBackground** (Minion handler `extract-conversation-facts` re-creates BudgetTracker from `data.max_cost_usd`; on `BudgetExhausted` mid-job catches + persists + marks `completed` with `result.budget_exhausted=true`). The companion cycle phase `conversation_facts_backfill` (default OFF) iterates `listSources(engine)`, creates ONE brain-wide tracker per tick + wraps the loop in `withBudgetTracker` + passes the tracker into every per-source call. Two-layer cost AND walltime caps: per-source (`max_cost_usd=$1`, `max_walltime_min=20`) AND brain-wide (`max_total_cost_usd=$5`, `max_total_walltime_min=30`). Pinned by `test/extract-conversation-facts.test.ts` (27 cases). Migration v94 adds partial index `idx_facts_extract_conversation_session ON facts(source_id, source_session) WHERE source LIKE 'cli:extract-conversation-facts%'` (`transaction:false` + invalid-index pre-drop on Postgres; plain CREATE INDEX on PGLite). `src/commands/doctor.ts:computeConversationFactsBacklogCheck` is 3-state (SKIPPED when disabled; OK when caught up; WARN when >10 pages lack the terminal row, with paste-ready `gbrain doctor --remediate` step). `src/commands/sources.ts:runAudit` adds `facts_backfill_estimate: {pages, est_segments, est_cost_usd, types}`. Schema-pack `gbrain-base.yaml` promotes `conversation` (temporal, extractable) + `atom` (annotation, NOT extractable) into the base seed; backstop uses hardcoded `ELIGIBLE_TYPES` in `src/core/facts/eligibility.ts:51` not pack extractable. `ALL_PAGE_TYPES` in `src/core/types.ts` extended with the two new types. +- `src/commands/extract-conversation-facts.ts` + `src/core/cycle/conversation-facts-backfill.ts` — bulk fact extraction for long-form conversation pages. Walks conversation/meeting/slack/email/imessage/imessage-daily pages, splits them into time-windowed segments (30-min gap or 30-msg cap), prepends a topical/temporal header, and uses the strict `extractFactsFromTurnWithOutcome()` path so provider and output failures remain retryable instead of becoming successful empty pages. Invariants: **strict per-source core** (`runExtractConversationFactsCore({sourceId, ...})` always takes one sourceId; CLI + cycle phase each do their own multi-source iteration because `PHASE_SCOPE='source'` is taxonomy-only); **bounded two-phase enumeration** (paginated `listPages({type, sourceId, limit:10})`; per-page body cap `MAX_PAGE_BODY_BYTES=25MB`); **page-global `row_num` accumulator** (the facts unique index is `(source_id, source_markdown_slug, row_num)`); **versioned snapshot-bound outcomes** (`cli:extract-conversation-facts:terminal:v2` for complete pages and a separate `non-extractable:v2` source for recognized pages with no eligible segment); **operation checkpoints are scheduling hints only** and never suppress a replay without a matching v2 outcome; **optional `opts.budgetTracker?`** is used as-is, while an absent tracker is created with `maxCostUsd`; **body reads cover compiled truth, timeline, and configured raw-transcript sidecars**; **`facts.extraction_enabled` kill-switch** with `--override-disabled`; **`--types LIST` allowlist** (`conversation,meeting,slack,email,imessage,imessage-daily`); and **`--background` via `maybeBackground`**. The companion `conversation_facts_backfill` cycle phase is default-off, iterates every source, and enforces per-source plus brain-wide cost and wall-time caps. Migration v94 provides the partial facts index used by outcome lookups. `computeConversationFactsBacklogCheck` reports fresh completed, scanned-not-extractable, and unfinished counts separately, warning when more than 10 eligible pages lack a fresh v2 outcome. `sources audit` exposes `facts_backfill_estimate: {pages, est_segments, est_cost_usd, types}`. Pinned by `test/extract-conversation-facts.test.ts` and `test/doctor-conversation-facts-backlog.test.ts`. - `src/core/link-extraction.ts` — shared library for the graph layer. `extractEntityRefs` (canonical) matches `[Name](people/slug)` markdown links and Obsidian `[[people/slug|Name]]` wikilinks; `extractPageLinks`, `inferLinkType` heuristics (attended/works_at/invested_in/founded/advises/source/mentions), `parseTimelineEntries`, `isAutoLinkEnabled`. `DIR_PATTERN` covers `people`, `companies`, `deals`, `topics`, `concepts`, `projects`, `entities`, `tech`, `finance`, `personal`, `openclaw`. Used by extract.ts, operations.ts auto-link post-hook, and backlinks.ts. **Opt-in global-basename wikilink resolution** (issue #972, default off): `WIKILINK_GENERIC_RE` catches bare `[[name]]` wikilinks outside `DIR_PATTERN` (third pass `2c` in `extractEntityRefs`); `EntityRef.needsResolution: true` tags refs from this pass (the ref's `slug` is the wikilink TARGET, `name` the optional display alias). `SlugResolver` gains optional `resolveBasenameMatches(name): Promise` (multi-match by design — emits one edge per matching page). The single shared basename matcher is `buildBasenameIndex(slugs)` + `queryBasenameIndex(index, name)` + `normalizeBasename` (keys raw/lower/slugified tail, stable-sorted shorter-first then lexical), used by `makeResolver`, the FS `resolveBasenameMatchesFromSlugs`, AND the doctor check so they cannot drift. `makeResolver(engine, {mode, sourceId})` builds the index lazily via `engine.getAllSlugs({sourceId})` — source-scoped so a bare `[[name]]` never resolves to a same-tail page in a different source. `extractPageLinks` gains `opts.globalBasename` (routes `needsResolution` refs through `resolveBasenameMatches` keyed on `ref.slug`, emits candidates tagged `linkType: 'wikilink_basename'` + `linkSource: 'wikilink-resolved'`, skips self-loops) and `opts.skipFrontmatter` (replaces the old `nullResolver` ternary). All three surfaces (FS extract, DB extract, `put_page` auto-link) tag provenance with `link_source='wikilink-resolved'`; `put_page` includes it in its reconcilable-edge set so stale basename edges are removed when the wikilink or the flag goes away. Exports `WIKILINK_BASENAME_LINK_TYPE` + `isGlobalBasenameEnabled(engine)` (resolution order: env `GBRAIN_LINK_RESOLUTION_GLOBAL_BASENAME` → DB config `link_resolution.global_basename` → default false). `gbrain doctor`'s `link_resolution_opportunity` check surfaces a paste-ready enable hint when ≥5 bare wikilinks would resolve AND ≥20% match. Migration v113 widens `links_link_source_check` to admit `'wikilink-resolved'`; v114 (#1941) then opens it to any kebab-case provenance (`^[a-z][a-z0-9]*(-[a-z0-9]+)*$`, ≤64 chars) so external derivers register their own tag (e.g. `citation-graph`) without a migration. `LINK_EXTRACTOR_VERSION_TS` also lives here (bump like `CHUNKER_VERSION` to invalidate prior extract-stale stamps). Pinned by `test/link-extraction.test.ts`, `test/extract-fs.test.ts`, `test/doctor.test.ts`, `test/e2e/global-basename-pglite.test.ts`. - `src/commands/extract.ts` — `gbrain extract links|timeline|all [--source fs|db] [--source-id ]`: batch link/timeline extraction. fs walks markdown files, db walks pages from the engine (mutation-immune snapshot iteration; use for live brains with no local checkout). No in-memory dedup pre-load — candidates buffered 100 at a time and flushed via `addLinksBatch` / `addTimelineEntriesBatch`; `ON CONFLICT DO NOTHING` enforces uniqueness at the DB layer, `created` counter returns real rows inserted. `ExtractOpts.slugs?: string[]` enables incremental extract via `extractForSlugs()` (single combined links+timeline pass); the cycle path threads sync's `pagesAffected` through. `walkMarkdownFiles(brainDir)` still runs to build `allSlugs` for link resolution. `--source-id ` scopes extraction to one source on federated brains (resolved via `resolveSourceWithTier()` before any SQL; failures hint `gbrain sources list`). `gbrain extract --stale [--source-id ] [--catch-up] [--dry-run] [--json]` branch (`extractStaleFromDB`) — incremental DB-source link+timeline sweep over pages whose `pages.links_extracted_at` watermark is stale. Stale predicate (shared by both engines + the doctor check): `links_extracted_at IS NULL OR links_extracted_at < LINK_EXTRACTOR_VERSION_TS::timestamptz OR updated_at > links_extracted_at` (the `updated_at` arm catches MCP `put_page` / `sync --no-extract` edited-since-extract). Three new `BrainEngine` methods (parity in postgres-engine.ts + pglite-engine.ts + bootstrap probes): `countStalePagesForExtraction(opts?)`, `listStalePagesForExtraction({batchSize, afterPageId?, sourceId?, versionTs?})` (returns page CONTENT to avoid N+1 `getPage`; `rowToStalePage` in utils.ts maps the row, `StalePageRow` in types.ts), `markPagesExtractedBatch(refs, defaultExtractedAt)` (3-array unnest `slug[],source_id[],ts[]`; each ref may carry its own `extractedAt`). `STALE_BATCH_SIZE` default 25 (`GBRAIN_EXTRACT_STALE_BATCH`; small because page bodies are unbounded — the LIMIT is the only fetch-time memory bound); `STALE_TIME_BUDGET_MS` 30min wall-clock (`--catch-up` removes the cap). Non-swallowing flush: link/timeline flush throws propagate and abort the batch; stamp LAST so a crash leaves pages unstamped and they re-extract idempotently (`addLinksBatch` ON CONFLICT DO NOTHING + timeline dedup). Race fix: `extractStaleFromDB` stamps with each row's READ `updated_at` (not `now()`), so a concurrent edit during the sweep keeps the page stale and it re-extracts next run rather than marked fresh-with-old-content. Source-correct stamping at DB-extract sites via `stampExtracted` (best-effort, never throws); `extractLinksFromDB` only stamps the combined watermark when `subcommand === 'all'` (a links-only run must not hide timeline staleness). `LINK_EXTRACTOR_VERSION_TS` lives in `src/core/link-extraction.ts` (bump like `CHUNKER_VERSION` to invalidate all prior stamps). Migration v112 (`pages_links_extracted_at`) adds nullable `TIMESTAMPTZ` + composite `(source_id, links_extracted_at)` index (CONCURRENTLY + invalid-remnant pre-drop on Postgres, plain on PGLite), NO backfill so the real backlog surfaces on first `gbrain doctor`. Schema parity in schema.sql + pglite-schema.ts + schema-embedded.ts + `REQUIRED_BOOTSTRAP_COVERAGE`. `src/commands/doctor.ts:checkLinksExtractionLag` (the `links_extraction_lag` check, also in `doctorReportRemote`) warn-only by default (>`GBRAIN_EXTRACTION_LAG_WARN_PCT`, default 20%; shared `EXTRACTION_LAG_WARN_PCT_DEFAULT` + `EXTRACTION_LAG_MIN_PAGES=100` + exported `_resolveEnvNumber`), hard-fails only when `GBRAIN_EXTRACTION_LAG_FAIL_PCT` is set; vacuous-skips <100 pages (no `--source`); pre-v112 brains graceful-skip via `isUndefinedColumnError`; strictly a SQL COUNT (safe on remote/thin-client). `src/commands/sync.ts` gains `--no-extract` (threaded through single-source + `--all` + `syncOneSource`), stamps `links_extracted_at` for `pagesAffected` at the inline-extract call site, and `maybeExtractionNudge` prints a one-line stderr nudge after a `synced | first_sync | up_to_date` sync that leaves a backlog (`shouldNudgeAfterSync` pure predicate; `GBRAIN_SYNC_NO_EXTRACT_NUDGE` suppresses). `src/core/retry.ts` adds `'extract.stale'` to `BATCH_AUDIT_SITES`; `src/core/doctor-categories.ts` adds `links_extraction_lag` to `BRAIN_CHECK_NAMES`. Pinned by `test/extract-stale.test.ts` (incl. edited-after-stamp regression + crash-contract), `test/sync-inline-extract-stamps.serial.test.ts`, `test/sync-nudge-status-gate.test.ts`, `test/doctor-links-extraction-lag.test.ts`, engine-parity (Postgres↔PGLite) for the 3 methods + v112 round-trip. The stale SELECT in both engines projects a deterministic full-µs UTC string `to_char(updated_at AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"') AS updated_at_iso` (carried on `StalePageRow.updated_at_iso`, populated by `rowToStalePage` in utils.ts with an ISO-only fallback — never `String(Date)`, which `::timestamptz` misparses); `extractStaleFromDB` stamps that exact-precision value, not a JS `Date` (which truncates to milliseconds), so on Postgres `links_extracted_at` equals the row's `updated_at` to the microsecond and `links_extraction_lag` clears — a ms-truncated stamp stays strictly below the µs `updated_at` and leaves every page perpetually stale, which `extract --stale` could never satisfy. `to_char` (not raw `::text`, which is `DateStyle`-fragile) keeps the projection deterministic. The `markPagesExtractedBatch` SQL is unchanged, so callers passing an explicit (e.g. backdated) `extractedAt` still control the stamp and the edited-since arm is exact. A deterministic PGLite regression in `test/extract-stale.test.ts` injects a µs `updated_at`, runs `--stale`, and asserts the lag is 0 and stays 0. - `src/core/extract/receipt-writer.ts` + `src/core/extract/rollup-writer.ts` + `src/commands/extract-status.ts` + `src/commands/extract-explain.ts` + `src/commands/extract-benchmark.ts` + `src/core/schema-pack/scaffold-extractable.ts` — unified extract operator surface. Every shipped extractor (deterministic `facts.conversation` in `src/commands/extract-conversation-facts.ts` + three LLM-backed cycle phases at `src/core/cycle/{extract-atoms,synthesize-concepts,propose-takes,extract-facts}.ts`) writes ONE receipt page per run (`writeReceipt`) + UPSERTs a row to `extract_rollup_7d` (`upsertExtractRollup`). Receipt slug `extracts/{date}/{kind}/{source_id}/{run_id_short}/round-{N}.md`; frontmatter stamps BOTH `type: extract_receipt` AND `dream_generated: true` (belt+suspenders against extraction-loop guard drift). `extract_receipt` joins `ALL_PAGE_TYPES` in `src/core/types.ts`; `extracts/` prefix gets a 0.3x source-boost demote in `src/core/search/source-boost.ts`. Migration v104 adds `extract_rollup_7d (kind, source_id, day, cost_usd, halt_count, eval_pass_count, eval_fail_count, round_completed_count, rollup_write_failures, updated_at)` with PK `(kind, source_id, day)` + `idx_extract_rollup_7d_day`. Rollup writes best-effort with process-scoped error-dedup so transient DB failures bump `rollup_write_failures` instead of crashing the cycle. `extract_health` doctor check reads last 7 days, warns at halt-rate > 10% AND when rollup_write_failures > 0; pre-v104 brains report `ok`. CLI: `gbrain extract status [--source-id ID] [--kind X] [--verbose] [--json]` (7-day rollup, sorted halt_rate desc + cost desc, top-5 + "more rows" hint, stable `schema_version: 1`); `gbrain extract --explain ` (resolution chain pack-declared vs built-in cycle phase, prompt_template + fixture_corpus paths with `✓`/`(missing)`, last 7d rollup); `gbrain extract benchmark --pack X --kind Y` (loads pack fixture corpus through strict path validation — rejects absolute paths, `..` traversal, null bytes, AND symlinks resolving outside pack root; ships as a stub-reporter). `src/core/schema-pack/manifest-v1.ts` widens `extractable` from `z.boolean()` to `z.union([z.boolean(), ExtractableSpecSchema])` (carries `prompt_template`, `fixture_corpus`, `eval_dimensions`, `benchmark_min_recall`, plus reserved `verifier_path` — parses but refuses at runtime); `extractableSpecsFromPack` + `getExtractableSpec` + `refuseVerifierPathInV042` in `src/core/schema-pack/extractable.ts`; `gbrain schema scaffold-extractable --pack ` declares the type extractable, generates 5 placeholder fixtures + a prompt template stub under `packs//{fixtures,prompts}/extract/`, refuses to overwrite without `--force`. Pinned by `test/extractable-spec-widening.test.ts` (22), `test/extract/receipt-writer.test.ts` (12, canonical PGLite block R3+R4), `test/extract/benchmark.test.ts` (17), `test/extract/status.test.ts` (15), `test/schema-pack/scaffold-extractable.test.ts` (15, privacy guards), `test/doctor-extract-health.test.ts` (8). @@ -305,6 +305,7 @@ per-release `**vX.Y.Z:**` narration — CI enforces this - `src/core/worker-pool.ts` — Canonical sliding-pool + bounded-semaphore primitive (extracted from `src/commands/embed.ts` sliding-pool sites and `src/commands/eval-cross-modal.ts` `runWithLimit` semaphore). Two exports: `runSlidingPool({items, workers, onItem, signal?, onError?, failureLabel?, onProgress?})` + `runWithLimit({items, limit, fn, signal?})`. Atomicity invariant: `const idx = nextIdx++` is one synchronous JS statement (no `await` between read and write — guaranteed by the single-threaded event loop), documented in the module header AND enforced by `scripts/check-worker-pool-atomicity.sh` (wired into `bun run verify`), which rejects importing `worker_threads` in any consuming file and inserting `await` between the `nextIdx` read and write. `MUST_ABORT_ERROR_TAGS` set is seeded with `BUDGET_EXHAUSTED` from `src/core/budget/budget-tracker.ts`; tagged errors (matched via `err.tag === 'BUDGET_EXHAUSTED'` to avoid cross-module import) bypass `onError` and hard-abort the pool via `AbortController.abort()` to in-flight `onItem` — the budget cap is a structural ceiling under concurrency. `failures[]` shape is `{idx, label, error}` records (NOT full items; callers supply `failureLabel(item) => string`) for bounded memory under huge brains. Pinned by `test/worker-pool.test.ts` + `test/scripts/check-worker-pool-atomicity.test.ts`. Drives every `--workers N` bulk command. - `src/commands/embed.ts` extension — both inline sliding-pool sites (`embedAll` simple at `:458-467` and `embedAllStale` paginated + AbortSignal at `:586-632`) call `runSlidingPool` from the shared worker-pool helper. Invariant-level contract preserved (counts + cost + AbortSignal propagation + per-batch rate-limit retry via `embedBatchWithBackoff`); byte-equality on progress-event ORDERING is NOT promised. The `GBRAIN_EMBED_CONCURRENCY || 20` default is preserved and embed bypasses `resolveWorkersWithClamp` because the 20-worker default would otherwise silently change every brain's embed hot path. Pinned by `test/embed-helper-migration.test.ts` (asserts the helper is wired in AND the pre-migration `let nextIdx = 0` + `Promise.all(Array.from({length: numWorkers}, ...))` shapes are gone). - `src/commands/extract-conversation-facts.ts` extension — `--workers N` for LLM-bound fact extraction over conversation pages, with a per-page advisory lock via `src/core/db-lock.ts:withRefreshingLock` (lock id `extract-conversation-facts::`, TTL `PER_PAGE_LOCK_TTL_MINUTES=2` with 20s refresh via `Math.max(15s, 120s/6)`; `LockUnavailableError` triggers skip-and-continue with rate-limited log per (source, minute) + `pages_lock_skipped` counter + CLI exits 3 when non-zero AND no hard failures). `deleteOrphanFactsForPage(engine, sourceId, slug)` provides delete-orphans-first replay safety — wipes facts from a prior crashed run for this (sourceId, slug) before re-extracting, closing the "terminal audit row written after partial insertFacts failure" class. `assertFactsEmbeddingDimMatchesConfig(engine)` is the startup preflight (throws `FactsEmbeddingDimMismatchError` with paste-ready ALTER hint BEFORE the first insert; cached per engine via WeakMap). Result type carries `pages_lock_skipped` + `orphan_facts_cleaned`. Checkpoint state is a shared `cpMap: Map` (NOT a per-page-mutated `cpEntries: string[]`) so atomic `Map.set` survives parallel workers. Minion handler `extract-conversation-facts` in `src/commands/jobs.ts` round-trips `workers` via `job.data.workers` for `--background --workers 20`. Cycle config key `cycle.conversation_facts_backfill.workers` (default 1; opt-in concurrency under brain-wide cost + walltime caps). Pinned by `test/extract-conversation-facts-workers.test.ts` + the existing extract-conversation-facts behavioral tests. +- `src/commands/extract-conversation-facts.ts` + `src/commands/doctor.ts` durable outcome authority — page completion survives operation-checkpoint GC through versioned terminal audit rows (`cli:extract-conversation-facts:terminal:v2`), while recognized pages with no eligible segment use the separate `cli:extract-conversation-facts:non-extractable:v2` source. Each outcome is bound to the exact parsed snapshot: regular pages use `content_hash` plus the UTC effective date; raw-conversation sidecars and legacy null-hash pages use a canonical SHA-256 over every parser-relevant input. Selection checks the token before locking, refetches under the lock, and verifies it again before writing the outcome, so an edit cannot be certified by stale work. The strict extraction path treats provider, refusal, truncation, malformed/schema-invalid output, segment-write, cleanup, and terminal-write failures as unfinished work; bulk failures increment `pages_failed`, affect CLI/cycle receipts and exit status, and never advance the legacy checkpoint. Checkpoints are only a scheduling hint: a slug without a matching v2 outcome is replayed delete-first. `no_match`, errors, cancellation, and dry runs never become durable negatives. Result, CLI, cycle, and doctor surfaces keep completed, scanned-not-extractable, unfinished, failed, and lock-skipped counts separate. See [Conversation backfill durable outcomes](../operations/conversation-backfill-outcomes.md) for the operator and maintainer contract. Pinned by `test/extract-conversation-facts.test.ts` and `test/doctor-conversation-facts-backlog.test.ts`. - `src/core/embedding-dim-check.ts` extension — facts.embedding dim drift surface. `readFactsEmbeddingDim(engine): Promise` covers both `vector(N)` and `halfvec(N)` shapes (migration v40 falls back to `vector` on pgvector < 0.7); regex ordering is halfvec-before-vector (substring "vec" appears in "halfvec"; naive `/vector/i` would shadow). `buildFactsAlterRecipe(dims, configured, type)` emits the paste-ready `DROP INDEX IF EXISTS idx_facts_embedding_hnsw; ALTER TABLE facts ALTER COLUMN embedding TYPE halfvec(N) USING embedding::halfvec(N); CREATE INDEX idx_facts_embedding_hnsw ON facts USING hnsw (embedding halfvec_cosine_ops) WHERE ...` flow (NOT bare REINDEX, which doesn't rewrite the index after a column-type change). `assertFactsEmbeddingDimMatchesConfig(engine)` is the preflight — throws `FactsEmbeddingDimMismatchError` (tagged `tag: 'FACTS_EMBEDDING_DIM_MISMATCH'` for parity with the worker-pool MUST_ABORT semantics) when configured dim ≠ column width; cached per-engine via `WeakMap`; PGLite engines silently skip. Doctor check `facts_embedding_width_consistency` (registered after `embedding_width_consistency`) reuses the same helpers with an identical ALTER recipe. Pinned by `test/embedding-dim-check-facts.test.ts`. - `src/core/postgres-engine.ts` extension — `insertFact` + `insertFacts` no longer hardcode `tx.unsafe(\`'${embedLit}'::vector\`)` for the embedding column. `resolveFactsEmbeddingCast()` (private) probes `pg_attribute` once per engine instance (cached in `_factsEmbeddingCastSuffix`) and returns `'::halfvec'` when migration v40 created the column as halfvec, else `'::vector'`; both insert paths use the cached suffix so the cast matches the actual column type (works on older pgvector that lacks implicit auto-cast). Test seam `__resetFactsEmbeddingCastCacheForTest()` clears the per-engine cache. - `src/core/cycle.ts` + `src/core/cycle/extract-atoms.ts` + `src/core/cycle/synthesize-concepts.ts` + `src/commands/extract.ts` + `src/commands/doctor.ts` + `src/core/op-checkpoint.ts` extensions — six daily-driver ops fixes. (1) Batch idempotency: `atomsExistingForHashes(engine, sourceId, hashes[])` (exported from `src/core/cycle/extract-atoms.ts`) replaces the per-hash loop (7K individual queries) with one batched SQL roundtrip returning already-extracted `content_hash16` values; fail-open (SQL error → empty set, extraction proceeds); powered by migration v104 `pages_atom_source_hash_idx` (partial expression index on `frontmatter->>'source_hash'` for atom rows where `deleted_at IS NULL`; Postgres `CREATE INDEX CONCURRENTLY` with invalid-remnant pre-drop, PGLite plain). (2) Shorter cycle lock TTL + active in-phase refresh: `LOCK_TTL_MINUTES = 5` (was 30); `buildYieldDuringPhase(lock, outer)` (exported, with `LockHandle`) calls `lock.refresh()` + any external hook on every fire, throttled to 30s via `maybeYield`, firing both in the main loop AND immediately after every `await chat(...)`; `synthesize_concepts` uses the same throttled hook. A crashed cycle releases its lock 6x faster while a healthy long-running cycle keeps it alive (residual: a single `await chat()` past 5 min can expire the lock mid-await — TODO-OPS-2). (3) Progress wiring: `progress?: ProgressReporter` opt on `ExtractAtomsOpts` and `SynthesizeConceptsOpts`; cycle.ts passes its phase-level reporter down (NOT a child reporter, which would collide on `cycle.extract_atoms.extract_atoms.work`); phases only call `tick()`/`heartbeat()`, cycle.ts owns `start()`/`finish()`. (4) `by-mention` resume: `mentionsFingerprint({source, type, since, gazetteerHash})` in `src/core/op-checkpoint.ts` — the gazetteer hash is load-bearing (adding entity pages mid-pause shifts the hash → new fingerprint → fresh scan against the new gazetteer, never silent skip); `gbrain extract links --by-mention` resumes via `op_checkpoints` with `flushAndCheckpoint` ordering (links flush to DB FIRST, page keys commit to checkpoint SECOND, persist THIRD, so a crash mid-batch leaves the page un-checkpointed and resume re-scans it); persist every 1000 items OR 30s; clean exit clears the checkpoint; `--dry-run` skips both load and write. (5) `sync_consolidation` doctor check (multi-source brains see a paste-ready `gbrain sync --all --parallel 4 --workers 4 --skip-failed`; single-source "not applicable"; SQL errors return `warn` via the check's own try/catch). (6) Test-isolation: `test/cycle-last-full-cycle-at.test.ts` + `test/schema-cli.test.ts` use per-test `GBRAIN_HOME=tempdir`. Pinned by `test/cycle/extract-atoms-batch.test.ts`, `test/cycle/cycle-lock-ttl.test.ts` (regression pin on `LOCK_TTL_MINUTES === 5`), `test/op-checkpoint-mentions-fingerprint.test.ts`, `test/cycle/extract-atoms-progress.test.ts`, `test/cycle/synthesize-concepts-progress.test.ts`, `test/cycle/yield-during-phase-refresh.test.ts`, `test/cycle/yield-during-phase-throttle.test.ts`, `test/extract-by-mention-resume.test.ts`, `test/doctor-sync-consolidation.test.ts`. Companion `sync --all` recipe block in `skills/cron-scheduler/SKILL.md`. `synthesize_concepts` writes concept pages through `importFromContent` (#2163: the same parse→chunk→embed pipeline put_page uses, with put_page's `isAvailable('embedding')` → `noEmbed` gate) so `concepts/` pages carry `content_chunks` + embeddings and are reachable by retrieval (where `source-boost.ts` weights them 1.3×). diff --git a/docs/operations/conversation-backfill-outcomes.md b/docs/operations/conversation-backfill-outcomes.md new file mode 100644 index 000000000..3949395c8 --- /dev/null +++ b/docs/operations/conversation-backfill-outcomes.md @@ -0,0 +1,227 @@ +# Conversation backfill durable outcomes + +`gbrain extract-conversation-facts` stores page-level outcomes in `facts` so +bulk runs, autopilot, and `gbrain doctor` can distinguish finished work from +retryable work without adding another state table. + +This is completion authority, not ordinary extracted knowledge. The authority +is deliberately narrow: a marker is valid only for the exact page or transcript +snapshot that was parsed, and only after every required operation succeeded. + +## Outcome protocol + +The current protocol is v2. Its source names are versioned so rows written by +older best-effort implementations cannot suppress a corrective replay. + +| Outcome | `facts.source` | Meaning | +|---|---|---| +| Complete | `cli:extract-conversation-facts:terminal:v2` | Every eligible segment was extracted and inserted successfully, the input remained unchanged, and the terminal write succeeded. | +| Scanned, not extractable | `cli:extract-conversation-facts:non-extractable:v2` | A recognized input was scanned successfully but contained no eligible multi-message segment. | +| Unfinished | no matching v2 outcome | Work is pending, failed, was not recognized, changed during extraction, or has only a legacy marker. | + +The non-extractable outcome is intentionally separate from completion. It does +not claim that knowledge facts were extracted. CLI counters, cycle details, and +doctor output preserve that distinction. + +## Snapshot identity + +Every v2 marker binds `source_session` to the parser input snapshot: + +```text +:: +``` + +There are two token forms. + +### Database-backed page body + +For pages parsed from `compiled_truth` and `timeline`, the token is: + +```text +page-- +``` + +`content_hash` covers title, type, compiled truth, timeline, and frontmatter. +The effective-date suffix covers the remaining date input used by parsing. This +identity does not depend on JavaScript's millisecond timestamp precision, so two +writes within one PostgreSQL millisecond still produce different tokens when +parser input changes. A legacy page with a null content hash uses a computed +SHA-256 fallback and is verified in-process by both extraction and doctor. + +### Raw transcript sidecar + +When frontmatter contains `raw_transcript`, the source text lives outside the +page row and may change without changing `pages.updated_at`. Its token is: + +```text +sidecar- +``` + +The digest covers the exact body given to the parser plus parser-relevant page +metadata: title, type, frontmatter, and effective date. Selection recomputes +the digest before skipping work. A sidecar-only edit therefore reopens the page. + +`gbrain doctor` cannot read sidecars in its SQL aggregate, so it enumerates those +pages in bounded batches and calls the same canonical verifier used by +extraction. Doctor and extraction therefore agree after sidecar-only edits. + +## Selection and locking + +Bulk extraction follows this sequence: + +1. Enumerate candidate pages in bounded batches. +2. Filter candidates with matching v2 outcomes. +3. Apply `--limit` to the remaining pages that actually need work. +4. Acquire the source-and-slug advisory lock. +5. Re-fetch the page under that lock. +6. Recompute and recheck the snapshot-bound outcome. +7. Prepare one immutable parser snapshot and process it. +8. Re-fetch and recompute the snapshot before writing an outcome. + +The pre-lock check avoids parser, filesystem, and model work for ordinary +completed pages. The under-lock refetch prevents a stale enumeration object +from becoming the certified input. The final comparison prevents an edit that +happens during model or insertion work from receiving a marker for old content. + +An edit can occur after the final comparison and before marker insertion. That +is still safe because the marker contains the old version token. Future +selection compares the token, not marker creation time, and reopens the page. + +Single-page `--slug` runs use the same under-lock path. + +## Strict extraction success + +The general `extractFactsFromTurn` API remains best-effort for interactive +callers. It historically returns an empty array for both a legitimate zero-fact +answer and several model failures. + +Conversation backfill instead uses `extractFactsFromTurnWithOutcome`, whose +result separates: + +- `{ ok: true, facts: [] }`, a successful extraction with no durable facts; +- `{ ok: true, facts: [...] }`, a successful extraction with facts; and +- `{ ok: false, reason, error? }`, an unavailable provider, provider error, + refusal, content filter, malformed output, or repeated truncation. + +Any failed segment aborts the page attempt. Any `insertFacts` failure also +aborts it. The page receives neither a checkpoint advancement nor a terminal +outcome. Facts inserted by earlier segments may remain temporarily, but the +next claim deletes this command's rows for the page and replays cleanly. + +Bulk workers continue past an individual page failure, but they do not hide it. +`pages_failed` counts failed claims, stderr names each page, the CLI exits 1, +the autopilot phase reports `warn`, and receipts/rollups classify the run as +incomplete. A tolerant pool is therefore observable without sacrificing the +rest of a large backfill. + +This distinction is load-bearing. Treating a provider outage as a successful +zero-fact response would make a transient failure durable and permanently hide +the page from later runs. + +## Non-extractable authority + +A non-extractable marker is written only when all of the following are true: + +- a deterministic or accepted parser format recognized the input; +- ordinary segmentation produced no eligible multi-message segment; +- the parser phase was not `no_match`; +- cleanup of prior command-owned rows succeeded; and +- the input snapshot was still current immediately before cleanup and write. + +A `no_match` result stays unfinished so a new parser pattern, optional fallback, +or corrected input can recover it. Oversize pages, disappeared pages, lock +contention, dry runs, aborts, cleanup errors, provider failures, extraction +failures, insertion failures, and outcome-write failures also stay unfinished. + +Cleanup errors are never interpreted as "zero rows deleted." Propagating them +prevents a fresh non-extractable marker from coexisting with stale extracted +facts that could not be removed. + +## Checkpoints are not authority + +Operation checkpoints are only progress hints. They do not prove which page +snapshot was processed, and old checkpoint entries do not include a snapshot +token. When a page lacks a matching v2 outcome, the command discards that +page's checkpoint entry and performs a delete-first full replay. + +This rule prevents two corruption classes: + +- edited text with timestamps older than the old watermark being skipped; and +- command-owned facts being deleted while the checkpoint skips the segments + needed to recreate them. + +Deleting `op_checkpoints` does not reopen pages with matching v2 outcomes. +Deleting or editing an outcome does not make a checkpoint authoritative. + +## `--limit` semantics + +`--limit N` caps pages that require processing, not completed pages inspected +while finding them. Durable filtering happens before clipping a batch. With a +completed page first and a pending page second, `--limit 1` processes the +pending page rather than consuming the limit on the completed page. + +`pages_considered` may therefore exceed `--limit` because it includes durable +outcomes observed during selection. Model-bearing page work does not exceed the +limit. + +## `--force` + +`--force` bypasses durable outcome selection and clears the page checkpoint. +It still uses delete-first replay, strict extraction outcomes, advisory locks, +and snapshot verification. Force means "recompute" rather than "relax safety." + +## Operator signals + +The result exposes separate counters: + +- `pages_skipped_completed` +- `pages_skipped_non_extractable` +- `pages_marked_non_extractable` +- `pages_failed` + +The CLI aggregates these across sources. The autopilot backfill phase includes +them in phase details. `gbrain doctor` reports `completed`, +`scanned_not_extractable`, and `backlog` independently. + +Run a small canary twice: + +```bash +gbrain extract-conversation-facts --source-id default --limit 10 --workers 1 --max-cost-usd 0.25 --yes +gbrain extract-conversation-facts --source-id default --limit 10 --workers 1 --max-cost-usd 0.25 --yes +gbrain doctor +``` + +On the second run, unchanged pages should move through durable skip counters. +Edit one page or raw transcript sidecar and rerun; that page should process +again and receive a marker with a new token. + +## Maintainer contracts + +- Version completion protocols when their success guarantees change. +- Require an exact `source`, page slug, and snapshot-bound `source_session`. +- Keep completion and non-extractable as different sources and counters. +- Re-fetch after acquiring the lock; never certify the enumeration object. +- Revalidate the snapshot before writing either durable outcome. +- Keep sidecar content in the version identity. +- Keep regular-page content hash and effective date in the version identity. +- Never turn model, insertion, cleanup, cancellation, or parser failures into + successful empty extraction. +- Never classify `no_match` or dry-run output as a durable negative. +- Do not make operation checkpoints completion authority. +- Apply work limits after durable filtering. +- Keep doctor source-scoped by both page and fact `source_id`. +- Give terminal completion precedence if both current outcome rows exist. +- Update CLI and cycle aggregation whenever a result counter changes. + +## Focused verification + +```bash +bun test test/extract-conversation-facts.test.ts +bun test test/doctor-conversation-facts-backlog.test.ts +bun x tsc --noEmit +``` + +The focused suite covers checkpoint garbage collection, same-timestamp edits, +edits during extraction, sidecar-only edits, legacy marker replay, provider and +insert failures, cleanup failure, recognized non-extractable scans, retryable +parser misses, post-filter limits, force replay, and doctor accounting. diff --git a/src/commands/doctor.ts b/src/commands/doctor.ts index 87e2267c8..79cc55ec6 100644 --- a/src/commands/doctor.ts +++ b/src/commands/doctor.ts @@ -3331,18 +3331,10 @@ export function computeNightlyQualityProbeHealthCheck( * - OK when enabled=true AND backlog==0 OR no eligible pages exist. * - WARN when enabled=true AND backlog>10. * - * Backlog query uses the page-level TERMINAL audit row check (Eng-v2 - * C7), source-scoped via explicit predicate (Eng-v2 C2). Partial- - * extraction pages stay in backlog because the terminal row isn't - * written until ALL segments complete. - * - * Known approximation (documented in the details field): "complete" - * means "terminal row exists" which means "all segments completed in - * a prior run." A page with the terminal row from one run + new - * messages since shows OK until the next run picks up new messages - * and writes a fresh terminal row. The backlog is therefore an UPPER - * BOUND on "pages with NO extraction at all", not "pages whose facts - * are current." + * Backlog uses versioned, source-scoped outcomes. Regular pages bind the marker + * to pages.updated_at; raw-transcript sidecars carry a SHA-256 snapshot token + * and are revalidated by the extraction command before it skips model work. + * Legacy/unversioned rows and partial extraction remain in backlog. */ export async function computeConversationFactsBacklogCheck( engine: BrainEngine, @@ -3384,35 +3376,112 @@ export async function computeConversationFactsBacklogCheck( } } - // Source-scoped NOT EXISTS (Eng-v2 C2 + C7): - // - facts.source matches TERMINAL audit source - // - source_session matches terminal: - // - source_id matches page's source_id (cross-source safety) - const rows = await engine.executeRaw<{ count: string | number }>( - `SELECT COUNT(*) AS count FROM pages p - WHERE p.type = ANY($1::text[]) - AND p.deleted_at IS NULL - AND NOT EXISTS ( - SELECT 1 FROM facts f - WHERE f.source = 'cli:extract-conversation-facts:terminal' - AND f.source_session = 'cli:extract-conversation-facts:terminal:' || p.slug - AND f.source_id = p.source_id - )`, + const rows = await engine.executeRaw<{ + backlog: string | number; + completed: string | number; + non_extractable: string | number; + }>( + `WITH outcomes AS ( + SELECT + p.source_id, + p.slug, + MAX(CASE WHEN f.source = 'cli:extract-conversation-facts:terminal:v2' THEN 1 ELSE 0 END) AS completed, + MAX(CASE WHEN f.source = 'cli:extract-conversation-facts:non-extractable:v2' THEN 1 ELSE 0 END) AS non_extractable + FROM pages p + LEFT JOIN facts f + ON f.source_id = p.source_id + AND f.source_markdown_slug = p.slug + AND f.source IN ( + 'cli:extract-conversation-facts:terminal:v2', + 'cli:extract-conversation-facts:non-extractable:v2' + ) + AND p.content_hash IS NOT NULL + AND f.source_session = f.source || ':' || p.slug || ':page-' || + p.content_hash || '-' || + COALESCE(TO_CHAR(p.effective_date AT TIME ZONE 'UTC', 'YYYY-MM-DD'), 'none') + WHERE p.type = ANY($1::text[]) + AND p.deleted_at IS NULL + AND COALESCE(BTRIM(p.frontmatter->>'raw_transcript'), '') = '' + AND p.content_hash IS NOT NULL + GROUP BY p.source_id, p.slug + ) + SELECT + COALESCE(SUM(CASE WHEN completed = 0 AND non_extractable = 0 THEN 1 ELSE 0 END), 0) AS backlog, + COALESCE(SUM(completed), 0) AS completed, + COALESCE(SUM(CASE WHEN completed = 0 THEN non_extractable ELSE 0 END), 0) AS non_extractable + FROM outcomes`, [types], ); - const backlog = Number(rows[0]?.count ?? 0); + let backlog = Number(rows[0]?.backlog ?? 0); + let completed = Number(rows[0]?.completed ?? 0); + let nonExtractable = Number(rows[0]?.non_extractable ?? 0); + + // SQL cannot read raw_transcript files or reproduce the fallback hash for a + // legacy NULL content_hash. Recompute those tokens through the command's + // canonical verifier. Pagination keeps memory bounded. + const { findFreshExtractionOutcomes } = await import( + './extract-conversation-facts.ts' + ); + const verifierSources = await engine.executeRaw<{ source_id: string }>( + `SELECT DISTINCT source_id + FROM pages + WHERE type = ANY($1::text[]) + AND deleted_at IS NULL + AND ( + COALESCE(BTRIM(frontmatter->>'raw_transcript'), '') <> '' + OR content_hash IS NULL + ) + ORDER BY source_id`, + [types], + ); + for (const { source_id: sourceId } of verifierSources) { + for (const type of types) { + let offset = 0; + // eslint-disable-next-line no-constant-condition + while (true) { + const batch = await engine.listPages({ + type: type as NonNullable[0]>['type'], + sourceId, + limit: 10, + offset, + }); + if (batch.length === 0) break; + const verifyInProcess = batch.filter((page) => { + const raw = page.frontmatter?.raw_transcript; + return (typeof raw === 'string' && raw.trim().length > 0) || + page.content_hash == null; + }); + if (verifyInProcess.length > 0) { + const outcomes = await findFreshExtractionOutcomes( + engine, + sourceId, + verifyInProcess, + ); + for (const page of verifyInProcess) { + const outcome = outcomes.get(page.slug); + if (outcome === 'complete') completed++; + else if (outcome === 'non_extractable') nonExtractable++; + else backlog++; + } + } + offset += batch.length; + if (batch.length < 10) break; + } + } + } if (backlog === 0) { return { name, status: 'ok', - message: 'all eligible pages have extraction terminal audit rows', + message: 'all eligible pages have fresh durable extraction outcomes', details: { backlog, + completed, + scanned_not_extractable: nonExtractable, types, - known_approximation: - 'backlog counts pages with NO extraction terminal row; pages with new messages since prior extraction may show OK until next run', + freshness_rule: 'v2 snapshot token (content hash + effective date or sidecar sha256)', }, }; } @@ -3426,10 +3495,11 @@ export async function computeConversationFactsBacklogCheck( message: `${backlog} eligible pages without extraction. Fix: ${fixHint}`, details: { backlog, + completed, + scanned_not_extractable: nonExtractable, types, fix_hint: fixHint, - known_approximation: - 'backlog counts pages with NO extraction terminal row; pages with new messages since prior extraction may show OK until next run', + freshness_rule: 'v2 snapshot token (content hash + effective date or sidecar sha256)', }, }; } @@ -3438,7 +3508,13 @@ export async function computeConversationFactsBacklogCheck( name, status: 'ok', message: `${backlog} eligible page(s) below warn threshold (>10)`, - details: { backlog, types }, + details: { + backlog, + completed, + scanned_not_extractable: nonExtractable, + types, + freshness_rule: 'v2 snapshot token (content hash + effective date or sidecar sha256)', + }, }; } catch (err) { return { diff --git a/src/commands/extract-conversation-facts.ts b/src/commands/extract-conversation-facts.ts index 6c0e74e2f..9ee54636d 100644 --- a/src/commands/extract-conversation-facts.ts +++ b/src/commands/extract-conversation-facts.ts @@ -43,11 +43,10 @@ * (source_id, source_markdown_slug, row_num); per-segment row_num * would collide on segment 2. Per-page counter increments across * segments. - * - Terminal audit row on completion. After all segments commit, one - * extra fact row with source='cli:extract-conversation-facts:terminal' - * marks the page complete. Doctor's backlog query checks for the - * terminal row, NOT any fact — partial extraction → no terminal → - * next run resumes. + * - Snapshot-bound terminal audit row on completion. After all segments + * commit, one v2 row binds completion to the exact page version or raw + * transcript digest. Partial extraction has no matching terminal and the + * next claim performs a delete-first full replay. * - Optional budgetTracker via opts. If a tracker is in opts, use it * as-is (NO `withBudgetTracker` wrap, which would REPLACE the active * tracker per gateway.ts AsyncLocalStorage semantics, defeating an @@ -68,7 +67,7 @@ import type { BrainEngine, NewFact } from '../core/engine.ts'; import type { Page } from '../core/types.ts'; import { - extractFactsFromTurn, + extractFactsFromTurnWithOutcome, isFactsExtractionEnabled, } from '../core/facts/extract.ts'; import { configureGatewayIfUninitialized, isAvailable, withBudgetTracker } from '../core/ai/gateway.ts'; @@ -172,7 +171,15 @@ export const PER_SEGMENT_SOURCE_PREFIX = 'cli:extract-conversation-facts'; * the per-segment source. Partial extraction = no terminal row = page * stays in backlog. */ -export const TERMINAL_AUDIT_SOURCE = 'cli:extract-conversation-facts:terminal'; +export const TERMINAL_AUDIT_SOURCE = 'cli:extract-conversation-facts:terminal:v2'; + +/** + * Durable outcome for a successfully scanned page that contains no eligible + * multi-message segment. Kept distinct from successful extraction so operator + * surfaces can report the truth without rescanning the page forever. + */ +export const NON_EXTRACTABLE_AUDIT_SOURCE = + 'cli:extract-conversation-facts:non-extractable:v2'; // --------------------------------------------------------------------------- // Public types. @@ -253,6 +260,14 @@ export interface ExtractConversationFactsResult { pages_skipped: number; pages_skipped_too_large: number; pages_skipped_disappeared: number; + /** Fresh terminal outcomes skipped before parsing or model work. */ + pages_skipped_completed: number; + /** Fresh scanned-not-extractable outcomes skipped before parser work. */ + pages_skipped_non_extractable: number; + /** Durable scanned-not-extractable outcomes written by this run. */ + pages_marked_non_extractable: number; + /** Pages whose claim reached extraction but failed before durable outcome. */ + pages_failed: number; /** * Pages whose built-in parse returned `no_match` and whose messages were * recovered by the explicitly enabled LLM fallback. @@ -591,31 +606,21 @@ async function deleteOrphanFactsForPage( sourceId: string, slug: string, ): Promise { - try { - // The two write-source variants this command may have left behind: - // - PER_SEGMENT_SOURCE_PREFIX ('cli:extract-conversation-facts') - // - TERMINAL_AUDIT_SOURCE ('cli:extract-conversation-facts:terminal') - // Using a LIKE prefix match covers both with one statement. - const rows = await engine.executeRaw<{ count: string }>( - `WITH del AS ( - DELETE FROM facts - WHERE source_id = $1 - AND source_markdown_slug = $2 - AND source LIKE 'cli:extract-conversation-facts%' - RETURNING 1 - ) - SELECT COUNT(*)::text AS count FROM del`, - [sourceId, slug], - ); - const n = parseInt(rows[0]?.count ?? '0', 10); - return Number.isFinite(n) ? n : 0; - } catch { - // Best-effort: a missing source_markdown_slug column on pre-v0.32 - // brains (or other rare DDL drift) falls through to "no orphans - // cleaned." The subsequent insertFacts call will surface any real - // schema issues with a clearer error. - return 0; - } + // A cleanup failure is authoritative: callers must not write a terminal or + // non-extractable marker while facts from an older snapshot may remain. + const rows = await engine.executeRaw<{ count: string }>( + `WITH del AS ( + DELETE FROM facts + WHERE source_id = $1 + AND source_markdown_slug = $2 + AND source LIKE 'cli:extract-conversation-facts%' + RETURNING 1 + ) + SELECT COUNT(*)::text AS count FROM del`, + [sourceId, slug], + ); + const n = parseInt(rows[0]?.count ?? '0', 10); + return Number.isFinite(n) ? n : 0; } // --------------------------------------------------------------------------- @@ -677,11 +682,150 @@ function cpEntriesToMap(entries: string[]): Map { return map; } +export type DurableExtractionOutcome = 'complete' | 'non_extractable'; + +interface ConversationPageSnapshot { + page: Page; + body: string; + versionToken: string; +} + +function hasRawTranscriptSidecar(page: Page): boolean { + const raw = page.frontmatter?.raw_transcript; + return typeof raw === 'string' && raw.trim().length > 0; +} + +function regularPageVersionToken(page: Page): string { + // content_hash covers title, type, compiled_truth, timeline, and frontmatter. + // Unlike JavaScript Date, it cannot collapse distinct PostgreSQL updates that + // happen within the same millisecond. effective_date is parser input too. + const hash = page.content_hash ?? createHash('sha256') + .update(JSON.stringify({ + title: page.title, + type: page.type, + compiled_truth: page.compiled_truth, + timeline: page.timeline || '', + frontmatter: page.frontmatter || {}, + })) + .digest('hex'); + const effectiveDate = page.effective_date + ? new Date(page.effective_date).toISOString().slice(0, 10) + : 'none'; + return `page-${hash}-${effectiveDate}`; +} + +function snapshotVersionToken(page: Page, body: string): string { + if (!hasRawTranscriptSidecar(page)) return regularPageVersionToken(page); + // Sidecar contents can change without touching pages.updated_at. Hash the + // exact parser input plus parser-relevant page metadata so those edits reopen + // the page without a schema migration. + return `sidecar-${createHash('sha256') + .update( + JSON.stringify({ + body, + title: page.title, + type: page.type, + frontmatter: page.frontmatter, + effective_date: page.effective_date ?? null, + }), + ) + .digest('hex')}`; +} + +async function preparePageSnapshot( + engine: BrainEngine, + page: Page, +): Promise { + const body = await readConversationBodyForParsing(engine, page); + return { page, body, versionToken: snapshotVersionToken(page, body) }; +} + +function outcomeSession(source: string, slug: string, versionToken: string): string { + return `${source}:${slug}:${versionToken}`; +} + +/** + * Find v2 outcomes bound to the exact parser input snapshot. Legacy outcome + * rows deliberately do not match and are replayed once under the strict v2 + * protocol. Sidecar files are hashed because pages.updated_at cannot see them. + */ +export async function findFreshExtractionOutcomes( + engine: BrainEngine, + sourceId: string, + pages: readonly Page[], +): Promise> { + if (pages.length === 0) return new Map(); + const expected = new Map(); + for (const page of pages) { + // Batch enumeration can already be stale. Refresh before deciding to skip + // so an edit between listPages and this check cannot match an old marker. + const current = await engine.getPage(page.slug, { sourceId }); + if (!current) continue; + const token = hasRawTranscriptSidecar(current) + ? (await preparePageSnapshot(engine, current)).versionToken + : regularPageVersionToken(current); + expected.set(current.slug, token); + } + const rows = await engine.executeRaw<{ + slug: string; + source: string; + source_session: string | null; + }>( + `SELECT source_markdown_slug AS slug, source, source_session + FROM facts + WHERE source_id = $1 + AND source_markdown_slug = ANY($2::text[]) + AND source = ANY($3::text[]) + ORDER BY source_markdown_slug, + CASE WHEN source = $4 THEN 0 ELSE 1 END`, + [ + sourceId, + pages.map((page) => page.slug), + [TERMINAL_AUDIT_SOURCE, NON_EXTRACTABLE_AUDIT_SOURCE], + TERMINAL_AUDIT_SOURCE, + ], + ); + const outcomes = new Map(); + for (const row of rows) { + if (outcomes.has(row.slug)) continue; + const token = expected.get(row.slug); + if (!token || row.source_session !== outcomeSession(row.source, row.slug, token)) { + continue; + } + outcomes.set( + row.slug, + row.source === TERMINAL_AUDIT_SOURCE ? 'complete' : 'non_extractable', + ); + } + return outcomes; +} + +function recordDurableOutcomeSkip( + state: ExtractCoreState, + outcome: DurableExtractionOutcome, +): void { + state.result.pages_considered++; + if (outcome === 'complete') state.result.pages_skipped_completed++; + else state.result.pages_skipped_non_extractable++; +} + +async function snapshotIsCurrent( + engine: BrainEngine, + sourceId: string, + snapshot: ConversationPageSnapshot, +): Promise { + const current = await engine.getPage(snapshot.page.slug, { sourceId }); + if (!current) return false; + const currentSnapshot = await preparePageSnapshot(engine, current); + return currentSnapshot.versionToken === snapshot.versionToken; +} + async function processPage( state: ExtractCoreState, - page: Page, + snapshot: ConversationPageSnapshot, sinceIso: string | undefined, ): Promise<{ newEndIso: string | null }> { + const { page, body } = snapshot; state.result.pages_considered++; // Body cap check first — pre-parse, pre-segment, pre-extraction. @@ -694,7 +838,6 @@ async function processPage( return { newEndIso: null }; } - const body = await readConversationBodyForParsing(state.engine, page); // v0.41.13.0: thread the full Page through the orchestrator so D8 // date-derivation chain (frontmatter.date > effective_date > // '1970-01-01') AND timezone_policy warnings apply. The historical @@ -733,9 +876,40 @@ async function processPage( ); } } + const allSegments = splitIntoSegments(messages); const segments = splitIntoSegments(messages, { sinceIso }); if (segments.length === 0) { state.result.pages_skipped++; + if ( + !state.dryRun && + parseResult.phase !== 'no_match' && + allSegments.length === 0 + ) { + if (await snapshotIsCurrent(state.engine, state.sourceId, snapshot)) { + const cleaned = await deleteOrphanFactsForPage( + state.engine, + state.sourceId, + page.slug, + ); + state.result.orphan_facts_cleaned += cleaned; + const rowNum = await peekRowNumStart( + state.engine, + state.sourceId, + page.slug, + ); + await writeNonExtractableAuditRow( + state.engine, + state.sourceId, + page.slug, + rowNum, + snapshot.versionToken, + messages.length === 0 + ? 'no conversation messages found' + : 'fewer than two eligible messages', + ); + state.result.pages_marked_non_extractable++; + } + } return { newEndIso: null }; } @@ -771,24 +945,22 @@ async function processPage( const text = renderSegmentForExtraction(page.title || page.slug, seg); const sessionId = `${PER_SEGMENT_SOURCE_PREFIX}:${page.slug}`; - let extracted: Awaited> = []; - try { - extracted = await extractFactsFromTurn({ - turnText: text, - sessionId, - source: PER_SEGMENT_SOURCE_PREFIX, - engine: state.engine, - abortSignal: state.signal, - }); - } catch (err) { - if (isAbortError(err)) throw err; - if (err instanceof BudgetExhausted) throw err; - // Per-segment LLM failures are best-effort; loop continues. - process.stderr.write( - `[extract-conversation-facts] segment ${seg.startIso}..${seg.endIso} extractor failed: ${(err as Error).message}\n`, + const extraction = await extractFactsFromTurnWithOutcome({ + turnText: text, + sessionId, + source: PER_SEGMENT_SOURCE_PREFIX, + engine: state.engine, + abortSignal: state.signal, + }); + if (!extraction.ok) { + const detail = extraction.error instanceof Error + ? `: ${extraction.error.message}` + : ''; + throw new Error( + `segment ${seg.startIso}..${seg.endIso} extraction failed (${extraction.reason})${detail}`, ); - extracted = []; } + const extracted = extraction.facts; state.result.segments_processed++; segmentsThisPage++; @@ -813,19 +985,9 @@ async function processPage( context: fact.context ?? `from ${page.slug} segment ${seg.startIso}..${seg.endIso}`, })); - try { - const ins = await state.engine.insertFacts(rows, { source_id: state.sourceId }); // gbrain-allow-direct-insert: canonical bulk extraction path for conversation pages — fences-as-system-of-record doesn't apply because conversations don't carry `## Facts` fences (the chat-log shape is the source-of-truth) - pageInsertedTotal += ins.inserted; - state.result.facts_inserted += ins.inserted; - } catch (err) { - if (isAbortError(err)) throw err; - // Batch failure is best-effort — segment is the transactional - // boundary, so a duplicate-key or constraint error rolls back - // this segment only. Loop continues. - process.stderr.write( - `[extract-conversation-facts] segment ${seg.startIso}..${seg.endIso} insertFacts failed: ${(err as Error).message}\n`, - ); - } + const ins = await state.engine.insertFacts(rows, { source_id: state.sourceId }); // gbrain-allow-direct-insert: canonical bulk extraction path for conversation pages — fences-as-system-of-record doesn't apply because conversations don't carry `## Facts` fences (the chat-log shape is the source-of-truth) + pageInsertedTotal += ins.inserted; + state.result.facts_inserted += ins.inserted; rowNum += extracted.length; } else { // dry-run: count for reporting, no DB write. @@ -841,20 +1003,28 @@ async function processPage( // segment (no break on segmentLimit; that's an explicit partial run). const fullyProcessed = state.segmentLimit === 0 || segmentsThisPage < state.segmentLimit; - if (!state.dryRun && fullyProcessed && newestEnd !== null) { - try { - await writeTerminalAuditRow(state.engine, state.sourceId, page.slug, rowNum); - rowNum++; - } catch (err) { - if (isAbortError(err)) throw err; - // Terminal-row write failure: page is NOT marked complete; next - // run resumes. Loud stderr so users see partial-success state. - process.stderr.write( - `[extract-conversation-facts] ${page.slug} terminal audit write failed: ${(err as Error).message}\n`, - ); - // Suppress the resume-state update so doctor still flags this page. - newestEnd = null; - } + if ( + !state.dryRun && + fullyProcessed && + newestEnd !== null && + await snapshotIsCurrent(state.engine, state.sourceId, snapshot) + ) { + // A terminal insert is part of the page transaction contract. Propagate + // failure so bulk accounting, CLI exit status, cycle status, and rollups all + // report the page as unfinished. + await writeTerminalAuditRow( + state.engine, + state.sourceId, + page.slug, + rowNum, + snapshot.versionToken, + ); + rowNum++; + } else if (!state.dryRun && fullyProcessed && newestEnd !== null) { + process.stderr.write( + `[extract-conversation-facts] ${page.slug} changed during extraction; leaving it unfinished for replay\n`, + ); + newestEnd = null; } if (!state.dryRun && newestEnd !== null) { @@ -879,13 +1049,14 @@ async function writeTerminalAuditRow( sourceId: string, slug: string, rowNum: number, + versionToken: string, ): Promise { const fact: NewFact & { row_num: number; source_markdown_slug: string } = { fact: 'EXTRACTION_COMPLETE', kind: 'fact', entity_slug: null, source: TERMINAL_AUDIT_SOURCE, - source_session: `${TERMINAL_AUDIT_SOURCE}:${slug}`, + source_session: outcomeSession(TERMINAL_AUDIT_SOURCE, slug, versionToken), confidence: 1.0, notability: 'low', row_num: rowNum, @@ -904,6 +1075,33 @@ async function writeTerminalAuditRow( * - If absent: create a fresh tracker scoped to `opts.maxCostUsd` * and run the body inside `withBudgetTracker`. */ +async function writeNonExtractableAuditRow( + engine: BrainEngine, + sourceId: string, + slug: string, + rowNum: number, + versionToken: string, + reason: string, +): Promise { + const fact: NewFact & { row_num: number; source_markdown_slug: string } = { + fact: 'EXTRACTION_NOT_APPLICABLE', + kind: 'fact', + entity_slug: null, + source: NON_EXTRACTABLE_AUDIT_SOURCE, + source_session: outcomeSession( + NON_EXTRACTABLE_AUDIT_SOURCE, + slug, + versionToken, + ), + confidence: 1.0, + notability: 'low', + context: `scanned, not extractable: ${reason}`, + row_num: rowNum, + source_markdown_slug: slug, + }; + await engine.insertFacts([fact], { source_id: sourceId }); // gbrain-allow-direct-insert: durable non-extractable audit outcome prevents repeated scans while remaining distinct from successful extraction +} + export async function runExtractConversationFactsCore( engine: BrainEngine, opts: ExtractConversationFactsCoreOpts, @@ -920,6 +1118,10 @@ export async function runExtractConversationFactsCore( pages_skipped: 0, pages_skipped_too_large: 0, pages_skipped_disappeared: 0, + pages_skipped_completed: 0, + pages_skipped_non_extractable: 0, + pages_marked_non_extractable: 0, + pages_failed: 0, pages_llm_fallback: 0, pages_lock_skipped: 0, orphan_facts_cleaned: 0, @@ -1012,21 +1214,41 @@ export async function runExtractConversationFactsCore( */ const processPageWithLock = async (page: Page): Promise => { const lockId = extractConversationFactsLockId(sourceId, page.slug); - - let sinceIso: string | undefined; - // Per-page resume: --force clears prior entries; normal path uses - // the latest endIso for this (sourceId, slug) from the shared map. if (opts.force) { state.cpMap.delete(cpMapKey(sourceId, page.slug)); } - const checkpointed = state.cpMap.get(cpMapKey(sourceId, page.slug)) ?? null; - sinceIso = pickLaterIso(checkpointed, opts.sinceIso); try { await withRefreshingLock( engine, lockId, - () => processPage(state, page, sinceIso), + async () => { + // Re-fetch under the advisory lock. Batch enumeration is only a + // candidate list; it must never become the snapshot we certify. + const currentPage = await engine.getPage(page.slug, { sourceId }); + if (!currentPage) { + state.result.pages_skipped_disappeared++; + return { newEndIso: null }; + } + + // Close the race between batch selection and lock acquisition. + if (!opts.force) { + const outcome = ( + await findFreshExtractionOutcomes(engine, sourceId, [currentPage]) + ).get(currentPage.slug); + if (outcome) { + recordDurableOutcomeSkip(state, outcome); + return { newEndIso: null }; + } + } + + // A checkpoint without a matching durable v2 outcome cannot prove + // which page snapshot it describes. Clear it and replay safely; + // delete-orphans-first makes that replay deterministic. + state.cpMap.delete(cpMapKey(sourceId, currentPage.slug)); + const snapshot = await preparePageSnapshot(engine, currentPage); + return processPage(state, snapshot, opts.sinceIso); + }, { ttlMinutes: PER_PAGE_LOCK_TTL_MINUTES }, ).then(() => undefined); } catch (err) { @@ -1077,15 +1299,33 @@ export async function runExtractConversationFactsCore( }); if (batch.length === 0) break; - // Respect --limit at batch granularity: clip the batch so we - // never overshoot the cap by `workers - 1` extra pages. let claimable = batch; - if (opts.limit) { - const remaining = opts.limit - processedPagesCount; - if (remaining < batch.length) claimable = batch.slice(0, remaining); + // Checkpoints are an intra-page cursor; fresh durable outcomes are + // the page-level selection authority and survive checkpoint GC. + if (!opts.force && claimable.length > 0) { + const fresh = await findFreshExtractionOutcomes( + engine, + sourceId, + claimable, + ); + claimable = claimable.filter((page) => { + const outcome = fresh.get(page.slug); + if (!outcome) return true; + recordDurableOutcomeSkip(state, outcome); + return false; + }); } - const pool = await runSlidingPool({ + // Apply --limit after durable filtering. The limit caps pages that + // need work, not already-completed pages scanned to find that work. + if (opts.limit) { + const remaining = opts.limit - processedPagesCount; + if (remaining < claimable.length) { + claimable = claimable.slice(0, remaining); + } + } + + const poolResult = await runSlidingPool({ items: claimable, workers, signal, @@ -1093,7 +1333,7 @@ export async function runExtractConversationFactsCore( onError: (error) => (isAbortError(error) ? 'abort' : 'continue'), failureLabel: (page) => page.slug, }); - const cancellation = pool.failures.find((failure) => + const cancellation = poolResult.failures.find((failure) => isAbortError(failure.error), ); if (cancellation) throw cancellation.error; @@ -1103,6 +1343,15 @@ export async function runExtractConversationFactsCore( name: 'AbortError', }); } + result.pages_failed += poolResult.errored; + for (const failure of poolResult.failures) { + const message = failure.error instanceof Error + ? failure.error.message + : String(failure.error); + process.stderr.write( + `[extract-conversation-facts] ${failure.label} failed: ${message}\n`, + ); + } processedPagesCount += claimable.length; offset += batch.length; @@ -1223,7 +1472,12 @@ async function writeRunReceiptAndRollup( extracted_at: now, total_rows: result.facts_inserted, cost_usd: result.spent_usd ?? 0, - summary: `Extracted ${result.facts_inserted} facts from ${result.pages_processed}/${result.pages_considered} eligible pages.`, + summary: + `Extracted ${result.facts_inserted} facts from ` + + `${result.pages_processed}/${result.pages_considered} eligible pages` + + (result.pages_failed > 0 + ? `; ${result.pages_failed} page(s) failed and remain unfinished.` + : '.'), }); } catch (err) { // Best-effort: receipt write failure shouldn't kill the run. @@ -1237,12 +1491,13 @@ async function writeRunReceiptAndRollup( // Rollup UPSERT: ALWAYS fire so doctor's extract_health sees the // cycle ran (even no-op runs are signal — they prove the extractor // was alive). Best-effort per F-OUT-19. + const incomplete = halted || result.pages_failed > 0; await upsertExtractRollup(engine, { kind: 'facts.conversation', source_id: sourceId, cost_delta: result.spent_usd ?? 0, - round_completed_delta: halted ? 0 : 1, - halt_delta: halted ? 1 : 0, + round_completed_delta: incomplete ? 0 : 1, + halt_delta: incomplete ? 1 : 0, }); } @@ -1470,6 +1725,10 @@ export async function runExtractConversationFacts( pages_skipped: 0, pages_skipped_too_large: 0, pages_skipped_disappeared: 0, + pages_skipped_completed: 0, + pages_skipped_non_extractable: 0, + pages_marked_non_extractable: 0, + pages_failed: 0, pages_llm_fallback: 0, pages_lock_skipped: 0, orphan_facts_cleaned: 0, @@ -1511,6 +1770,10 @@ export async function runExtractConversationFacts( aggregate.pages_skipped += perSource.pages_skipped; aggregate.pages_skipped_too_large += perSource.pages_skipped_too_large; aggregate.pages_skipped_disappeared += perSource.pages_skipped_disappeared; + aggregate.pages_skipped_completed += perSource.pages_skipped_completed; + aggregate.pages_skipped_non_extractable += perSource.pages_skipped_non_extractable; + aggregate.pages_marked_non_extractable += perSource.pages_marked_non_extractable; + aggregate.pages_failed += perSource.pages_failed; aggregate.pages_llm_fallback += perSource.pages_llm_fallback; aggregate.pages_lock_skipped += perSource.pages_lock_skipped; aggregate.orphan_facts_cleaned += perSource.orphan_facts_cleaned; @@ -1543,6 +1806,18 @@ export async function runExtractConversationFacts( if (aggregate.pages_skipped_disappeared > 0) { console.log(` Skipped ${aggregate.pages_skipped_disappeared} page(s) that disappeared between enumeration and fetch.`); } + if (aggregate.pages_skipped_completed > 0) { + console.log(` Skipped ${aggregate.pages_skipped_completed} page(s) with fresh durable completion outcomes.`); + } + if (aggregate.pages_skipped_non_extractable > 0) { + console.log(` Skipped ${aggregate.pages_skipped_non_extractable} page(s) previously scanned as not extractable.`); + } + if (aggregate.pages_marked_non_extractable > 0) { + console.log(` Marked ${aggregate.pages_marked_non_extractable} page(s) as scanned, not extractable.`); + } + if (aggregate.pages_failed > 0) { + console.error(` Failed ${aggregate.pages_failed} page(s); they remain unfinished and will retry.`); + } if (aggregate.pages_llm_fallback > 0) { console.log(` Parsed ${aggregate.pages_llm_fallback} page(s) with the opt-in LLM fallback.`); } @@ -1562,6 +1837,9 @@ export async function runExtractConversationFacts( // anyBudgetExhausted doesn't trigger exit 3; the budget message // above already tells the user what to do, and exit 0 is the right // signal for "ran to the cap intentionally." + if (aggregate.pages_failed > 0) { + process.exit(1); + } if (aggregate.pages_lock_skipped > 0 && !anyBudgetExhausted) { process.exit(3); } diff --git a/src/core/cycle/conversation-facts-backfill.ts b/src/core/cycle/conversation-facts-backfill.ts index 803976560..4186cb464 100644 --- a/src/core/cycle/conversation-facts-backfill.ts +++ b/src/core/cycle/conversation-facts-backfill.ts @@ -260,6 +260,10 @@ export async function runPhaseConversationFactsBackfill( pages_skipped: 0, pages_skipped_too_large: 0, pages_skipped_disappeared: 0, + pages_skipped_completed: 0, + pages_skipped_non_extractable: 0, + pages_marked_non_extractable: 0, + pages_failed: 1, pages_llm_fallback: 0, // v0.41.15.0 (D6 + D11): new counters from the per-page lock // + delete-orphans-first replay safety. @@ -297,6 +301,10 @@ export async function runPhaseConversationFactsBackfill( const totals = { pages_processed: 0, pages_skipped: 0, + pages_skipped_completed: 0, + pages_skipped_non_extractable: 0, + pages_marked_non_extractable: 0, + pages_failed: 0, facts_inserted: 0, sources_processed: 0, }; @@ -304,10 +312,16 @@ export async function runPhaseConversationFactsBackfill( if (!r.error) totals.sources_processed++; totals.pages_processed += r.pages_processed; totals.pages_skipped += r.pages_skipped; + totals.pages_skipped_completed += r.pages_skipped_completed; + totals.pages_skipped_non_extractable += r.pages_skipped_non_extractable; + totals.pages_marked_non_extractable += r.pages_marked_non_extractable; + totals.pages_failed += r.pages_failed; totals.facts_inserted += r.facts_inserted; } - const anyError = Object.values(perSourceResults).some((r) => r.error); + const anyError = Object.values(perSourceResults).some( + (r) => r.error || r.pages_failed > 0, + ); const status = anyError ? 'warn' : 'ok'; const summary = `${totals.facts_inserted} facts inserted across ${totals.sources_processed}/${sources.length} sources, ~$${totalSpent.toFixed(4)} spent`; @@ -321,6 +335,10 @@ export async function runPhaseConversationFactsBackfill( sources_processed: totals.sources_processed, pages_processed: totals.pages_processed, pages_skipped: totals.pages_skipped, + pages_skipped_completed: totals.pages_skipped_completed, + pages_skipped_non_extractable: totals.pages_skipped_non_extractable, + pages_marked_non_extractable: totals.pages_marked_non_extractable, + pages_failed: totals.pages_failed, facts_inserted: totals.facts_inserted, spent_usd: totalSpent, skipped_by_brain_wide_cap: skippedByBrainWideCap, diff --git a/src/core/facts/extract.ts b/src/core/facts/extract.ts index 1dc2b2956..4534947b8 100644 --- a/src/core/facts/extract.ts +++ b/src/core/facts/extract.ts @@ -215,20 +215,38 @@ const EXTRACTOR_SYSTEM = [ const MAX_TURN_TEXT_CHARS = 8000; -export async function extractFactsFromTurn(input: ExtractInput): Promise { - if (input.isDreamGenerated) return []; - if (!input.turnText) return []; +export type ExtractFactsOutcome = + | { ok: true; facts: ExtractedFact[] } + | { + ok: false; + reason: + | 'chat_unavailable' + | 'provider_error' + | 'refusal' + | 'content_filter' + | 'non_terminal_stop' + | 'malformed_output' + | 'truncated_output'; + error?: unknown; + }; + +/** Strict extraction contract for callers that persist completion authority. */ +export async function extractFactsFromTurnWithOutcome( + input: ExtractInput, +): Promise { + if (input.isDreamGenerated) return { ok: true, facts: [] }; + if (!input.turnText) return { ok: true, facts: [] }; // Anti-loop + sanitization. let cleaned = input.turnText.slice(0, MAX_TURN_TEXT_CHARS); for (const p of INJECTION_PATTERNS) cleaned = cleaned.replace(p.rx, p.replacement); cleaned = cleaned.trim(); - if (!cleaned) return []; + if (!cleaned) return { ok: true, facts: [] }; if (!isAvailable('chat')) { // No chat gateway → no extraction. Caller still inserts facts via direct // `gbrain take add` paths. - return []; + return { ok: false, reason: 'chat_unavailable' }; } const cap = Math.max(1, Math.min(input.maxFactsPerTurn ?? 10, 25)); @@ -271,19 +289,29 @@ export async function extractFactsFromTurn(input: ExtractInput): Promise\n`, ); + return { ok: false, reason: 'truncated_output' }; } } } catch (err) { - // Re-throw aborts; absorb other errors as "no extraction" — caller's - // `put_page` backstop will still record the page itself. + // Re-throw aborts. Strict callers receive a failure outcome; the historical + // wrapper below converts that outcome to [] for best-effort call sites. if (isAbort(err)) throw err; - return []; + return { ok: false, reason: 'provider_error', error: err }; } - if (result.stopReason === 'refusal' || result.stopReason === 'content_filter') return []; + if (result.stopReason === 'refusal') return { ok: false, reason: 'refusal' }; + if (result.stopReason === 'content_filter') { + return { ok: false, reason: 'content_filter' }; + } + if (result.stopReason !== 'end') { + return { ok: false, reason: 'non_terminal_stop' }; + } - const parsedRaw = parseExtractorJson(result.text); - if (!parsedRaw) return []; + const parsedShape = parseExtractorJsonDetailed(result.text); + if (!parsedShape || parsedShape.invalidCandidates > 0) { + return { ok: false, reason: 'malformed_output' }; + } + const parsedRaw = parsedShape.facts; const facts: ExtractedFact[] = []; for (const candidate of parsedRaw.slice(0, cap)) { @@ -345,7 +373,13 @@ export async function extractFactsFromTurn(input: ExtractInput): Promise { + const outcome = await extractFactsFromTurnWithOutcome(input); + return outcome.ok ? outcome.facts : []; } interface RawExtracted { @@ -368,30 +402,46 @@ interface RawExtracted { * the model included it. Production callers should use extractFactsFromTurn. */ export function parseExtractorJson(raw: string): RawExtracted[] | null { + return parseExtractorJsonDetailed(raw)?.facts ?? null; +} + +interface ParsedExtractorShape { + facts: RawExtracted[]; + invalidCandidates: number; +} + +function parseExtractorJsonDetailed(raw: string): ParsedExtractorShape | null { const cleaned = raw.trim().replace(/^```(?:json)?\s*/, '').replace(/\s*```$/, ''); // Strict. - const direct = tryArrayShape(cleaned); + const direct = tryArrayShapeDetailed(cleaned); if (direct) return direct; // Substring scan for embedded {"facts":[...]} shape. const m = cleaned.match(/\{[\s\S]*?"facts"[\s\S]*\}/); if (m) { - const sub = tryArrayShape(m[0]); + const sub = tryArrayShapeDetailed(m[0]); if (sub) return sub; } return null; } -function tryArrayShape(s: string): RawExtracted[] | null { +function tryArrayShapeDetailed(s: string): ParsedExtractorShape | null { try { const parsed = JSON.parse(s) as unknown; if (typeof parsed !== 'object' || parsed === null) return null; const arr = (parsed as Record).facts; if (!Array.isArray(arr)) return null; const out: RawExtracted[] = []; + let invalidCandidates = 0; for (const item of arr) { - if (typeof item !== 'object' || item === null) continue; + if (typeof item !== 'object' || item === null) { + invalidCandidates++; + continue; + } const o = item as Record; - if (typeof o.fact !== 'string' || typeof o.kind !== 'string') continue; + if (typeof o.fact !== 'string' || typeof o.kind !== 'string') { + invalidCandidates++; + continue; + } out.push({ fact: o.fact, kind: o.kind, @@ -408,7 +458,7 @@ function tryArrayShape(s: string): RawExtracted[] | null { period: typeof o.period === 'string' ? o.period : null, }); } - return out; + return { facts: out, invalidCandidates }; } catch { return null; } diff --git a/src/core/pglite-engine.ts b/src/core/pglite-engine.ts index bae0181b2..c842a08b4 100644 --- a/src/core/pglite-engine.ts +++ b/src/core/pglite-engine.ts @@ -976,6 +976,7 @@ export class PGLiteEngine implements BrainEngine { } const { rows } = await this.db.query( `SELECT id, source_id, slug, type, title, compiled_truth, timeline, frontmatter, content_hash, created_at, updated_at, deleted_at, + effective_date, effective_date_source, source_kind, source_uri, ingested_via, ingested_at FROM pages WHERE ${where.join(' AND ')} LIMIT 1`, params diff --git a/src/core/postgres-engine.ts b/src/core/postgres-engine.ts index ad119a5b9..173e8bd2f 100644 --- a/src/core/postgres-engine.ts +++ b/src/core/postgres-engine.ts @@ -1028,6 +1028,7 @@ export class PostgresEngine implements BrainEngine { const deletedCondition = includeDeleted ? tx`` : tx`AND deleted_at IS NULL`; const rows = await tx` SELECT id, source_id, slug, type, title, compiled_truth, timeline, frontmatter, content_hash, created_at, updated_at, deleted_at, + effective_date, effective_date_source, source_kind, source_uri, ingested_via, ingested_at FROM pages WHERE slug = ${slug} ${sourceCondition} ${deletedCondition} diff --git a/test/doctor-conversation-facts-backlog.test.ts b/test/doctor-conversation-facts-backlog.test.ts new file mode 100644 index 000000000..86e61264b --- /dev/null +++ b/test/doctor-conversation-facts-backlog.test.ts @@ -0,0 +1,156 @@ +import { afterAll, beforeAll, beforeEach, describe, expect, test } from 'bun:test'; +import { createHash } from 'node:crypto'; +import { mkdtempSync, mkdirSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { PGLiteEngine } from '../src/core/pglite-engine.ts'; +import { resetPgliteState } from './helpers/reset-pglite.ts'; +import { computeConversationFactsBacklogCheck } from '../src/commands/doctor.ts'; +import { + NON_EXTRACTABLE_AUDIT_SOURCE, + TERMINAL_AUDIT_SOURCE, +} from '../src/commands/extract-conversation-facts.ts'; + +let engine: PGLiteEngine; + +beforeAll(async () => { + engine = new PGLiteEngine(); + await engine.connect({}); + await engine.initSchema(); +}); + +afterAll(async () => { + await engine.disconnect(); +}); + +beforeEach(async () => { + await resetPgliteState(engine); + await engine.setConfig('cycle.conversation_facts_backfill.enabled', 'true'); +}); + +async function seedPage(slug: string, type: string): Promise { + await engine.putPage(slug, { + type, + title: slug, + compiled_truth: 'A page body long enough for the doctor backlog fixture.', + timeline: '', + frontmatter: {}, + }); +} + +async function seedOutcome(slug: string, source: string): Promise { + const pages = await engine.executeRaw<{ + content_hash: string; + effective_date: Date | null; + }>( + `SELECT content_hash, effective_date + FROM pages WHERE source_id = 'default' AND slug = $1`, + [slug], + ); + const effectiveDate = pages[0]!.effective_date + ? new Date(pages[0]!.effective_date).toISOString().slice(0, 10) + : 'none'; + const version = `page-${pages[0]!.content_hash}-${effectiveDate}`; + await engine.executeRaw( + `INSERT INTO facts ( + fact, kind, source, source_session, confidence, notability, + row_num, source_markdown_slug, source_id + ) VALUES ($1, 'fact', $2, $3, 1.0, 'low', 0, $4, 'default')`, + [ + source === TERMINAL_AUDIT_SOURCE + ? 'EXTRACTION_COMPLETE' + : 'EXTRACTION_NOT_APPLICABLE', + source, + `${source}:${slug}:${version}`, + slug, + ], + ); +} + +describe('conversation_facts_backlog durable outcomes', () => { + test('reports complete, scanned-not-extractable, and backlog separately', async () => { + await seedPage('meetings/complete', 'meeting'); + await seedPage('slack/not-applicable', 'slack'); + await seedPage('meetings/pending', 'meeting'); + await seedOutcome('meetings/complete', TERMINAL_AUDIT_SOURCE); + await seedOutcome('slack/not-applicable', NON_EXTRACTABLE_AUDIT_SOURCE); + + const result = await computeConversationFactsBacklogCheck(engine); + expect(result.details?.backlog).toBe(1); + expect(result.details?.completed).toBe(1); + expect(result.details?.scanned_not_extractable).toBe(1); + }); + + test('a content change invalidates the prior outcome', async () => { + await seedPage('meetings/growing', 'meeting'); + await seedOutcome('meetings/growing', TERMINAL_AUDIT_SOURCE); + await engine.putPage('meetings/growing', { + type: 'meeting', + title: 'meetings/growing', + compiled_truth: 'The meeting body changed after its durable outcome.', + timeline: '', + frontmatter: {}, + }); + + const result = await computeConversationFactsBacklogCheck(engine); + expect(result.details?.backlog).toBe(1); + expect(result.details?.completed).toBe(0); + }); + + test('a sidecar-only edit invalidates doctor completion', async () => { + const repoDir = mkdtempSync(join(tmpdir(), 'gbrain-doctor-sidecar-')); + try { + const relativePath = 'meetings/sidecar.raw/transcript.txt'; + const transcriptPath = join(repoDir, relativePath); + mkdirSync(join(repoDir, 'meetings/sidecar.raw'), { recursive: true }); + const body = 'Speaker A: Initial statement.\nSpeaker B: Initial reply.'; + writeFileSync(transcriptPath, body, 'utf8'); + await engine.setConfig('sync.repo_path', repoDir); + const frontmatter = { raw_transcript: relativePath }; + await engine.putPage('meetings/sidecar', { + type: 'meeting', + title: 'Sidecar meeting', + compiled_truth: 'Summary only.', + timeline: '', + frontmatter, + }); + const page = await engine.getPage('meetings/sidecar', { sourceId: 'default' }); + const token = `sidecar-${createHash('sha256') + .update(JSON.stringify({ + body, + title: page!.title, + type: page!.type, + frontmatter: page!.frontmatter, + effective_date: page!.effective_date ?? null, + })) + .digest('hex')}`; + await engine.executeRaw( + `INSERT INTO facts ( + fact, kind, source, source_session, confidence, notability, + row_num, source_markdown_slug, source_id + ) VALUES ( + 'EXTRACTION_COMPLETE', 'fact', $1, $2, 1.0, 'low', 0, $3, 'default' + )`, + [ + TERMINAL_AUDIT_SOURCE, + `${TERMINAL_AUDIT_SOURCE}:meetings/sidecar:${token}`, + 'meetings/sidecar', + ], + ); + const before = await computeConversationFactsBacklogCheck(engine); + expect(before.details?.completed).toBe(1); + expect(before.details?.backlog).toBe(0); + + writeFileSync( + transcriptPath, + 'Speaker A: Edited sidecar statement.\nSpeaker B: Edited reply.', + 'utf8', + ); + const after = await computeConversationFactsBacklogCheck(engine); + expect(after.details?.completed).toBe(0); + expect(after.details?.backlog).toBe(1); + } finally { + rmSync(repoDir, { recursive: true, force: true }); + } + }); +}); diff --git a/test/extract-conversation-facts.test.ts b/test/extract-conversation-facts.test.ts index 23033830d..63057aa78 100644 --- a/test/extract-conversation-facts.test.ts +++ b/test/extract-conversation-facts.test.ts @@ -36,6 +36,7 @@ import { SEGMENT_TEXT_CHAR_LIMIT, MAX_PAGE_BODY_BYTES, TERMINAL_AUDIT_SOURCE, + NON_EXTRACTABLE_AUDIT_SOURCE, PER_SEGMENT_SOURCE_PREFIX, ALLOWED_TYPES, } from '../src/commands/extract-conversation-facts.ts'; @@ -259,6 +260,10 @@ const SAMPLE_BODY = [ describe('runExtractConversationFactsCore', () => { let engine: PGLiteEngine; let repoDir: string; + let chatFailure: Error | null = null; + let chatHook: (() => Promise) | null = null; + let chatStopReason: ChatResult['stopReason'] = 'end'; + let chatTextOverride: string | null = null; let fallbackCalls = 0; let fallbackContents: string[] = []; let fallbackControlError: Error | null = null; @@ -312,9 +317,13 @@ describe('runExtractConversationFactsCore', () => { providerId: 'stub', }; } + if (chatFailure) throw chatFailure; + const hook = chatHook; + chatHook = null; + if (hook) await hook(); callIndex++; return { - text: JSON.stringify({ + text: chatTextOverride ?? JSON.stringify({ facts: [{ fact: `synthetic fact #${callIndex}`, kind: 'event', @@ -324,7 +333,7 @@ describe('runExtractConversationFactsCore', () => { }], }), blocks: [], - stopReason: 'end', + stopReason: chatStopReason, usage: { input_tokens: 100, output_tokens: 50, @@ -353,6 +362,10 @@ describe('runExtractConversationFactsCore', () => { }); beforeEach(async () => { + chatFailure = null; + chatHook = null; + chatStopReason = 'end'; + chatTextOverride = null; fallbackCalls = 0; fallbackContents = []; fallbackControlError = null; @@ -536,8 +549,8 @@ describe('runExtractConversationFactsCore', () => { sleepMs: 0, }); expect(second.pages_processed).toBe(0); - expect(second.pages_skipped).toBe(1); - // The content-hash cache serves the deterministic replay for free. + // The durable completion outcome skips the replay before any parse. + expect(second.pages_skipped_completed).toBe(1); expect(fallbackCalls).toBe(1); }); }); @@ -562,7 +575,8 @@ describe('runExtractConversationFactsCore', () => { sleepMs: 0, }); expect(second.pages_processed).toBe(0); - expect(second.pages_skipped).toBe(1); + // The durable completion outcome skips the replay before any parse. + expect(second.pages_skipped_completed).toBe(1); expect(fallbackCalls).toBe(3); }); }); @@ -724,12 +738,487 @@ describe('runExtractConversationFactsCore', () => { // Terminal audit row present. const terminalRows = await engine.executeRaw<{ count: string | number }>( - `SELECT COUNT(*) AS count FROM facts WHERE source = $1 AND source_session = $2`, - [TERMINAL_AUDIT_SOURCE, `${TERMINAL_AUDIT_SOURCE}:conversations/imessage/alice-example`], + `SELECT COUNT(*) AS count FROM facts + WHERE source = $1 AND source_session LIKE $2`, + [TERMINAL_AUDIT_SOURCE, `${TERMINAL_AUDIT_SOURCE}:conversations/imessage/alice-example:page-%`], ); expect(Number(terminalRows[0]?.count ?? 0)).toBe(1); }); + test('terminal outcome skips a completed page after checkpoint GC', async () => { + await runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'conversations/imessage/alice-example', + sleepMs: 0, + }); + await engine.executeRaw( + `DELETE FROM op_checkpoints WHERE op = 'extract-conversation-facts'`, + ); + + const second = await runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'conversations/imessage/alice-example', + sleepMs: 0, + }); + expect(second.pages_skipped_completed).toBe(1); + expect(second.pages_processed).toBe(0); + expect(second.segments_processed).toBe(0); + }); + + test('page edits make an older terminal outcome stale', async () => { + await runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'conversations/imessage/alice-example', + sleepMs: 0, + }); + await engine.putPage('conversations/imessage/alice-example', { + type: 'conversation', + title: 'iMessage: Alice Example', + compiled_truth: SAMPLE_BODY + '\n' + [ + fmt('Alice Example', '2024-03-17', '9:00 AM', 'new tail'), + fmt('Bob Demo', '2024-03-17', '9:01 AM', 'new response'), + ].join('\n'), + timeline: '', + frontmatter: {}, + }); + + const second = await runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'conversations/imessage/alice-example', + sleepMs: 0, + }); + expect(second.pages_skipped_completed).toBe(0); + expect(second.pages_processed).toBe(1); + }); + + test('records and then skips a definitive scan with no eligible segment', async () => { + await engine.putPage('conversations/single-message', { + type: 'slack', + title: 'Single message', + compiled_truth: fmt('Alice Example', '2024-03-15', '9:00 AM', 'only one'), + timeline: '', + frontmatter: {}, + }); + const first = await runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'conversations/single-message', + types: ['slack'], + sleepMs: 0, + }); + expect(first.pages_marked_non_extractable).toBe(1); + const markers = await engine.executeRaw<{ count: string | number }>( + `SELECT COUNT(*) AS count FROM facts + WHERE source = $1 AND source_session LIKE $2`, + [ + NON_EXTRACTABLE_AUDIT_SOURCE, + `${NON_EXTRACTABLE_AUDIT_SOURCE}:conversations/single-message:page-%`, + ], + ); + expect(Number(markers[0]?.count ?? 0)).toBe(1); + + const second = await runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'conversations/single-message', + types: ['slack'], + sleepMs: 0, + }); + expect(second.pages_skipped_non_extractable).toBe(1); + expect(second.pages_marked_non_extractable).toBe(0); + }); + + test('does not classify an unrecognized parser miss as non-extractable', async () => { + await engine.putPage('meetings/unrecognized-format', { + type: 'meeting', + title: 'Unrecognized meeting format', + compiled_truth: 'Alice spoke first. Bob answered later.', + timeline: '', + frontmatter: {}, + }); + const result = await runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'meetings/unrecognized-format', + types: ['meeting'], + sleepMs: 0, + }); + expect(result.pages_marked_non_extractable).toBe(0); + const markers = await engine.executeRaw<{ count: string | number }>( + `SELECT COUNT(*) AS count FROM facts WHERE source = $1 AND source_markdown_slug = $2`, + [NON_EXTRACTABLE_AUDIT_SOURCE, 'meetings/unrecognized-format'], + ); + expect(Number(markers[0]?.count ?? 0)).toBe(0); + }); + + test('same-timestamp text edits replay instead of trusting a stale checkpoint', async () => { + await runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'conversations/imessage/alice-example', + sleepMs: 0, + }); + await engine.putPage('conversations/imessage/alice-example', { + type: 'conversation', + title: 'iMessage: Alice Example', + compiled_truth: SAMPLE_BODY.replace( + 'Staff engineer on the platform team.', + 'Principal engineer on the infrastructure team.', + ), + timeline: '', + frontmatter: {}, + }); + + const second = await runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'conversations/imessage/alice-example', + sleepMs: 0, + }); + expect(second.pages_skipped_completed).toBe(0); + expect(second.segments_processed).toBe(2); + }); + + test('an edit during extraction cannot mint a terminal for the old snapshot', async () => { + chatHook = async () => { + await engine.putPage('conversations/imessage/alice-example', { + type: 'conversation', + title: 'iMessage: Alice Example', + compiled_truth: SAMPLE_BODY.replace('Nice.', 'Updated while extraction ran.'), + timeline: '', + frontmatter: {}, + }); + }; + const first = await runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'conversations/imessage/alice-example', + sleepMs: 0, + }); + expect(first.pages_processed).toBe(1); + const terminals = await engine.executeRaw<{ count: string | number }>( + `SELECT COUNT(*) AS count FROM facts WHERE source = $1`, + [TERMINAL_AUDIT_SOURCE], + ); + expect(Number(terminals[0]?.count ?? 0)).toBe(0); + + const retry = await runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'conversations/imessage/alice-example', + sleepMs: 0, + }); + expect(retry.pages_skipped_completed).toBe(0); + expect(retry.pages_processed).toBe(1); + }); + + test('raw transcript sidecar edits invalidate the durable outcome', async () => { + await runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'meetings/raw-speaker-example', + types: ['meeting'], + sleepMs: 0, + }); + writeFileSync( + join(repoDir, 'meetings/raw-speaker-example.raw/transcript.txt'), + [ + 'Speaker A: The sidecar changed after the first extraction.', + 'Speaker B: Then the snapshot hash must force a replay.', + ].join('\n'), + 'utf8', + ); + const second = await runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'meetings/raw-speaker-example', + types: ['meeting'], + sleepMs: 0, + }); + expect(second.pages_skipped_completed).toBe(0); + expect(second.pages_processed).toBe(1); + }); + + test('provider failure leaves no terminal and retries on the next run', async () => { + chatFailure = new Error('synthetic provider outage'); + await expect( + runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'conversations/imessage/alice-example', + sleepMs: 0, + }), + ).rejects.toThrow('provider_error'); + const terminals = await engine.executeRaw<{ count: string | number }>( + `SELECT COUNT(*) AS count FROM facts WHERE source = $1`, + [TERMINAL_AUDIT_SOURCE], + ); + expect(Number(terminals[0]?.count ?? 0)).toBe(0); + + chatFailure = null; + const retry = await runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'conversations/imessage/alice-example', + sleepMs: 0, + }); + expect(retry.pages_processed).toBe(1); + }); + + test('non-terminal model stop leaves no terminal outcome', async () => { + chatStopReason = 'other'; + await expect( + runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'conversations/imessage/alice-example', + sleepMs: 0, + }), + ).rejects.toThrow('non_terminal_stop'); + const terminals = await engine.executeRaw<{ count: string | number }>( + `SELECT COUNT(*) AS count FROM facts WHERE source = $1`, + [TERMINAL_AUDIT_SOURCE], + ); + expect(Number(terminals[0]?.count ?? 0)).toBe(0); + }); + + test('schema-invalid model facts leave no terminal outcome', async () => { + chatTextOverride = JSON.stringify({ facts: [{ fact: 123, kind: 'fact' }] }); + await expect( + runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'conversations/imessage/alice-example', + sleepMs: 0, + }), + ).rejects.toThrow('malformed_output'); + const terminals = await engine.executeRaw<{ count: string | number }>( + `SELECT COUNT(*) AS count FROM facts WHERE source = $1`, + [TERMINAL_AUDIT_SOURCE], + ); + expect(Number(terminals[0]?.count ?? 0)).toBe(0); + }); + + test('insert failure leaves no terminal and retries from a clean replay', async () => { + const engineAny = engine as any; + const originalInsertFacts = engineAny.insertFacts.bind(engine); + engineAny.insertFacts = async (facts: Array<{ source?: string }>, opts: unknown) => { + if (facts.some((fact) => fact.source === PER_SEGMENT_SOURCE_PREFIX)) { + throw new Error('synthetic insert outage'); + } + return originalInsertFacts(facts, opts); + }; + try { + await expect( + runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'conversations/imessage/alice-example', + sleepMs: 0, + }), + ).rejects.toThrow('synthetic insert outage'); + } finally { + engineAny.insertFacts = originalInsertFacts; + } + const terminals = await engine.executeRaw<{ count: string | number }>( + `SELECT COUNT(*) AS count FROM facts WHERE source = $1`, + [TERMINAL_AUDIT_SOURCE], + ); + expect(Number(terminals[0]?.count ?? 0)).toBe(0); + const retry = await runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'conversations/imessage/alice-example', + sleepMs: 0, + }); + expect(retry.pages_processed).toBe(1); + }); + + test('terminal insert failure is reported as unfinished in bulk mode', async () => { + const engineAny = engine as any; + const originalInsertFacts = engineAny.insertFacts.bind(engine); + engineAny.insertFacts = async (facts: Array<{ source?: string }>, opts: unknown) => { + if (facts.some((fact) => fact.source === TERMINAL_AUDIT_SOURCE)) { + throw new Error('synthetic terminal insert outage'); + } + return originalInsertFacts(facts, opts); + }; + try { + const result = await runExtractConversationFactsCore(engine, { + sourceId: 'default', + types: ['conversation'], + sleepMs: 0, + }); + expect(result.pages_failed).toBe(1); + expect(result.pages_processed).toBe(0); + } finally { + engineAny.insertFacts = originalInsertFacts; + } + const terminals = await engine.executeRaw<{ count: string | number }>( + `SELECT COUNT(*) AS count FROM facts WHERE source = $1`, + [TERMINAL_AUDIT_SOURCE], + ); + expect(Number(terminals[0]?.count ?? 0)).toBe(0); + }); + + test('cleanup failure cannot mint a non-extractable marker', async () => { + await engine.putPage('conversations/cleanup-failure', { + type: 'slack', + title: 'Cleanup failure', + compiled_truth: fmt('Alice Example', '2024-03-15', '9:00 AM', 'one message'), + timeline: '', + frontmatter: {}, + }); + const engineAny = engine as any; + const originalExecuteRaw = engineAny.executeRaw.bind(engine); + engineAny.executeRaw = async (sql: string, params?: unknown[]) => { + if (sql.includes('WITH del AS')) throw new Error('synthetic cleanup outage'); + return originalExecuteRaw(sql, params); + }; + try { + await expect( + runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'conversations/cleanup-failure', + types: ['slack'], + sleepMs: 0, + }), + ).rejects.toThrow('synthetic cleanup outage'); + } finally { + engineAny.executeRaw = originalExecuteRaw; + } + const markers = await engine.executeRaw<{ count: string | number }>( + `SELECT COUNT(*) AS count FROM facts WHERE source = $1`, + [NON_EXTRACTABLE_AUDIT_SOURCE], + ); + expect(Number(markers[0]?.count ?? 0)).toBe(0); + }); + + test('--limit counts pending work after completed pages are filtered', async () => { + for (const slug of ['conversations/a-complete', 'conversations/b-pending']) { + await engine.putPage(slug, { + type: 'slack', + title: slug, + compiled_truth: SAMPLE_BODY, + timeline: '', + frontmatter: {}, + }); + } + await runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'conversations/a-complete', + types: ['slack'], + sleepMs: 0, + }); + const bulk = await runExtractConversationFactsCore(engine, { + sourceId: 'default', + types: ['slack'], + limit: 1, + sleepMs: 0, + }); + expect(bulk.pages_skipped_completed).toBe(1); + expect(bulk.pages_processed).toBe(1); + const pendingTerminal = await engine.executeRaw<{ count: string | number }>( + `SELECT COUNT(*) AS count FROM facts + WHERE source = $1 AND source_markdown_slug = $2`, + [TERMINAL_AUDIT_SOURCE, 'conversations/b-pending'], + ); + expect(Number(pendingTerminal[0]?.count ?? 0)).toBe(1); + }); + + test('bulk mode reports provider failures instead of returning a clean result', async () => { + chatFailure = new Error('synthetic bulk provider outage'); + const result = await runExtractConversationFactsCore(engine, { + sourceId: 'default', + types: ['conversation'], + sleepMs: 0, + }); + expect(result.pages_failed).toBe(1); + expect(result.pages_processed).toBe(0); + }); + + test('content identity reopens a page even when updated_at is unchanged', async () => { + await runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'conversations/imessage/alice-example', + sleepMs: 0, + }); + const original = await engine.executeRaw<{ updated_at: Date }>( + `SELECT updated_at FROM pages + WHERE source_id = 'default' AND slug = 'conversations/imessage/alice-example'`, + ); + await engine.putPage('conversations/imessage/alice-example', { + type: 'conversation', + title: 'iMessage: Alice Example', + compiled_truth: SAMPLE_BODY.replace('Nice.', 'Changed at the same timestamp.'), + timeline: '', + frontmatter: {}, + }); + await engine.executeRaw( + `UPDATE pages SET updated_at = $1 + WHERE source_id = 'default' AND slug = 'conversations/imessage/alice-example'`, + [original[0]!.updated_at], + ); + const result = await runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'conversations/imessage/alice-example', + sleepMs: 0, + }); + expect(result.pages_skipped_completed).toBe(0); + expect(result.pages_processed).toBe(1); + }); + + test('effective_date survives locked refetch and invalidates completion', async () => { + await engine.putPage('meetings/effective-date', { + type: 'meeting', + title: 'Effective date meeting', + compiled_truth: [ + 'Speaker A: We approved the proposal.', + 'Speaker B: I will publish it tomorrow.', + ].join('\n'), + timeline: '', + frontmatter: {}, + }); + await engine.executeRaw( + `UPDATE pages SET effective_date = '2026-01-01T00:00:00Z' + WHERE source_id = 'default' AND slug = 'meetings/effective-date'`, + ); + const first = await runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'meetings/effective-date', + types: ['meeting'], + sleepMs: 0, + }); + expect(first.pages_processed).toBe(1); + const firstTerminal = await engine.executeRaw<{ source_session: string }>( + `SELECT source_session FROM facts + WHERE source = $1 AND source_markdown_slug = 'meetings/effective-date'`, + [TERMINAL_AUDIT_SOURCE], + ); + expect(firstTerminal[0]!.source_session.endsWith('-2026-01-01')).toBe(true); + + await engine.executeRaw( + `UPDATE pages SET effective_date = '2026-01-02T00:00:00Z' + WHERE source_id = 'default' AND slug = 'meetings/effective-date'`, + ); + const second = await runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'meetings/effective-date', + types: ['meeting'], + sleepMs: 0, + }); + expect(second.pages_skipped_completed).toBe(0); + expect(second.pages_processed).toBe(1); + }); + + test('legacy terminal rows do not suppress strict v2 replay', async () => { + await engine.executeRaw( + `INSERT INTO facts ( + fact, kind, source, source_session, confidence, notability, + row_num, source_markdown_slug, source_id + ) VALUES ( + 'EXTRACTION_COMPLETE', 'fact', $1, $2, 1.0, 'low', 0, $3, 'default' + )`, + [ + 'cli:extract-conversation-facts:terminal', + 'cli:extract-conversation-facts:terminal:conversations/imessage/alice-example', + 'conversations/imessage/alice-example', + ], + ); + const result = await runExtractConversationFactsCore(engine, { + sourceId: 'default', + slug: 'conversations/imessage/alice-example', + sleepMs: 0, + }); + expect(result.pages_skipped_completed).toBe(0); + expect(result.pages_processed).toBe(1); + }); + test('row_num accumulator: segment 2 facts start after segment 1 (Codex C1)', async () => { await runExtractConversationFactsCore(engine, { sourceId: 'default', @@ -764,7 +1253,7 @@ describe('runExtractConversationFactsCore', () => { slug: 'conversations/imessage/alice-example', sleepMs: 0, }); - expect(second.pages_skipped).toBe(1); + expect(second.pages_skipped_completed).toBe(1); // Re-run with force: re-processes. const third = await runExtractConversationFactsCore(engine, { sourceId: 'default',