Merge JME Phase 2 — consolidator + dedup/supersede + nightly batch (audited R1+R2)

Jarvis-authored core (5843f60 upsertFact/consolidate, f3db341 Option-A
wiring) + operator-session audit fixes (76494d9 true-cosine dedup, nightly
batch, prune units; 2f38de4 empty-response retain). R1 FAIL -> all three
criticals fixed -> R2 PASS. Consolidation runs as the jme-consolidate cron
(02:45 MX) - one Haiku extraction per night over the settled episodic
window, cosine-dedup upsert, exact-window delete, 7d orphan sweep.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0186BhSFY9sFFHCPeFnpbvki

Author: Jarvis (Piotr)

src/memory/jme.test.ts

@@ -12,6 +12,15 @@
 import { describe, it, expect, beforeEach, vi } from "vitest";
 import Database from "better-sqlite3";
 
+// Phase 2 mocks (hoisted before dynamic import)
+const inferMock = vi.fn();
+vi.mock("../inference/adapter.js", () => ({
+  infer: (...args: unknown[]) => inferMock(...args),
+}));
+vi.mock("../inference/claude-sdk.js", () => ({
+  HAIKU_MODEL_ID: "claude-haiku-test",
+}));
+
 // ── Mock DB & dependencies ───────────────────────────────────────────────────
 
 // We create an in-memory DB and run the JME schema manually
@@ -286,3 +295,301 @@ describe("JME — stats", () => {
     expect(stats.factsWithEmbedding).toBe(0); // embed() returns null in tests
   });
 });
+
+// ── Phase 2: Consolidator + Dedup ─────────────────────────────────────────────
+
+describe("JME — upsertFact (dedup)", () => {
+  it("inserts a new fact when no near-duplicate exists (embed unavailable)", async () => {
+    const { upsertFact, jmeStats } = await getJme();
+
+    // embed() mock returns null → no vector comparison → plain insert
+    const outcome = await upsertFact({
+      sourceTask: "t1",
+      factText: "Fede prefers concise responses",
+      category: "preference",
+    });
+
+    expect(outcome).toBe("inserted");
+    expect(jmeStats().factsTotal).toBe(1);
+  });
+
+  it("skips fact when TRUE cosine similarity >= SKIP_THRESHOLD (near-identical)", async () => {
+    const { upsertFact, jmeStats, CONSOLIDATOR_SKIP_THRESHOLD } =
+      await getJme();
+    const {
+      embed: embedMock,
+      cosineSimilarity: cosSim,
+      deserializeEmbedding: deser,
+      serializeEmbedding: ser,
+    } = await import("./embeddings.js");
+
+    const vec = new Float32Array([1, 0, 0]);
+    const blob = Buffer.from(vec.buffer);
+
+    // Seed fact WITH a stored embedding
+    vi.mocked(embedMock).mockResolvedValue(vec);
+    vi.mocked(ser).mockReturnValue(blob);
+    await upsertFact({
+      sourceTask: "t1",
+      factText: "Fede likes coffee",
+      category: "preference",
+    });
+
+    // Second upsert: cosine reads near-identical everywhere (queryMemory's
+    // candidate scan AND the true-cosine re-check both go through cosSim)
+    vi.mocked(deser).mockReturnValue(vec);
+    vi.mocked(cosSim).mockReturnValue(CONSOLIDATOR_SKIP_THRESHOLD + 0.01);
+
+    const outcome = await upsertFact({
+      sourceTask: "t2",
+      factText: "Fede likes coffee",
+      category: "preference",
+    });
+
+    expect(outcome).toBe("skipped");
+    expect(jmeStats().factsTotal).toBe(1); // still only 1 fact
+
+    // restore defaults
+    vi.mocked(embedMock).mockResolvedValue(null);
+    vi.mocked(cosSim).mockReturnValue(0);
+    vi.mocked(ser).mockImplementation((v: Float32Array) =>
+      Buffer.from(v.buffer, v.byteOffset, v.byteLength),
+    );
+  });
+
+  it("supersedes the MATCHED row when cosine is between DEDUP and SKIP", async () => {
+    const {
+      upsertFact,
+      jmeStats,
+      CONSOLIDATOR_DEDUP_THRESHOLD,
+      CONSOLIDATOR_SKIP_THRESHOLD,
+    } = await getJme();
+    const {
+      embed: embedMock,
+      cosineSimilarity: cosSim,
+      deserializeEmbedding: deser,
+      serializeEmbedding: ser,
+    } = await import("./embeddings.js");
+
+    const vec = new Float32Array([1, 0, 0]);
+    const blob = Buffer.from(vec.buffer);
+
+    vi.mocked(embedMock).mockResolvedValue(vec);
+    vi.mocked(ser).mockReturnValue(blob);
+    await upsertFact({
+      sourceTask: "t1",
+      factText: "Fede uses Valle de Bravo to rest",
+      category: "event",
+    });
+    const seeded = mockDb
+      .prepare(`SELECT id FROM jme_facts LIMIT 1`)
+      .get() as { id: number };
+
+    const midSim =
+      (CONSOLIDATOR_DEDUP_THRESHOLD + CONSOLIDATOR_SKIP_THRESHOLD) / 2;
+    vi.mocked(deser).mockReturnValue(vec);
+    vi.mocked(cosSim).mockReturnValue(midSim);
+
+    const outcome = await upsertFact({
+      sourceTask: "t2",
+      factText: "Fede uses Valle de Bravo for rest and recovery",
+      category: "event",
+    });
+
+    expect(outcome).toBe("superseded");
+    // Old fact expired + new fact inserted = 2 rows total
+    expect(jmeStats().factsTotal).toBe(2);
+    // W2: the supersede must expire the MATCHED row (by id), not an arbitrary one
+    const oldRow = mockDb
+      .prepare(`SELECT expires_at FROM jme_facts WHERE id = ?`)
+      .get(seeded.id) as { expires_at: number | null };
+    expect(oldRow.expires_at).not.toBeNull();
+
+    // restore
+    vi.mocked(embedMock).mockResolvedValue(null);
+    vi.mocked(cosSim).mockReturnValue(0);
+    vi.mocked(ser).mockImplementation((v: Float32Array) =>
+      Buffer.from(v.buffer, v.byteOffset, v.byteLength),
+    );
+  });
+
+  it("NEVER skips on the FTS-only path — keyword overlap is not similarity (audit C2)", async () => {
+    const { upsertFact, jmeStats, writeFact } = await getJme();
+    // embed stays null (default): the fused score would be keyword-only, with
+    // the top FTS hit normalized to 1.0 — the exact condition that silently
+    // dropped unrelated facts pre-fix.
+    await writeFact({
+      sourceTask: "t1",
+      factText: "Fede prefers dark roast coffee in the morning",
+      category: "preference",
+    });
+
+    const outcome = await upsertFact({
+      sourceTask: "t2",
+      factText: "The coffee machine in the office broke yesterday",
+      category: "event",
+    });
+
+    expect(outcome).toBe("inserted");
+    expect(jmeStats().factsTotal).toBe(2);
+  });
+
+  it("clamps out-of-range confidence into [0,1] (audit W3)", async () => {
+    const { upsertFact } = await getJme();
+    await upsertFact({
+      sourceTask: "t1",
+      factText: "Fede runs a VPS with mission-control",
+      category: "project",
+      confidence: 7,
+    });
+    const row = mockDb
+      .prepare(`SELECT confidence FROM jme_facts LIMIT 1`)
+      .get() as { confidence: number };
+    expect(row.confidence).toBe(1);
+  });
+});
+
+describe("JME — consolidateAll (nightly batch)", () => {
+  beforeEach(() => {
+    inferMock.mockReset();
+  });
+
+  /** Insert a turn old enough for the consolidator's 30-min settle window. */
+  function insertSettledTurn(taskId: string, role: string, content: string) {
+    mockDb
+      .prepare(
+        `INSERT INTO jme_turns (task_id, role, content, channel, ts) VALUES (?, ?, ?, ?, ?)`,
+      )
+      .run(taskId, role, content, "telegram", Date.now() - 31 * 60 * 1000);
+  }
+
+  it("consolidates settled turns ACROSS tasks in one call and deletes exactly them", async () => {
+    const { consolidateAll, jmeStats } = await getJme();
+
+    insertSettledTurn("task-a", "user", "I prefer short answers");
+    insertSettledTurn("task-a", "jarvis", "Noted!");
+    insertSettledTurn("task-b", "user", "Vamos a despertar Pipesong");
+
+    inferMock.mockResolvedValueOnce({
+      content: JSON.stringify([
+        {
+          factText: "Fede prefers short answers",
+          category: "preference",
+          confidence: 0.9,
+        },
+      ]),
+    });
+
+    const result = await consolidateAll();
+
+    expect(result.turnsProcessed).toBe(3); // one batch, both tasks
+    expect(inferMock).toHaveBeenCalledTimes(1); // ONE Haiku call for the window
+    expect(result.factsExtracted).toBe(1);
+    expect(result.factsInserted).toBe(1);
+    expect(jmeStats().turnsTotal).toBe(0);
+    expect(jmeStats().factsTotal).toBe(1);
+  });
+
+  it("leaves turns younger than the settle window for the next run", async () => {
+    const { consolidateAll, writeEpisodic, jmeStats } = await getJme();
+
+    insertSettledTurn("task-old", "user", "settled message");
+    writeEpisodic({ taskId: "task-live", role: "user", content: "just now" });
+    inferMock.mockResolvedValueOnce({ content: "[]" });
+
+    const result = await consolidateAll();
+
+    expect(result.turnsProcessed).toBe(1);
+    expect(jmeStats().turnsTotal).toBe(1); // the fresh turn survives
+  });
+
+  it("returns zeros without calling Haiku when nothing is settled", async () => {
+    const { consolidateAll, writeEpisodic } = await getJme();
+
+    writeEpisodic({ taskId: "task-live", role: "user", content: "hi" });
+
+    const result = await consolidateAll();
+
+    expect(result.turnsProcessed).toBe(0);
+    expect(inferMock).not.toHaveBeenCalled();
+  });
+
+  it("a valid empty [] consumes the window (nothing durable is a valid verdict)", async () => {
+    const { consolidateAll, jmeStats } = await getJme();
+
+    insertSettledTurn("task-empty", "user", "hola");
+    inferMock.mockResolvedValueOnce({ content: "[]" });
+
+    const result = await consolidateAll();
+
+    expect(result.factsExtracted).toBe(0);
+    expect(jmeStats().turnsTotal).toBe(0); // consumed
+  });
+
+  it("an EMPTY Haiku response retains the turns (transient failure, not a verdict)", async () => {
+    const { consolidateAll, jmeStats } = await getJme();
+
+    insertSettledTurn("task-blank", "user", "hola");
+    inferMock.mockResolvedValueOnce({ content: "" });
+
+    const result = await consolidateAll();
+
+    expect(result.factsExtracted).toBe(0);
+    expect(jmeStats().turnsTotal).toBe(1); // NOT deleted — retried next run
+  });
+
+  it("malformed JSON leaves the turns IN PLACE for tomorrow's retry", async () => {
+    const { consolidateAll, jmeStats } = await getJme();
+
+    insertSettledTurn("task-bad", "user", "hello");
+    inferMock.mockResolvedValueOnce({ content: "not valid json {{{" });
+
+    const result = await consolidateAll();
+
+    expect(result.factsExtracted).toBe(0);
+    expect(jmeStats().turnsTotal).toBe(1); // NOT deleted — unconsumed data
+  });
+});
+
+// ---------------------------------------------------------------------------
+// pruneStaleTurns — global 7d sweep
+// ---------------------------------------------------------------------------
+describe("JME — pruneStaleTurns", () => {
+  beforeEach(() => {
+    vi.resetModules();
+    vi.clearAllMocks();
+  });
+
+  it("removes turns older than TURN_RETENTION_DAYS and leaves recent ones", async () => {
+    const { writeEpisodic, pruneStaleTurns, TURN_RETENTION_DAYS } = await getJme();
+
+    // Write two turns — one recent, one >7d old
+    writeEpisodic({ taskId: "task-recent", role: "user", content: "fresh" });
+
+    // Manually backdate a turn by injecting directly into mockDb (column is `ts`)
+    // Same unit production writes: writeEpisodic stores Date.now() MILLISECONDS
+    // (audit C1: the old seconds-based fixture masked the units mismatch).
+    const oldTimestamp = Date.now() - (TURN_RETENTION_DAYS + 1) * 86_400_000;
+    mockDb.prepare(
+      `INSERT INTO jme_turns (task_id, role, content, channel, ts) VALUES (?, ?, ?, ?, ?)`,
+    ).run("task-old", "user", "stale content", "telegram", oldTimestamp);
+
+    const deleted = pruneStaleTurns();
+
+    // The stale turn should be deleted, the recent one preserved
+    expect(deleted).toBe(1);
+    const remaining = mockDb.prepare(`SELECT task_id FROM jme_turns`).all() as { task_id: string }[];
+    const ids = remaining.map((r) => r.task_id);
+    expect(ids).toContain("task-recent");
+    expect(ids).not.toContain("task-old");
+  });
+
+  it("returns 0 when no stale turns exist", async () => {
+    const { writeEpisodic, pruneStaleTurns } = await getJme();
+
+    writeEpisodic({ taskId: "task-new", role: "user", content: "brand new" });
+
+    const deleted = pruneStaleTurns();
+    expect(deleted).toBe(0);
+  });
+});