diff --git a/docs/specs/2026-04-28-memory-claude-mem-parity-design.md b/docs/specs/2026-04-28-memory-claude-mem-parity-design.md new file mode 100644 index 0000000..ee54389 --- /dev/null +++ b/docs/specs/2026-04-28-memory-claude-mem-parity-design.md @@ -0,0 +1,416 @@ +# CCE Memory: claude-mem Feature Parity + +**Status:** approved by product, in implementation +**Branch:** `feature/memory-claude-mem-parity` +**Date:** 2026-04-28 + +## Goal + +Bring CCE's cross-session memory to feature parity with `claude-mem`'s +production-grade capture pipeline, while keeping CCE's per-project, +local-first, no-required-external-services posture. + +Today CCE relies on a single `SessionStart` hook plus manual MCP +`record_decision` / `record_code_area` calls. Sessions are written to +per-project JSON files. Effectively, memory only grows when an agent +explicitly chooses to record — which means most sessions leave no trace. + +## Approved decisions + +Six design decisions were locked during brainstorming: + +1. **Capture model.** Auto-capture from hooks **plus** explicit MCP tools. + A `source` column (`manual` | `auto` | `migrated`) distinguishes them so + recall can rank manual entries higher. +2. **Compression timing.** A background worker inside `cce serve` (the + long-running per-project MCP server process) drains a + `pending_compressions` queue on a 5–10 s tick. Hooks themselves stay + thin appenders. +3. **Summary granularity.** Per-turn summaries plus a per-session + rollup. Maps directly onto the three retrieval layers. +4. **Recall MCP surface.** Extend the existing `session_recall(topic)` + to return compact-index hits (backward compatible with this project's + `CLAUDE.md` documentation), and add two new tools + `session_timeline(session_id)` and `session_event(event_id)` for + layer-2 and layer-3 drill-down. +5. **Migration of existing JSON sessions.** A one-shot + `cce sessions migrate` CLI command. Idempotent. User-invoked, not + auto-triggered on startup. +6. **Dashboard scope (v1).** Three panels: sessions list, session + timeline (drill-into one session), decisions search (FTS5, faceted by + `source`). Hot-files / queue-health views deferred to a follow-up. + +## Compressor tiering (locked separately) + +Per the user's "no Ollama on this machine, keep things small" +preference: + +``` +Tier 1 (default): extractive summarisation using BAAI/bge-small-en-v1.5. + The model is already loaded in cce serve for the index; + reusing it adds zero new dependencies, zero extra RAM. + Algorithm: sentence-split the turn, embed each sentence + with bge-small, pick the top-K closest to the turn's + centroid, concatenate. +Tier 2 (optional): Ollama if running. Existing compressor.py path. Opt-in. +Tier 3 (fallback): truncation. Used if the embedder isn't ready yet + (e.g. very early SessionStart before the model loads). +``` + +Extractive output is always real source text — no hallucination. That +matters for a memory system whose summaries survive across sessions. + +## Architecture + +``` + ┌─────────────────────────────────────┐ + │ Claude Code (per-project) │ + 5 hooks ───────────────► │ SessionStart │ + (settings.json) │ UserPromptSubmit │ + │ PostToolUse │ thin shell: + │ Stop │ ~50ms each, ─────► HTTP POST + │ SessionEnd │ no LLM │ to local + └─────────────────────────────────────┘ │ serve_http + ▼ + ┌───────────────────────────────────────────────────────────────────────────────┐ + │ cce serve (already running per-project — adds 2 things) │ + │ │ + │ ┌─ HTTP /hooks/ ─► append raw event to memory.db (1-2 ms) │ + │ │ │ + │ └─ async tick loop (5–10s) ─► drain pending_compressions queue │ + │ ┌──────────────────────────────┐ │ + │ │ extractive (bge-small) │ │ + │ │ → Ollama (opt-in) │ │ + │ │ → truncation fallback │ │ + │ └──────────────────────────────┘ │ + │ │ + │ MCP tools: session_recall (extended), session_timeline (new), │ + │ session_event (new), record_decision, record_code_area │ + └───────────────────────────────────────────────────────────────────────────────┘ + │ + ▼ + ~/.cce/projects//memory.db (per-project SQLite, FTS5) + │ + ▼ + Dashboard /memory pages (3 panels) +``` + +### Invariants + +- Hooks are thin shells; zero latency from LLM work. +- Compression is decoupled from hooks. If it crashes, capture continues. +- Per-project SQLite means two projects cannot collide. Backup = copy + one file. +- The existing `compressor.py` abstraction is reused. + +## Hook contract + +Each Claude Code hook becomes a thin shell that POSTs JSON to +`http://127.0.0.1:/hooks/`. The port is written by +`cce serve` to `~/.cce/projects//serve.port` on startup. + +Hook script (`~/.cce/hooks/cce_hook.sh`, ~15 lines): + +```sh +#!/bin/sh +# Claude Code hooks pipe JSON on stdin. Forward to cce serve. +# Failure is silent — memory is best-effort, never breaks the user's flow. +PORT=$(cat ~/.cce/projects/$(basename "$PWD")/serve.port 2>/dev/null) || exit 0 +curl -sf -m 1 -X POST -H "Content-Type: application/json" \ + --data-binary @- "http://127.0.0.1:${PORT}/hooks/$1" >/dev/null 2>&1 \ + || true +``` + +| Hook | Payload | Action in `cce serve` | +|---|---|---| +| `SessionStart` | `{session_id, project, started_at}` | Insert row in `sessions`. Show CCE status (existing behaviour preserved). | +| `UserPromptSubmit` | `{session_id, prompt_number, prompt_text, timestamp}` | Insert into `prompts`. Enqueue compression for the *previous* turn (if any). | +| `PostToolUse` | `{session_id, prompt_number, tool_name, tool_input_json, tool_output_json, timestamp}` | Insert into `tool_events` + `tool_event_payloads`. | +| `Stop` | `{session_id, prompt_number, ended_at}` | Mark turn complete. Enqueue compression for the just-ended turn. | +| `SessionEnd` | `{session_id, ended_at, exit_reason}` | Mark session complete. Enqueue session-rollup compression. | + +### Failure modes + +- `cce serve` not running: curl fails fast (-m 1), hook returns OK, no + capture for that event. `cce status` surfaces "memory capture: offline". +- Port file missing: hook is a no-op. +- DB write fails inside `cce serve`: log to stderr, increment a + `hook_errors` counter exposed in dashboard. +- Tool payloads can be huge: stored uncompressed at hook time; + compression worker later replaces the row's raw JSON with a summary + and moves originals to a separate retention-bound table. + +## SQLite schema + +Eight tables in `~/.cce/projects//memory.db`. Timestamps stored as +both ISO text and epoch int (text for humans, int for sorting). FTS5 +virtual tables shown last. + +```sql +-- A session = one Claude Code invocation in this project. +CREATE TABLE sessions ( + id TEXT PRIMARY KEY, + project TEXT NOT NULL, + started_at_epoch INTEGER NOT NULL, + started_at TEXT NOT NULL, + ended_at_epoch INTEGER, + ended_at TEXT, + exit_reason TEXT, + prompt_count INTEGER DEFAULT 0, + status TEXT CHECK(status IN ('active','completed','failed')) NOT NULL DEFAULT 'active', + rollup_summary TEXT, + rollup_summary_at_epoch INTEGER +); +CREATE INDEX idx_sessions_started ON sessions(started_at_epoch DESC); + +-- One row per UserPromptSubmit. +CREATE TABLE prompts ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT NOT NULL REFERENCES sessions(id) ON DELETE CASCADE, + prompt_number INTEGER NOT NULL, + prompt_text TEXT NOT NULL, + created_at_epoch INTEGER NOT NULL, + created_at TEXT NOT NULL, + UNIQUE(session_id, prompt_number) +); +CREATE INDEX idx_prompts_session ON prompts(session_id, prompt_number); + +-- One row per PostToolUse. +CREATE TABLE tool_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT NOT NULL REFERENCES sessions(id) ON DELETE CASCADE, + prompt_number INTEGER NOT NULL, + tool_name TEXT NOT NULL, + payload_id INTEGER, + summary TEXT, + created_at_epoch INTEGER NOT NULL, + created_at TEXT NOT NULL +); +CREATE INDEX idx_events_session_turn ON tool_events(session_id, prompt_number); + +-- Sidecar table holding raw tool input/output. Read on layer-3 drill-down. +CREATE TABLE tool_event_payloads ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + raw_input TEXT NOT NULL, + raw_output TEXT, + size_bytes INTEGER NOT NULL +); + +-- One row per turn after compression. Layer-2 of progressive disclosure. +CREATE TABLE turn_summaries ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT NOT NULL REFERENCES sessions(id) ON DELETE CASCADE, + prompt_number INTEGER NOT NULL, + summary TEXT NOT NULL, + tier TEXT NOT NULL, + created_at_epoch INTEGER NOT NULL, + UNIQUE(session_id, prompt_number) +); + +-- Manual decisions (record_decision MCP tool). +CREATE TABLE decisions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT REFERENCES sessions(id) ON DELETE SET NULL, + decision TEXT NOT NULL, + reason TEXT NOT NULL, + source TEXT NOT NULL CHECK(source IN ('manual','migrated','auto')) DEFAULT 'manual', + created_at_epoch INTEGER NOT NULL, + created_at TEXT NOT NULL +); +CREATE INDEX idx_decisions_created ON decisions(created_at_epoch DESC); +CREATE INDEX idx_decisions_source ON decisions(source); + +-- Manual code-area annotations. +CREATE TABLE code_areas ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT REFERENCES sessions(id) ON DELETE SET NULL, + file_path TEXT NOT NULL, + description TEXT NOT NULL, + source TEXT NOT NULL CHECK(source IN ('manual','migrated','auto')) DEFAULT 'manual', + created_at_epoch INTEGER NOT NULL +); +CREATE INDEX idx_code_areas_file ON code_areas(file_path); + +-- Compression queue. +CREATE TABLE pending_compressions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + kind TEXT NOT NULL CHECK(kind IN ('turn','session_rollup')), + session_id TEXT NOT NULL, + prompt_number INTEGER, + enqueued_at_epoch INTEGER NOT NULL, + attempts INTEGER NOT NULL DEFAULT 0, + last_error TEXT, + UNIQUE(kind, session_id, prompt_number) +); + +-- FTS5 over prompts, decisions, summaries. +CREATE VIRTUAL TABLE prompts_fts USING fts5( + prompt_text, content='prompts', content_rowid='id' +); +CREATE VIRTUAL TABLE decisions_fts USING fts5( + decision, reason, content='decisions', content_rowid='id' +); +CREATE VIRTUAL TABLE turn_summaries_fts USING fts5( + summary, content='turn_summaries', content_rowid='id' +); + +-- Schema version for forward-compatible migrations. +CREATE TABLE schema_versions ( + version INTEGER PRIMARY KEY, + applied_at_epoch INTEGER NOT NULL +); +INSERT INTO schema_versions (version, applied_at_epoch) VALUES (1, strftime('%s','now')); +``` + +### Schema notes + +- `tool_events.payload_id` is nullable so the compression worker can null + it out (and delete the payload row) for events older than the + retention window — keeping only the `summary`. Default retention: 30 + days for raw payloads. Tunable. +- No `touched_files` table. The view is computed on demand from + `tool_events.tool_input_json` for tools whose input contains a + `file_path`. Cheaper than maintaining a counter table. +- Three FTS tables, not one combined. Lets `session_recall` weight + prompts vs decisions vs summaries differently (decisions outrank + prompts outrank summaries by default). +- `source='migrated'` on decisions/code_areas is what + `cce sessions migrate` writes when it imports JSON. + +## Background compression worker + +Runs inside `cce serve`'s asyncio loop. Pseudocode: + +```python +async def compression_loop(): + while not shutting_down: + try: + row = await db.fetch_oldest_pending() + if row is None: + await asyncio.sleep(5) + continue + try: + if row.kind == "turn": + await compress_turn(row.session_id, row.prompt_number) + else: + await compress_session_rollup(row.session_id) + await db.delete_pending(row.id) + except Exception as exc: + await db.bump_attempts(row.id, str(exc)) + except Exception: + log.exception("compression loop iteration failed") + await asyncio.sleep(10) +``` + +### Extractive algorithm (turn level) + +``` +Input: all rows in prompts + tool_events for (session_id, prompt_number) +1. Build candidate sentence list: + - The user prompt (split into sentences) + - For each tool_event: + - Tool name + brief input descriptor (e.g. "Edit cli.py") + - First N sentences of tool_output (capped) +2. Embed each candidate with the bge-small model already loaded. +3. Compute centroid = mean of all candidate embeddings. +4. Pick top-K (default 3) sentences by cosine similarity to centroid. +5. Concatenate in original order, prefixed by "[turn N]". +6. Write to turn_summaries with tier='extractive'. +``` + +### Session rollup + +Concatenate all `turn_summaries` for the session, run the same +extractive algorithm against that text with K=5. Write to +`sessions.rollup_summary`. + +If at any point the embedder isn't loaded yet (rare; only on extreme +cold start), fall back to truncation: first 200 chars of each item, +joined. + +## MCP tool surface + +| Tool | Behaviour | +|---|---| +| `session_recall(topic)` | **Extended.** FTS5 search across `decisions_fts`, `prompts_fts`, `turn_summaries_fts`. Decisions weighted highest. Returns top-N hits with `{layer: 'index'\|'timeline'\|'event', id, snippet, session_id}`. Backward compatible — the tool name + topic argument are unchanged; the return shape is a strict superset. | +| `session_timeline(session_id, limit=20)` | **New.** Returns the session's `turn_summaries` in order, plus session metadata. Layer 2. | +| `session_event(event_id)` | **New.** Returns the raw payload for a single `tool_events` row (input + output JSON) if still within retention. Layer 3. | +| `record_decision(decision, reason)` | Unchanged. Writes to `decisions` with `source='manual'`. | +| `record_code_area(file_path, description)` | Unchanged. Writes to `code_areas` with `source='manual'`. | + +## Migration command + +`cce sessions migrate` (idempotent): + +1. Locate the per-project DB. If absent, create it (`memory/db.py` + bootstrap). +2. Walk `~/.cce/projects//sessions/*.json` and the legacy + `~/.claude-context-engine/projects//sessions/*.json`. +3. For each JSON file, parse and import: + - decisions → `decisions` with `source='migrated'` + - code_areas → `code_areas` with `source='migrated'` + - touched_files counts → ignored (replaced by computed view) +4. Skip files already imported (track by source filename in a + `migrated_files` table). +5. After successful import, archive consumed JSON to + `~/.cce/projects//sessions/migrated.zip` and remove the source + files. Idempotent rerun is a no-op. + +## Dashboard panels (v1) + +Integrated into the existing dashboard at `dashboard/_page.py` and +`dashboard/server.py`. + +1. **Sessions list** (`/memory`). Table: started, ended, prompts, + tool-uses, status, rollup summary. Click a row → session timeline. +2. **Session timeline** (`/memory/sessions/`). Header: session + metadata + rollup. Body: ordered turn summaries; each turn + expandable to its tool events. Tool events expandable to raw + payload (if still within retention). +3. **Decisions search** (`/memory/decisions`). FTS5 search box, results + list with `source` facet (manual / auto / migrated). Each result + links back to its session. + +Hot-files panel and queue-health panel are deferred — they're +observability nice-to-haves that we'd add once the core lands and we +know what fails. + +## Phasing + +This work ships as **5 sequential PRs** off +`feature/memory-claude-mem-parity`. Each PR is independently reviewable +and leaves the tree in a working state. + +| PR | Scope | Notes | +|---|---|---| +| **1. Foundation** | `memory/db.py` schema + `memory/migrate.py` + `cce sessions migrate` CLI + tests. | No behaviour change to existing capture path. | +| **2. Capture** | 5 hooks + `cce_hook.sh` + HTTP endpoints in `serve_http.py`. Old JSON path retired. | New writes land in `memory.db`. | +| **3. Compress** | Background asyncio worker in `cce serve`. Extractive summariser using bge-small. | Truncation final fallback. | +| **4. Recall** | Extended `session_recall`; new `session_timeline`, `session_event` MCP tools. | Backward compatible at the topic-argument level. | +| **5. Dashboard** | 3 panels (sessions list, timeline, decisions search). | Builds on existing dashboard scaffolding. | + +## Testing strategy + +- **Unit tests** for the schema bootstrap (PR 1), migration importer + (PR 1), extractive summariser (PR 3), MCP tool routing (PR 4). +- **Integration test** in PR 2: post a payload to each + `/hooks/` endpoint, assert the row lands in the correct table. +- **End-to-end** in PR 3: drive a fake session through hooks, wait for + the compression worker to drain, assert turn_summaries + rollup are + populated with the expected tiers. +- The existing 301-test suite must stay green at every commit. + +## Out of scope (explicit non-goals) + +- Cross-project memory views. Per-project DBs by design; inspecting + multiple projects at once means opening multiple dashboards. +- Encrypted memory-at-rest. Defer until/unless a concrete user need. +- Auto-extracting decisions from transcripts (vs. just summarising + turns). The compressor only summarises; decisions remain manual MCP + input. Auto-decision extraction is a follow-up if extractive turn + summaries prove rich enough to mine. +- Web viewer as a separate process. Integrated into the existing CCE + dashboard. +- Embedded abstractive LLM (e.g. flan-t5-small). Extractive with + bge-small is sufficient for v1. Revisit if turn summaries prove too + shallow. diff --git a/src/context_engine/cli.py b/src/context_engine/cli.py index 8ec3276..a7847c0 100644 --- a/src/context_engine/cli.py +++ b/src/context_engine/cli.py @@ -1838,6 +1838,56 @@ def sessions_prune(ctx: click.Context, threshold: int, keep: int) -> None: ) +@sessions.command(name="migrate") +@click.option( + "--no-archive", + is_flag=True, + default=False, + help="Don't archive consumed JSON files into migrated.zip after import.", +) +@click.pass_context +def sessions_migrate(ctx: click.Context, no_archive: bool) -> None: + """Import legacy per-session JSON files into the per-project memory.db. + + Idempotent: rerun is a no-op once everything has been imported. Imported + decisions and code areas are tagged with source='migrated' so future + session_recall can rank them appropriately. + """ + from context_engine.memory import db as memory_db, migrate as memory_migrate + + config = ctx.obj["config"] + project_name = Path.cwd().name + storage_base = Path(config.storage_path) / project_name + db_path = memory_db.memory_db_path(storage_base) + + conn = memory_db.connect(db_path) + try: + summary = memory_migrate.migrate( + conn, + project_name=project_name, + storage_base=storage_base, + archive=not no_archive, + ) + finally: + conn.close() + + if not summary.sources_scanned: + click.echo(f" {DOT} No legacy session directories found.") + return + + click.echo(f" {CHECK} Scanned: {len(summary.sources_scanned)} source dir(s)") + if summary.files_imported == 0 and summary.files_skipped > 0: + click.echo(f" {DOT} {summary.files_skipped} file(s) already imported. Nothing to do.") + return + click.echo( + f" {CHECK} Imported {summary.files_imported} file(s) → " + f"{summary.decisions_imported} decision(s), " + f"{summary.code_areas_imported} code area(s)." + ) + if summary.files_archived: + click.echo(f" {CHECK} Archived {summary.files_archived} file(s) to migrated.zip") + + async def _run_index( config, project_dir: str, diff --git a/src/context_engine/memory/__init__.py b/src/context_engine/memory/__init__.py new file mode 100644 index 0000000..4f50458 --- /dev/null +++ b/src/context_engine/memory/__init__.py @@ -0,0 +1,6 @@ +"""Per-project memory store — SQLite tables backing cross-session recall. + +This package introduces the new memory.db storage. The legacy JSON-per-session +capture path in `context_engine.integration.session_capture` continues to work +unchanged; it is retired in a follow-up PR once hooks land. +""" diff --git a/src/context_engine/memory/db.py b/src/context_engine/memory/db.py new file mode 100644 index 0000000..52e5dfb --- /dev/null +++ b/src/context_engine/memory/db.py @@ -0,0 +1,246 @@ +"""Per-project memory.db bootstrap and connection helper. + +Schema version 1 — see docs/specs/2026-04-28-memory-claude-mem-parity-design.md. + +Idempotent: opening an existing db is a no-op; opening an empty file creates +the schema and stamps version=1. +""" +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +CURRENT_VERSION = 1 + +_SCHEMA_V1 = [ + """ + CREATE TABLE sessions ( + id TEXT PRIMARY KEY, + project TEXT NOT NULL, + started_at_epoch INTEGER NOT NULL, + started_at TEXT NOT NULL, + ended_at_epoch INTEGER, + ended_at TEXT, + exit_reason TEXT, + prompt_count INTEGER DEFAULT 0, + status TEXT CHECK(status IN ('active','completed','failed')) NOT NULL DEFAULT 'active', + rollup_summary TEXT, + rollup_summary_at_epoch INTEGER + ) + """, + "CREATE INDEX idx_sessions_started ON sessions(started_at_epoch DESC)", + + """ + CREATE TABLE prompts ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT NOT NULL REFERENCES sessions(id) ON DELETE CASCADE, + prompt_number INTEGER NOT NULL, + prompt_text TEXT NOT NULL, + created_at_epoch INTEGER NOT NULL, + created_at TEXT NOT NULL, + UNIQUE(session_id, prompt_number) + ) + """, + "CREATE INDEX idx_prompts_session ON prompts(session_id, prompt_number)", + + """ + CREATE TABLE tool_event_payloads ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + raw_input TEXT NOT NULL, + raw_output TEXT, + size_bytes INTEGER NOT NULL + ) + """, + + """ + CREATE TABLE tool_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT NOT NULL REFERENCES sessions(id) ON DELETE CASCADE, + prompt_number INTEGER NOT NULL, + tool_name TEXT NOT NULL, + payload_id INTEGER REFERENCES tool_event_payloads(id) ON DELETE SET NULL, + summary TEXT, + created_at_epoch INTEGER NOT NULL, + created_at TEXT NOT NULL + ) + """, + "CREATE INDEX idx_events_session_turn ON tool_events(session_id, prompt_number)", + + """ + CREATE TABLE turn_summaries ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT NOT NULL REFERENCES sessions(id) ON DELETE CASCADE, + prompt_number INTEGER NOT NULL, + summary TEXT NOT NULL, + tier TEXT NOT NULL, + created_at_epoch INTEGER NOT NULL, + UNIQUE(session_id, prompt_number) + ) + """, + + """ + CREATE TABLE decisions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT REFERENCES sessions(id) ON DELETE SET NULL, + decision TEXT NOT NULL, + reason TEXT NOT NULL, + source TEXT NOT NULL CHECK(source IN ('manual','migrated','auto')) DEFAULT 'manual', + created_at_epoch INTEGER NOT NULL, + created_at TEXT NOT NULL + ) + """, + "CREATE INDEX idx_decisions_created ON decisions(created_at_epoch DESC)", + "CREATE INDEX idx_decisions_source ON decisions(source)", + + """ + CREATE TABLE code_areas ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT REFERENCES sessions(id) ON DELETE SET NULL, + file_path TEXT NOT NULL, + description TEXT NOT NULL, + source TEXT NOT NULL CHECK(source IN ('manual','migrated','auto')) DEFAULT 'manual', + created_at_epoch INTEGER NOT NULL + ) + """, + "CREATE INDEX idx_code_areas_file ON code_areas(file_path)", + + """ + CREATE TABLE pending_compressions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + kind TEXT NOT NULL CHECK(kind IN ('turn','session_rollup')), + session_id TEXT NOT NULL, + prompt_number INTEGER, + enqueued_at_epoch INTEGER NOT NULL, + attempts INTEGER NOT NULL DEFAULT 0, + last_error TEXT, + UNIQUE(kind, session_id, prompt_number) + ) + """, + + # Tracks files consumed by `cce sessions migrate` so reruns are idempotent. + """ + CREATE TABLE migrated_files ( + source_path TEXT PRIMARY KEY, + imported_at_epoch INTEGER NOT NULL + ) + """, + + # FTS5 virtual tables — search index for session_recall. + "CREATE VIRTUAL TABLE prompts_fts USING fts5(prompt_text, content='prompts', content_rowid='id')", + "CREATE VIRTUAL TABLE decisions_fts USING fts5(decision, reason, content='decisions', content_rowid='id')", + "CREATE VIRTUAL TABLE turn_summaries_fts USING fts5(summary, content='turn_summaries', content_rowid='id')", + + # Triggers keep the FTS shadow tables in sync with their source tables. + """ + CREATE TRIGGER prompts_ai AFTER INSERT ON prompts BEGIN + INSERT INTO prompts_fts(rowid, prompt_text) VALUES (new.id, new.prompt_text); + END + """, + """ + CREATE TRIGGER prompts_ad AFTER DELETE ON prompts BEGIN + INSERT INTO prompts_fts(prompts_fts, rowid, prompt_text) VALUES('delete', old.id, old.prompt_text); + END + """, + """ + CREATE TRIGGER prompts_au AFTER UPDATE ON prompts BEGIN + INSERT INTO prompts_fts(prompts_fts, rowid, prompt_text) VALUES('delete', old.id, old.prompt_text); + INSERT INTO prompts_fts(rowid, prompt_text) VALUES (new.id, new.prompt_text); + END + """, + + """ + CREATE TRIGGER decisions_ai AFTER INSERT ON decisions BEGIN + INSERT INTO decisions_fts(rowid, decision, reason) VALUES (new.id, new.decision, new.reason); + END + """, + """ + CREATE TRIGGER decisions_ad AFTER DELETE ON decisions BEGIN + INSERT INTO decisions_fts(decisions_fts, rowid, decision, reason) VALUES('delete', old.id, old.decision, old.reason); + END + """, + """ + CREATE TRIGGER decisions_au AFTER UPDATE ON decisions BEGIN + INSERT INTO decisions_fts(decisions_fts, rowid, decision, reason) VALUES('delete', old.id, old.decision, old.reason); + INSERT INTO decisions_fts(rowid, decision, reason) VALUES (new.id, new.decision, new.reason); + END + """, + + """ + CREATE TRIGGER turn_summaries_ai AFTER INSERT ON turn_summaries BEGIN + INSERT INTO turn_summaries_fts(rowid, summary) VALUES (new.id, new.summary); + END + """, + """ + CREATE TRIGGER turn_summaries_ad AFTER DELETE ON turn_summaries BEGIN + INSERT INTO turn_summaries_fts(turn_summaries_fts, rowid, summary) VALUES('delete', old.id, old.summary); + END + """, + """ + CREATE TRIGGER turn_summaries_au AFTER UPDATE ON turn_summaries BEGIN + INSERT INTO turn_summaries_fts(turn_summaries_fts, rowid, summary) VALUES('delete', old.id, old.summary); + INSERT INTO turn_summaries_fts(rowid, summary) VALUES (new.id, new.summary); + END + """, + + """ + CREATE TABLE schema_versions ( + version INTEGER PRIMARY KEY, + applied_at_epoch INTEGER NOT NULL + ) + """, +] + + +def connect(db_path: str | Path) -> sqlite3.Connection: + """Open (or create) the per-project memory.db at `db_path`. + + Bootstraps the schema if the file is empty. Idempotent: re-opening an + initialised db just returns a configured connection. + """ + db_path = Path(db_path) + db_path.parent.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(str(db_path)) + conn.row_factory = sqlite3.Row + # Foreign keys must be enabled per-connection in SQLite. + conn.execute("PRAGMA foreign_keys = ON") + # WAL gives concurrent readers (the dashboard) decent isolation while the + # MCP server writes; no impact on single-process use. + conn.execute("PRAGMA journal_mode = WAL") + _ensure_schema(conn) + return conn + + +def _ensure_schema(conn: sqlite3.Connection) -> None: + cur = conn.cursor() + # If schema_versions exists, the schema has been bootstrapped at least + # once. Future migrations would compare CURRENT_VERSION here. + row = cur.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name='schema_versions'" + ).fetchone() + if row is not None: + return + cur.execute("BEGIN") + try: + for stmt in _SCHEMA_V1: + cur.execute(stmt) + cur.execute( + "INSERT INTO schema_versions (version, applied_at_epoch) " + "VALUES (?, strftime('%s','now'))", + (CURRENT_VERSION,), + ) + conn.commit() + except Exception: + conn.rollback() + raise + + +def schema_version(conn: sqlite3.Connection) -> int: + row = conn.execute( + "SELECT MAX(version) AS v FROM schema_versions" + ).fetchone() + return int(row["v"]) if row and row["v"] is not None else 0 + + +def memory_db_path(storage_base: str | Path) -> Path: + """Canonical location of the memory db inside a project's storage dir.""" + return Path(storage_base) / "memory.db" diff --git a/src/context_engine/memory/migrate.py b/src/context_engine/memory/migrate.py new file mode 100644 index 0000000..76279bd --- /dev/null +++ b/src/context_engine/memory/migrate.py @@ -0,0 +1,229 @@ +"""Import legacy per-session JSON files into the new memory.db. + +Walks each project's `sessions/` directory (and the legacy +`~/.claude-context-engine/projects//sessions/` path), parses each +*.json, imports decisions and code_areas with `source='migrated'`, then +archives the consumed files into `migrated.zip` and removes them. + +Idempotent — `migrated_files` tracks what has already been imported so a +rerun is a no-op. +""" +from __future__ import annotations + +import json +import logging +import sqlite3 +import time +import zipfile +from dataclasses import dataclass, field +from pathlib import Path + +log = logging.getLogger(__name__) + +_DECISIONS_LOG_NAME = "decisions_log.json" + + +@dataclass +class MigrationSummary: + decisions_imported: int = 0 + code_areas_imported: int = 0 + files_imported: int = 0 + files_archived: int = 0 + files_skipped: int = 0 + sources_scanned: list[str] = field(default_factory=list) + + +def candidate_session_dirs(project_name: str, primary_storage_base: Path) -> list[Path]: + """Return every directory we should scan for legacy session JSON. + + Currently: + - /sessions/ (current path) + - ~/.claude-context-engine/projects//sessions/ (pre-rebrand) + """ + legacy_root = Path.home() / ".claude-context-engine" / "projects" / project_name / "sessions" + return [ + Path(primary_storage_base) / "sessions", + legacy_root, + ] + + +def migrate( + conn: sqlite3.Connection, + project_name: str, + storage_base: str | Path, + *, + archive: bool = True, +) -> MigrationSummary: + """Import all legacy JSON sessions for `project_name` into the open db. + + `storage_base` is the per-project storage directory (e.g. + ~/.cce/projects/). `archive=True` zips and deletes consumed + JSONs after successful import; pass False from tests that want to + re-read the source files. + """ + storage_base = Path(storage_base) + summary = MigrationSummary() + + consumed_per_dir: dict[Path, list[Path]] = {} + + for sessions_dir in candidate_session_dirs(project_name, storage_base): + if not sessions_dir.exists(): + continue + summary.sources_scanned.append(str(sessions_dir)) + + json_files = sorted( + f for f in sessions_dir.glob("*.json") + # decisions_log.json is the consolidated archive — also a valid + # source of decisions, just in a different shape. + ) + consumed: list[Path] = [] + for f in json_files: + if _already_imported(conn, f): + summary.files_skipped += 1 + continue + try: + imported = _import_one(conn, f) + except (json.JSONDecodeError, OSError) as exc: + log.warning("Skipping unreadable session file %s: %s", f, exc) + continue + summary.decisions_imported += imported.decisions + summary.code_areas_imported += imported.code_areas + summary.files_imported += 1 + _mark_imported(conn, f) + consumed.append(f) + if consumed: + consumed_per_dir[sessions_dir] = consumed + + conn.commit() + + if archive and consumed_per_dir: + for sessions_dir, files in consumed_per_dir.items(): + archived = _archive_and_remove(sessions_dir, files) + summary.files_archived += archived + + return summary + + +@dataclass +class _ImportCounts: + decisions: int = 0 + code_areas: int = 0 + + +def _import_one(conn: sqlite3.Connection, source: Path) -> _ImportCounts: + """Import a single legacy JSON file. Returns counts of imported rows.""" + counts = _ImportCounts() + data = json.loads(source.read_text()) + + # decisions_log.json is a top-level list of decision dicts, not a session. + if source.name == _DECISIONS_LOG_NAME and isinstance(data, list): + for d in data: + _insert_decision( + conn, + session_id=None, + decision=d.get("decision", ""), + reason=d.get("reason", ""), + timestamp=d.get("timestamp"), + ) + counts.decisions += 1 + return counts + + # Per-session JSON: {"id", "decisions": [...], "code_areas": [...], ...} + if not isinstance(data, dict): + return counts + + session_id = data.get("id") + # We do not synthesise a sessions row from migrated data — the legacy + # JSON files predate the new sessions schema and miss timestamps in a + # form we can trust. Importing decisions with session_id=None is the + # safer choice; the FK is nullable on purpose. + for d in data.get("decisions", []) or []: + _insert_decision( + conn, + session_id=session_id if _session_exists(conn, session_id) else None, + decision=d.get("decision", ""), + reason=d.get("reason", ""), + timestamp=d.get("timestamp"), + ) + counts.decisions += 1 + for c in data.get("code_areas", []) or []: + _insert_code_area( + conn, + session_id=session_id if _session_exists(conn, session_id) else None, + file_path=c.get("file_path", ""), + description=c.get("description", ""), + timestamp=c.get("timestamp"), + ) + counts.code_areas += 1 + return counts + + +def _insert_decision(conn, *, session_id, decision, reason, timestamp): + epoch = int(timestamp) if timestamp else int(time.time()) + iso = time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(epoch)) + conn.execute( + "INSERT INTO decisions (session_id, decision, reason, source, " + "created_at_epoch, created_at) VALUES (?, ?, ?, 'migrated', ?, ?)", + (session_id, decision, reason, epoch, iso), + ) + + +def _insert_code_area(conn, *, session_id, file_path, description, timestamp): + epoch = int(timestamp) if timestamp else int(time.time()) + conn.execute( + "INSERT INTO code_areas (session_id, file_path, description, source, " + "created_at_epoch) VALUES (?, ?, ?, 'migrated', ?)", + (session_id, file_path, description, epoch), + ) + + +def _session_exists(conn, session_id) -> bool: + if not session_id: + return False + row = conn.execute( + "SELECT 1 FROM sessions WHERE id = ?", (session_id,) + ).fetchone() + return row is not None + + +def _already_imported(conn, source: Path) -> bool: + row = conn.execute( + "SELECT 1 FROM migrated_files WHERE source_path = ?", + (str(source),), + ).fetchone() + return row is not None + + +def _mark_imported(conn, source: Path) -> None: + conn.execute( + "INSERT OR IGNORE INTO migrated_files (source_path, imported_at_epoch) " + "VALUES (?, ?)", + (str(source), int(time.time())), + ) + + +def _archive_and_remove(sessions_dir: Path, files: list[Path]) -> int: + """Append `files` to `sessions_dir/migrated.zip` and remove the originals. + + Returns the number of files actually written to the zip. + """ + if not files: + return 0 + archive_path = sessions_dir / "migrated.zip" + written = 0 + with zipfile.ZipFile(archive_path, mode="a", compression=zipfile.ZIP_DEFLATED) as zf: + existing = set(zf.namelist()) + for f in files: + arcname = f.name + if arcname in existing: + # Already in the archive from a previous run; just delete. + pass + else: + zf.write(f, arcname=arcname) + written += 1 + for f in files: + try: + f.unlink() + except OSError as exc: + log.warning("Could not remove migrated file %s: %s", f, exc) + return written diff --git a/tests/memory/__init__.py b/tests/memory/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/memory/test_db.py b/tests/memory/test_db.py new file mode 100644 index 0000000..b3a776f --- /dev/null +++ b/tests/memory/test_db.py @@ -0,0 +1,94 @@ +"""Tests for memory.db schema bootstrap.""" +from __future__ import annotations + +from pathlib import Path + +import pytest + +from context_engine.memory import db as memory_db + + +def test_connect_creates_schema_on_empty_file(tmp_path: Path): + db_path = tmp_path / "memory.db" + conn = memory_db.connect(db_path) + try: + assert memory_db.schema_version(conn) == memory_db.CURRENT_VERSION + # All declared tables exist. + tables = { + row["name"] + for row in conn.execute( + "SELECT name FROM sqlite_master WHERE type='table'" + ) + } + for required in [ + "sessions", "prompts", "tool_events", "tool_event_payloads", + "turn_summaries", "decisions", "code_areas", + "pending_compressions", "migrated_files", "schema_versions", + ]: + assert required in tables, f"missing table {required}" + # FTS virtual tables exist too. + for fts in ["prompts_fts", "decisions_fts", "turn_summaries_fts"]: + assert fts in tables, f"missing fts {fts}" + finally: + conn.close() + + +def test_connect_is_idempotent(tmp_path: Path): + db_path = tmp_path / "memory.db" + c1 = memory_db.connect(db_path) + c1.close() + c2 = memory_db.connect(db_path) + try: + # Re-opening must not re-stamp the version row. + rows = list(c2.execute("SELECT version FROM schema_versions")) + assert len(rows) == 1 + assert rows[0]["version"] == memory_db.CURRENT_VERSION + finally: + c2.close() + + +def test_foreign_keys_enabled(tmp_path: Path): + db_path = tmp_path / "memory.db" + conn = memory_db.connect(db_path) + try: + result = conn.execute("PRAGMA foreign_keys").fetchone()[0] + assert result == 1 + finally: + conn.close() + + +def test_decisions_fts_search(tmp_path: Path): + """A decision inserted into the parent table is searchable via fts.""" + db_path = tmp_path / "memory.db" + conn = memory_db.connect(db_path) + try: + conn.execute( + "INSERT INTO decisions (decision, reason, source, " + "created_at_epoch, created_at) " + "VALUES (?, ?, 'manual', 1700000000, '2023-11-14T22:13:20')", + ("Use SQLite for memory store", "Per-project files, FTS5 builtin"), + ) + conn.commit() + rows = conn.execute( + "SELECT decisions.decision FROM decisions_fts " + "JOIN decisions ON decisions.id = decisions_fts.rowid " + "WHERE decisions_fts MATCH 'sqlite'" + ).fetchall() + assert len(rows) == 1 + assert "SQLite" in rows[0]["decision"] + finally: + conn.close() + + +def test_status_check_constraint(tmp_path: Path): + db_path = tmp_path / "memory.db" + conn = memory_db.connect(db_path) + try: + with pytest.raises(Exception): + conn.execute( + "INSERT INTO sessions (id, project, started_at_epoch, " + "started_at, status) VALUES (?, ?, ?, ?, 'bogus')", + ("abc", "demo", 1700000000, "2023-11-14T22:13:20"), + ) + finally: + conn.close() diff --git a/tests/memory/test_migrate.py b/tests/memory/test_migrate.py new file mode 100644 index 0000000..d08d45a --- /dev/null +++ b/tests/memory/test_migrate.py @@ -0,0 +1,177 @@ +"""Tests for `cce sessions migrate` — JSON to memory.db importer.""" +from __future__ import annotations + +import json +import zipfile +from pathlib import Path + +from context_engine.memory import db as memory_db, migrate as memory_migrate + + +def _write_session(sessions_dir: Path, session_id: str, body: dict) -> Path: + sessions_dir.mkdir(parents=True, exist_ok=True) + p = sessions_dir / f"{session_id}.json" + p.write_text(json.dumps(body)) + return p + + +def test_migrate_imports_decisions_from_session_json(tmp_path: Path): + storage = tmp_path / "storage" + sessions = storage / "sessions" + _write_session(sessions, "abc123", { + "id": "abc123", + "decisions": [ + {"decision": "Use bge-small for extractive", "reason": "Already loaded", "timestamp": 1700000000}, + {"decision": "5 hooks not 1", "reason": "claude-mem parity", "timestamp": 1700000100}, + ], + "code_areas": [ + {"file_path": "src/foo.py", "description": "memory db init", "timestamp": 1700000200}, + ], + }) + + db_path = memory_db.memory_db_path(storage) + conn = memory_db.connect(db_path) + try: + summary = memory_migrate.migrate( + conn, project_name="demo", storage_base=storage, archive=False, + ) + finally: + conn.close() + + assert summary.files_imported == 1 + assert summary.decisions_imported == 2 + assert summary.code_areas_imported == 1 + + # Verify the rows landed and source='migrated'. + conn = memory_db.connect(db_path) + try: + rows = list(conn.execute( + "SELECT decision, source FROM decisions ORDER BY created_at_epoch" + )) + assert len(rows) == 2 + assert all(r["source"] == "migrated" for r in rows) + assert "bge-small" in rows[0]["decision"] + finally: + conn.close() + + +def test_migrate_is_idempotent(tmp_path: Path): + storage = tmp_path / "storage" + sessions = storage / "sessions" + _write_session(sessions, "abc123", { + "id": "abc123", + "decisions": [{"decision": "X", "reason": "Y", "timestamp": 1700000000}], + "code_areas": [], + }) + + db_path = memory_db.memory_db_path(storage) + + conn = memory_db.connect(db_path) + try: + s1 = memory_migrate.migrate( + conn, project_name="demo", storage_base=storage, archive=False, + ) + finally: + conn.close() + assert s1.files_imported == 1 + + # Re-write the source so the second pass would see content if it + # weren't tracking already-imported files. + _write_session(sessions, "abc123", { + "id": "abc123", + "decisions": [{"decision": "X", "reason": "Y", "timestamp": 1700000000}], + "code_areas": [], + }) + + conn = memory_db.connect(db_path) + try: + s2 = memory_migrate.migrate( + conn, project_name="demo", storage_base=storage, archive=False, + ) + finally: + conn.close() + assert s2.files_imported == 0 + assert s2.files_skipped == 1 + + conn = memory_db.connect(db_path) + try: + n_decisions = conn.execute("SELECT COUNT(*) AS n FROM decisions").fetchone()["n"] + finally: + conn.close() + # Only the first migrate should have imported the decision. + assert n_decisions == 1 + + +def test_migrate_archives_consumed_files(tmp_path: Path): + storage = tmp_path / "storage" + sessions = storage / "sessions" + p = _write_session(sessions, "abc123", { + "id": "abc123", + "decisions": [{"decision": "X", "reason": "Y", "timestamp": 1700000000}], + "code_areas": [], + }) + + db_path = memory_db.memory_db_path(storage) + conn = memory_db.connect(db_path) + try: + summary = memory_migrate.migrate( + conn, project_name="demo", storage_base=storage, archive=True, + ) + finally: + conn.close() + + assert summary.files_archived == 1 + assert not p.exists() + archive = sessions / "migrated.zip" + assert archive.exists() + with zipfile.ZipFile(archive) as zf: + names = zf.namelist() + assert "abc123.json" in names + + +def test_migrate_decisions_log_top_level_list(tmp_path: Path): + """decisions_log.json (the consolidated archive) is a top-level list.""" + storage = tmp_path / "storage" + sessions = storage / "sessions" + sessions.mkdir(parents=True, exist_ok=True) + log_path = sessions / "decisions_log.json" + log_path.write_text(json.dumps([ + {"decision": "Pick A", "reason": "B", "timestamp": 1700000000, "session_id": "s1"}, + {"decision": "Pick C", "reason": "D", "timestamp": 1700000100, "session_id": "s2"}, + ])) + + db_path = memory_db.memory_db_path(storage) + conn = memory_db.connect(db_path) + try: + summary = memory_migrate.migrate( + conn, project_name="demo", storage_base=storage, archive=False, + ) + finally: + conn.close() + assert summary.decisions_imported == 2 + + conn = memory_db.connect(db_path) + try: + rows = list(conn.execute( + "SELECT decision, source FROM decisions ORDER BY created_at_epoch" + )) + finally: + conn.close() + assert len(rows) == 2 + assert all(r["source"] == "migrated" for r in rows) + + +def test_migrate_no_legacy_dirs_returns_empty_summary(tmp_path: Path): + storage = tmp_path / "storage" + db_path = memory_db.memory_db_path(storage) + conn = memory_db.connect(db_path) + try: + summary = memory_migrate.migrate( + conn, project_name="ghost", storage_base=storage, archive=False, + ) + finally: + conn.close() + assert summary.files_imported == 0 + # candidate_session_dirs returns paths that may not exist; only existing + # ones go in `sources_scanned`. + assert summary.sources_scanned == []