build-digests.mjs

// build-digests.mjs — the WRITE path (the recursive fold).
//
// Consolidation write-path for an append-only agent-memory store.
//
// WHAT IT DOES
//   Reads a memory store of atomic "observations" (READ-ONLY) and maintains one
//   evolving "digest" per (project, type) cell — e.g. "my-app / gotcha". Each
//   run folds the NEW observations (those created after the cell's watermark)
//   into the existing digest using an LLM, bumps the version, and advances the
//   watermark. Re-running with no new observations is a no-op. This is the
//   recursive loop:
//       digest(n) = merge(digest(n-1), new observations)
//
// THE MERGE
//   mergeDigest() shells out to `claude -p` over your Claude subscription (no
//   API key, no metered cost). It rewrites the digest to consolidate
//   duplicates, resolve contradictions toward newer information, and compress —
//   producing evolving understanding, not an append log. The merge step is a
//   single swappable seam: replace mergeDigest() to use any provider.
//
// SAFETY
//   - Source DB is opened readonly: it cannot be mutated by this script.
//   - The digest DB is a SEPARATE file; the source store is never written.
//
// USAGE
//   bun build-digests.mjs                 # all projects
//   bun build-digests.mjs my-app          # one project
//   MERGE_MODEL=haiku bun build-digests.mjs my-app   # cheaper/faster backfill
//
// CONFIG (env)
//   CLAUDE_MEM_DB   path to the source observation store
//                   (default: ~/.claude-mem/claude-mem.db)
//   DIGEST_DB       path to the digest output DB (default: ./digests.db)
//   MERGE_MODEL     model passed to `claude -p` (default: sonnet)

import { Database } from "bun:sqlite";
import { homedir } from "os";
import { join } from "path";

const SOURCE_DB =
  process.env.CLAUDE_MEM_DB || join(homedir(), ".claude-mem", "claude-mem.db");
const DIGEST_DB = process.env.DIGEST_DB || join(process.cwd(), "digests.db");

// The eight memory-type facets a digest is keyed on. These match claude-mem's
// `concepts` enum; adjust to whatever categorical tags your store uses.
const CORE_TYPES = [
  "what-changed",
  "how-it-works",
  "pattern",
  "problem-solution",
  "gotcha",
  "why-it-exists",
  "trade-off",
  "decision",
];

// Cap on how many new observations we fold into a cell per run. Bounds the
// prompt size so a cold, high-volume cell doesn't send one giant request.
// Backlog drains over repeated runs; the watermark advances to the newest
// observation actually folded each time, so nothing is skipped or double-counted.
const MAX_NEW_PER_RUN = 40;

// Hard size budget per digest cell, in characters (~4 chars/token). The digest
// is a tight executive summary injected at the START of future sessions, not an
// exhaustive record — so it must stay small regardless of how many observations
// fold into it. ~2800 chars ≈ ~700 tokens; a whole 8-cell project ≈ ~5-6K tokens.
//
// NOTE: this budget is enforced in the PROMPT, not in code. A model treats it as
// a strong suggestion, not a hard stop. For a real guarantee, truncate after the
// merge. See README "Known limitations".
const DIGEST_BUDGET_CHARS = 2800;

// Optional CLI filter: only build cells for this project.
const PROJECT_FILTER = process.argv[2] || null;

// Merge model. Defaults to sonnet (best consolidation quality). Override with
// MERGE_MODEL=haiku for heavy backfills where the subscription rolling limit is
// the binding constraint — Haiku burns far less of the limit per call, and (in
// practice) held the size budget better, at the cost of looser recency resolution.
//
// MERGE_MODEL=mock uses a deterministic, no-LLM merge so you can run the whole
// pipeline (watermark, drain, resume, economics) offline with no `claude` CLI
// and no API cost. The mock output is NOT a real consolidation — it just proves
// the plumbing. Use it with the seeded example store to see the shape end-to-end.
const MERGE_MODEL = process.env.MERGE_MODEL || "sonnet";
const USE_MOCK = MERGE_MODEL === "mock";

const src = new Database(SOURCE_DB, { readonly: true });
const out = new Database(DIGEST_DB);

out.run(`
  CREATE TABLE IF NOT EXISTS digests (
    id               INTEGER PRIMARY KEY,
    project          TEXT NOT NULL,
    type             TEXT NOT NULL,
    digest           TEXT NOT NULL,
    source_count     INTEGER NOT NULL,
    source_max_epoch INTEGER NOT NULL,
    version          INTEGER NOT NULL,
    updated_at_epoch INTEGER NOT NULL,
    UNIQUE(project, type)
  )
`);

// ---------------------------------------------------------------------------
// Read + cluster observations into (project, type) cells.
// ---------------------------------------------------------------------------
function parseFacts(raw) {
  if (!raw) return [];
  try {
    const arr = JSON.parse(raw);
    return Array.isArray(arr) ? arr.filter((f) => typeof f === "string") : [];
  } catch {
    return [];
  }
}

function loadCells() {
  const rows = src
    .query(
      `SELECT project, concepts, title, facts, narrative, created_at_epoch
         FROM observations`
    )
    .all();

  const cells = new Map(); // "project type" -> [{epoch,title,facts[],narrative}]
  for (const r of rows) {
    if (!r.project) continue;
    if (PROJECT_FILTER && r.project !== PROJECT_FILTER) continue;
    let concepts = [];
    try {
      concepts = JSON.parse(r.concepts || "[]");
    } catch {
      concepts = [];
    }
    for (const c of concepts) {
      const type = String(c).toLowerCase().trim();
      if (!CORE_TYPES.includes(type)) continue;
      const key = `${r.project} ${type}`;
      let bucket = cells.get(key);
      if (!bucket) cells.set(key, (bucket = []));
      bucket.push({
        epoch: r.created_at_epoch || 0,
        title: r.title || "",
        facts: parseFacts(r.facts),
        narrative: r.narrative || "",
      });
    }
  }
  return cells;
}

// ---------------------------------------------------------------------------
// THE SWAPPABLE SEAM — backed by an LLM merge. Signature:
//   (project, type, existingDigest, newObservations) -> string
// Replace the body of mergeDigest() to use any provider you like.
// ---------------------------------------------------------------------------
const TYPE_GLOSS = {
  "what-changed": "concrete changes made",
  "how-it-works": "how things work / architecture",
  pattern: "reusable approaches and patterns",
  "problem-solution": "problems hit and how they were solved",
  gotcha: "pitfalls and surprises to avoid",
  "why-it-exists": "rationale and intent behind decisions",
  "trade-off": "trade-offs weighed",
  decision: "decisions made",
};

function buildPrompt(project, type, existingDigest, newObservations) {
  const obsText = newObservations
    .map((o) => {
      const facts = o.facts.length
        ? o.facts.map((f) => `    - ${f}`).join("\n")
        : "    (no discrete facts)";
      return `- ${o.title}\n${facts}`;
    })
    .join("\n");

  return `You maintain an evolving knowledge digest for one slice of an engineering memory system.

PROJECT: ${project}
CATEGORY: ${type} (${TYPE_GLOSS[type] || type})

EXISTING DIGEST (may be empty on the first build):
${existingDigest ? existingDigest : "(none yet)"}

NEW OBSERVATIONS to fold in (each is a distilled memory from a work session):
${obsText}

Rewrite the digest so it incorporates the new observations. Requirements:
- HARD SIZE LIMIT: the rewritten digest MUST stay under ${DIGEST_BUDGET_CHARS} characters. This is a tight executive summary injected at the start of future work sessions, not an exhaustive log. Staying under budget is more important than retaining every detail.
- When you are near or over the limit, consolidate aggressively: generalize repetitive specifics into single statements, keep only the most load-bearing / current / reusable points, and DROP superseded, one-off, or stale items. Newer information wins over older.
- Merge duplicates and near-duplicates; resolve contradictions in favor of the newest information.
- Preserve exact identifiers (file paths, commands, version numbers, error strings) ONLY for the handful of most important, still-current items — not for everything.
- Use markdown bullets grouped under a few short **bold subheadings**.
- Output ONLY the digest body. No preamble, no "Here is", no closing remarks.`;
}

// Strip ANTHROPIC_API_KEY for child `claude` calls so the CLI authenticates with
// the Claude subscription (OAuth login) instead of API credits. With the key
// present, `claude -p` bills the API credit balance — a separate, depletable
// pool from the subscription's allowance. An EMPTY key still wins its slot, so
// the key must be REMOVED from the child environment, not blanked.
const SUBSCRIPTION_ENV = { ...process.env };
delete SUBSCRIPTION_ENV.ANTHROPIC_API_KEY;

// Deterministic, no-LLM merge for offline demos and plumbing tests. Folds new
// observation titles/facts into the running digest as bullets, dedupes exact
// lines, and truncates to budget. Not a real consolidation — just proves the
// watermark/drain/resume/economics path works without a provider.
function mockMerge(type, existingDigest, newObservations) {
  const lines = new Set(
    existingDigest
      .split("\n")
      .map((l) => l.trim())
      .filter((l) => l.startsWith("- "))
  );
  for (const o of newObservations) {
    if (o.title) lines.add(`- ${o.title}`);
    for (const f of o.facts) lines.add(`  - ${f}`);
  }
  let body = `**${type}**\n` + [...lines].join("\n");
  if (body.length > DIGEST_BUDGET_CHARS) body = body.slice(0, DIGEST_BUDGET_CHARS);
  return body;
}

async function mergeDigest(project, type, existingDigest, newObservations) {
  if (USE_MOCK) return mockMerge(type, existingDigest, newObservations);
  const prompt = buildPrompt(project, type, existingDigest, newObservations);
  // `claude -p` reads the prompt on stdin and prints the result on stdout.
  // On Windows, `cmd /c` resolves the claude.cmd shim; on macOS/Linux you can
  // call "claude" directly.
  const isWin = process.platform === "win32";
  const cmd = isWin
    ? ["cmd", "/c", "claude", "-p", "--model", MERGE_MODEL]
    : ["claude", "-p", "--model", MERGE_MODEL];
  const proc = Bun.spawn(cmd, {
    stdin: new TextEncoder().encode(prompt),
    stdout: "pipe",
    stderr: "pipe",
    env: SUBSCRIPTION_ENV,
  });
  const text = await new Response(proc.stdout).text();
  await proc.exited;
  if (proc.exitCode !== 0) {
    const err = await new Response(proc.stderr).text();
    throw new Error(`claude exited ${proc.exitCode}: ${err.slice(0, 300)}`);
  }
  return text.trim();
}

// ---------------------------------------------------------------------------
// Main fold (async — one merge call per changed batch, sequential).
// ---------------------------------------------------------------------------
const getDigest = out.query(
  `SELECT digest, source_count, source_max_epoch, version
     FROM digests WHERE project = ? AND type = ?`
);
const upsert = out.query(`
  INSERT INTO digests
    (project, type, digest, source_count, source_max_epoch, version, updated_at_epoch)
  VALUES (?, ?, ?, ?, ?, ?, ?)
  ON CONFLICT(project, type) DO UPDATE SET
    digest=excluded.digest,
    source_count=excluded.source_count,
    source_max_epoch=excluded.source_max_epoch,
    version=excluded.version,
    updated_at_epoch=excluded.updated_at_epoch
`);

const cells = loadCells();
let batchesDone = 0,
  cellsDrained = 0,
  cellsNoop = 0,
  failed = 0;

console.log(
  `Cells to consider: ${cells.size}` +
    (PROJECT_FILTER ? ` (filtered to project "${PROJECT_FILTER}")` : "") +
    `  [model: ${MERGE_MODEL}]`
);

// DRAIN MODE: for each cell, keep folding MAX_NEW_PER_RUN-observation batches
// into the running digest until no observations remain past the watermark. The
// digest is persisted after EVERY batch, so an interrupted run is fully
// resumable — re-running picks up from the last persisted watermark.
for (const [key, observations] of cells) {
  const sep = key.indexOf(" ");
  const project = key.slice(0, sep);
  const type = key.slice(sep + 1);

  const sorted = observations.slice().sort((a, b) => a.epoch - b.epoch);
  const existing = getDigest.get(project, type);

  let digest = existing ? existing.digest : "";
  let watermark = existing ? existing.source_max_epoch : 0;
  let sourceCount = existing ? existing.source_count : 0;
  let version = existing ? existing.version : 0;

  const pendingAtStart = sorted.filter((o) => o.epoch > watermark).length;
  if (pendingAtStart === 0) {
    cellsNoop++;
    continue;
  }

  console.log(`\n${project}/${type}  draining ${pendingAtStart} observations:`);
  let cellFailed = false;

  while (true) {
    const fresh = sorted.filter((o) => o.epoch > watermark);
    if (fresh.length === 0) break;
    const batch = fresh.slice(0, MAX_NEW_PER_RUN);
    const newMax = batch.reduce((m, o) => Math.max(m, o.epoch), watermark);

    process.stdout.write(`  +${batch.length} (${fresh.length} left) ... `);
    try {
      const merged = await mergeDigest(project, type, digest, batch);
      if (!merged) throw new Error("empty digest returned");
      digest = merged;
      watermark = newMax;
      sourceCount += batch.length;
      version += 1;
      upsert.run(project, type, digest, sourceCount, watermark, version, Date.now());
      batchesDone++;
      console.log(`v${version} (${digest.length} chars)`);
    } catch (err) {
      failed++;
      cellFailed = true;
      console.log(`FAILED: ${err.message}`);
      break; // watermark not advanced — cell resumes on next run
    }
  }
  if (!cellFailed) cellsDrained++;
}

console.log(
  `\nDone. ${cellsDrained} cells drained, ${cellsNoop} already current, ` +
    `${batchesDone} merge calls, ${failed} failures`
);
console.log(`Digest DB: ${DIGEST_DB}`);

src.close();
out.close();