mirror of
https://github.com/garrytan/gbrain.git
synced 2026-07-30 11:22:34 +00:00
* v0.42.66.0 fix(extract): make conversation backfill outcomes durable (takeover of #3293) Versioned, snapshot-bound terminal audit rows become the durable authority for conversation fact backfill completion; checkpoint GC can no longer repeat completed model work, and best-effort empty results no longer mask provider/output failures as complete. Supersedes #3293 (rebased onto current master; only version-trio conflicts). Co-authored-by: FloridaStyle <danwiggins@users.noreply.github.com> Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * chore: drop version-trio bump — individual fixes do not carry version bumps (release PRs do) * merge: reconcile durable-outcome skip accounting with master's LLM fallback tests The two fallback replay tests from #3371 asserted the legacy checkpoint pages_skipped counter; under this PR's durable-outcome authority a completed page is skipped via pages_skipped_completed before any parse. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: Garry Tan <garrytan@gmail.com> Co-authored-by: FloridaStyle <danwiggins@users.noreply.github.com> Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Time Attakc
co-authored by
Claude Fable 5
Garry Tan
FloridaStyle
Claude Fable 5
Garry Tan
FloridaStyle
parent
54c0c93376
commit
32d42454e9
File diff suppressed because one or more lines are too long
@@ -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
|
||||
<outcome-source>:<page-slug>:<version-token>
|
||||
```
|
||||
|
||||
There are two token forms.
|
||||
|
||||
### Database-backed page body
|
||||
|
||||
For pages parsed from `compiled_truth` and `timeline`, the token is:
|
||||
|
||||
```text
|
||||
page-<pages.content_hash>-<effective-date>
|
||||
```
|
||||
|
||||
`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-<SHA-256>
|
||||
```
|
||||
|
||||
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.
|
||||
+109
-33
@@ -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:<slug>
|
||||
// - 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<Parameters<BrainEngine['listPages']>[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 {
|
||||
|
||||
@@ -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<number> {
|
||||
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<string, string> {
|
||||
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<ConversationPageSnapshot> {
|
||||
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<Map<string, DurableExtractionOutcome>> {
|
||||
if (pages.length === 0) return new Map();
|
||||
const expected = new Map<string, string>();
|
||||
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<string, DurableExtractionOutcome>();
|
||||
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<boolean> {
|
||||
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<ReturnType<typeof extractFactsFromTurn>> = [];
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> => {
|
||||
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);
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
|
||||
+68
-18
@@ -215,20 +215,38 @@ const EXTRACTOR_SYSTEM = [
|
||||
|
||||
const MAX_TURN_TEXT_CHARS = 8000;
|
||||
|
||||
export async function extractFactsFromTurn(input: ExtractInput): Promise<ExtractedFact[]> {
|
||||
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<ExtractFactsOutcome> {
|
||||
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<Extract
|
||||
`(model=${model}); facts for this turn are likely lost. ` +
|
||||
`Raise the cap: gbrain config set facts.extraction_max_tokens <n>\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<Extract
|
||||
});
|
||||
}
|
||||
|
||||
return facts;
|
||||
return { ok: true, facts };
|
||||
}
|
||||
|
||||
/** Historical best-effort API retained for interactive callers. */
|
||||
export async function extractFactsFromTurn(input: ExtractInput): Promise<ExtractedFact[]> {
|
||||
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<string, unknown>).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<string, unknown>;
|
||||
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;
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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}
|
||||
|
||||
@@ -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<void> {
|
||||
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<void> {
|
||||
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 });
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -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<void>) | 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',
|
||||
|
||||
Reference in New Issue
Block a user