diff --git a/hooks/lib/consolidate-core.ts b/hooks/lib/consolidate-core.ts index f7bd59f..b1980e3 100644 --- a/hooks/lib/consolidate-core.ts +++ b/hooks/lib/consolidate-core.ts @@ -65,8 +65,16 @@ export interface ConsolidateApplyResult { /** Injected model behavior — defaults to the cheap cascade; tests pass a fake. */ export interface ConsolidateDeps { summarize: (prompt: string) => Promise; + /** Called with the running cumulative totals after each cluster commits, so a + * parent that kills us on timeout can still report the work we landed. */ + onProgress?: (progress: { written: number; demoted: number }) => void; } +/** stderr marker the child prints after each committed cluster. The parent + * (src/commands/consolidate.ts) matches the same literal to recover committed + * counts when it kills us on timeout — restated there per the hooks/src split. */ +export const PROGRESS_PREFIX = 'RECALL_CONSOLIDATE_PROGRESS '; + /** * Build the summarization prompt for one cluster. Pure. The output structure * (## ONE SENTENCE SUMMARY + ## MAIN IDEAS) is required so evaluateQuality's @@ -122,11 +130,15 @@ export async function applyConsolidation( const result: ConsolidateApplyResult = { written: 0, demoted: 0, redactions: [], skipped: [] }; const db = openDb(dbPath); try { - const hasProvenance = columnExists(db, 'loa_entries', 'provenance'); - const hasSourceIds = columnExists(db, 'loa_entries', 'source_ids'); - const cols = ['title', 'description', 'fabric_extract', 'project', 'tags', 'importance']; - if (hasProvenance) cols.push('provenance'); - if (hasSourceIds) cols.push('source_ids'); + // Lineage columns are guaranteed by migration 13 (#140). If they are absent + // the schema is unmigrated/corrupt — fail loudly rather than silently writing + // NULL provenance/source_ids and losing the lineage we just computed. + if (!columnExists(db, 'loa_entries', 'provenance') || !columnExists(db, 'loa_entries', 'source_ids')) { + throw new Error( + 'loa_entries is missing provenance/source_ids columns — run `recall migrate` (schema must be >= v13) before consolidating', + ); + } + const cols = ['title', 'description', 'fabric_extract', 'project', 'tags', 'importance', 'provenance', 'source_ids']; const insertSql = `INSERT INTO loa_entries (${cols.join(', ')}) VALUES (${cols.map(() => '?').join(', ')})`; const insertLoa = db.prepare(insertSql); @@ -168,21 +180,26 @@ export async function applyConsolidation( cluster.project, 'consolidated,derived', DERIVED_LOA_IMPORTANCE, + 'derived', + sourceIds, ]; - if (hasProvenance) values.push('derived'); - if (hasSourceIds) values.push(sourceIds); - // 4) Write + demote atomically. Importance only ever moves to the planner's - // clampImportance target. Sources are demoted, never deleted. + // 4) Write + demote atomically. Sources are demoted, never deleted. + // Re-clamp in the child (defense-in-depth): MIN(importance, …) so a demote + // can only ever LOWER importance — a directly-piped payload can't raise it — + // and MAX(1, …) re-pins the absolute floor regardless of the supplied value. const writeCluster = db.transaction(() => { insertLoa.run(...values); - const demote = db.prepare(`UPDATE ${cluster.table} SET importance = ? WHERE id = ?`); + const demote = db.prepare( + `UPDATE ${cluster.table} SET importance = MAX(1, MIN(importance, CAST(? AS INTEGER))) WHERE id = ?`, + ); for (const r of cluster.records) demote.run(r.newImportance, r.id); }); writeCluster(); result.written += 1; result.demoted += cluster.records.length; + deps.onProgress?.({ written: result.written, demoted: result.demoted }); } result.redactions = [...redactions]; @@ -228,7 +245,13 @@ async function main(): Promise { const result = await applyConsolidation( resolveDbPath(), payload.clusters ?? [], - { summarize: runExtractionCascade }, + { + summarize: runExtractionCascade, + // Cumulative progress to stderr — the parent reads the last marker to report + // committed work if it has to SIGTERM us on timeout (stdout carries only the + // final JSON result, so progress must not pollute it). + onProgress: (p) => process.stderr.write(`${PROGRESS_PREFIX}${JSON.stringify(p)}\n`), + }, ); process.stdout.write(JSON.stringify(result)); process.exit(0); diff --git a/src/commands/consolidate.ts b/src/commands/consolidate.ts index 0c3f9d0..e0dd711 100644 --- a/src/commands/consolidate.ts +++ b/src/commands/consolidate.ts @@ -58,6 +58,29 @@ export interface ConsolidateRunResult { applied: ConsolidateApplyResult | null; } +/** stderr marker the hook-side child emits after each committed cluster. Restated + * here (not imported) because src cannot cross the hooks/src boundary — the + * literal must stay in lockstep with PROGRESS_PREFIX in consolidate-core.ts. */ +const PROGRESS_PREFIX = 'RECALL_CONSOLIDATE_PROGRESS '; + +/** Recover the child's last cumulative {written, demoted} from its captured + * stderr. Used when the subprocess is killed on timeout so the summary reflects + * the per-cluster transactions it committed before SIGTERM. null if none found. */ +export function parseConsolidateProgress(stderr: string): { written: number; demoted: number } | null { + let latest: { written: number; demoted: number } | null = null; + for (const line of stderr.split('\n')) { + const at = line.indexOf(PROGRESS_PREFIX); + if (at === -1) continue; + try { + const parsed = JSON.parse(line.slice(at + PROGRESS_PREFIX.length)); + if (typeof parsed?.written === 'number' && typeof parsed?.demoted === 'number') { + latest = { written: parsed.written, demoted: parsed.demoted }; + } + } catch { /* ignore a partially-flushed marker line */ } + } + return latest; +} + /** Parse a `d` duration into days; null on malformed input. */ function parseDays(value: string): number | null { const match = value.match(/^(\d+)d$/); @@ -99,8 +122,11 @@ function spawnConsolidateChild(plan: ConsolidatePlan): Promise 0 || applied.demoted > 0) { + console.error( + `Partial progress before failure: wrote ${applied.written.toLocaleString()} derived summary(ies), ` + + `demoted ${applied.demoted.toLocaleString()} source record(s).` + ); + } process.exitCode = 1; return { plan, applied }; } diff --git a/tests/hooks/consolidate-core.test.ts b/tests/hooks/consolidate-core.test.ts index 5ca4980..5554f8b 100644 --- a/tests/hooks/consolidate-core.test.ts +++ b/tests/hooks/consolidate-core.test.ts @@ -10,6 +10,10 @@ // The model cascade is injected as a fake, so no real `claude`/Ollama is invoked. import { describe, test, expect, beforeEach, afterEach } from 'bun:test'; +import { mkdtempSync, rmSync } from 'fs'; +import { join } from 'path'; +import { tmpdir } from 'os'; +import { Database } from 'bun:sqlite'; import { setupTestDb, teardownTestDb } from '../helpers/setup'; import { getDb } from '../../src/db/connection'; import { addDecision } from '../../src/lib/memory'; @@ -41,12 +45,12 @@ afterEach(() => { teardownTestDb(); }); function seedDecisions(texts: string[], importance = 3): number[] { return texts.map((t) => addDecision({ decision: t, status: 'active', project: 'p', importance })); } -function clusterFor(ids: number[]): ConsolidateInputCluster { +function clusterFor(ids: number[], newImportance = 1, window = '2020-01-01'): ConsolidateInputCluster { return { table: 'decisions', project: 'p', - window: '2020-01-01', - records: ids.map((id) => ({ id, text: `decision ${id}`, newImportance: 1 })), + window, + records: ids.map((id) => ({ id, text: `decision ${id}`, newImportance })), }; } function loaRows(): Array<{ provenance: string | null; source_ids: string | null; fabric_extract: string; importance: number }> { @@ -117,3 +121,73 @@ describe('applyConsolidation — write + demote', () => { for (const id of ids) expect(importanceOf(id)).toBe(3); }); }); + +// --------------------------------------------------------------------------- +// Defense-in-depth hardening (issue #145). +// --------------------------------------------------------------------------- + +describe('applyConsolidation — child re-clamps the demote target (#145.1)', () => { + const summarize = async () => VALID_SUMMARY; + + test('a hostile newImportance can only LOWER importance — never raises it', async () => { + // Sources at importance 4; payload tries to RAISE them to 9. + const ids = seedDecisions(['decision one', 'decision two', 'decision three'], 4); + + const result = await applyConsolidation(dbPath, [clusterFor(ids, 9)], { summarize }); + + expect(result.written).toBe(1); + // MIN(importance, 9) = 4 → demotion never raises above the current value. + for (const id of ids) expect(importanceOf(id)).toBe(4); + }); + + test('an out-of-range newImportance is re-pinned to the absolute floor of 1', async () => { + const ids = seedDecisions(['decision one', 'decision two', 'decision three'], 4); + + const result = await applyConsolidation(dbPath, [clusterFor(ids, 0)], { summarize }); + + expect(result.written).toBe(1); + // MAX(1, MIN(4, 0)) = 1 → never below the floor, even from a sub-floor payload. + for (const id of ids) expect(importanceOf(id)).toBe(1); + }); +}); + +describe('applyConsolidation — NULL-lineage guard (#145.2)', () => { + test('fails loudly when loa_entries lacks provenance/source_ids columns', async () => { + // A bare DB whose loa_entries predates migration 13 (#140) — no lineage cols. + const dir = mkdtempSync(join(tmpdir(), 'recall-unmigrated-')); + const barePath = join(dir, 'bare.db'); + const bare = new Database(barePath); + bare.exec('CREATE TABLE loa_entries (id INTEGER PRIMARY KEY, title TEXT, fabric_extract TEXT)'); + bare.close(); + try { + const ids = [1, 2, 3]; + await expect( + applyConsolidation(barePath, [clusterFor(ids)], { summarize: async () => VALID_SUMMARY }), + ).rejects.toThrow(/provenance\/source_ids/); + } finally { + rmSync(dir, { recursive: true, force: true }); + } + }); +}); + +describe('applyConsolidation — progress callback (#145.3)', () => { + test('emits cumulative {written, demoted} after each committed cluster', async () => { + const a = seedDecisions(['a one', 'a two', 'a three']); + const b = seedDecisions(['b one', 'b two']); + const progress: Array<{ written: number; demoted: number }> = []; + + const result = await applyConsolidation( + dbPath, + [clusterFor(a, 1, '2020-01-01'), clusterFor(b, 1, '2020-02-01')], + { summarize: async () => VALID_SUMMARY, onProgress: (p) => progress.push(p) }, + ); + + expect(result.written).toBe(2); + // Cumulative totals, one event per committed cluster — exactly what the parent + // reads from the last stderr marker to report committed work on timeout. + expect(progress).toEqual([ + { written: 1, demoted: 3 }, + { written: 2, demoted: 5 }, + ]); + }); +}); diff --git a/tests/lib/consolidate.test.ts b/tests/lib/consolidate.test.ts index 23abd60..c797d62 100644 --- a/tests/lib/consolidate.test.ts +++ b/tests/lib/consolidate.test.ts @@ -15,6 +15,7 @@ import { addDecision, addLearning, createLoaEntry } from '../../src/lib/memory'; import { planConsolidate, type ConsolidatePlan } from '../../src/lib/consolidate'; import { runConsolidate, + parseConsolidateProgress, type ConsolidateApplyResult, } from '../../src/commands/consolidate'; import type { LineageStatus } from '../../src/lib/dedup'; @@ -184,4 +185,46 @@ describe('runConsolidate — dry-run / execute gate', () => { expect(process.exitCode).toBe(1); expect(spy.calls.length).toBe(0); }); + + test('surfaces partial progress when apply errors after committing work (#145.3)', async () => { + oldDecision('alpha'); oldDecision('beta'); oldDecision('gamma'); + const applied: ConsolidateApplyResult = { + written: 2, demoted: 5, redactions: [], skipped: [], + error: 'consolidate engine failed: timed out', + }; + + const result = await runConsolidate({ execute: true }, { apply: async () => applied }); + + expect(process.exitCode).toBe(1); + const out = logged.join('\n'); + expect(out).toContain('Consolidation failed: consolidate engine failed: timed out'); + expect(out).toContain('Partial progress before failure: wrote 2 derived summary(ies), demoted 5 source record(s).'); + expect(result!.applied).toBe(applied); + }); +}); + +// parseConsolidateProgress — recover committed counts from a killed child's stderr (#145.3). +describe('parseConsolidateProgress', () => { + const PREFIX = 'RECALL_CONSOLIDATE_PROGRESS '; + + test('returns the LAST cumulative marker, ignoring surrounding stderr noise', () => { + const stderr = [ + 'some cascade diagnostic line', + `${PREFIX}{"written":1,"demoted":3}`, + 'another noisy line from the nested model call', + `${PREFIX}{"written":2,"demoted":5}`, + ].join('\n'); + expect(parseConsolidateProgress(stderr)).toEqual({ written: 2, demoted: 5 }); + }); + + test('matches a marker even when other text precedes it on the line', () => { + expect(parseConsolidateProgress(`2026-01-01 worker: ${PREFIX}{"written":7,"demoted":9}`)) + .toEqual({ written: 7, demoted: 9 }); + }); + + test('ignores a malformed/partially-flushed marker and returns null when none parse', () => { + expect(parseConsolidateProgress(`${PREFIX}{"written":1,"demo`)).toBeNull(); + expect(parseConsolidateProgress('no markers here at all')).toBeNull(); + expect(parseConsolidateProgress('')).toBeNull(); + }); });