chore: recall consolidate defense-in-depth hardening (child re-clamp, NULL-lineage guard, timeout count accuracy)

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<string | null>;
+  /** 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<void> {
   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);

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 `<n>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<ConsolidateApplyR
     });
     return Promise.resolve(JSON.parse(out) as ConsolidateApplyResult);
   } catch (err: any) {
+    // On timeout the child is SIGTERM'd after committing N per-cluster transactions;
+    // recover those counts from its captured stderr so the summary doesn't understate.
+    const progress = parseConsolidateProgress(String(err?.stderr ?? ''));
     return Promise.resolve({
-      written: 0, demoted: 0, redactions: [], skipped: [],
+      written: progress?.written ?? 0, demoted: progress?.demoted ?? 0, redactions: [], skipped: [],
       error: `consolidate engine failed: ${(err?.message || String(err)).slice(0, 200)}`,
     });
   }
@@ -198,6 +224,12 @@ export async function runConsolidate(
 
   if (applied.error) {
     console.error(`\nConsolidation failed: ${applied.error}`);
+    if (applied.written > 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 };
   }

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 },
+    ]);
+  });
+});

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();
+  });
 });