feat(memory) PR 4/5: recall — extend session_recall + new session_timeline + session_event - #5
Conversation
… session_event
PR 4 of 5 against feature/memory-claude-mem-parity, stacked on PR 3.
Wires the MCP retrieval surface to memory.db while preserving the
existing JSON path for backward compatibility.
- ContextEngineMCP opens memory.db on startup and seeds an
INSERT OR IGNORE sessions row so manual record_decision /
record_code_area dual-writes don't fail the FK constraint when the
SessionStart hook hasn't fired yet (test envs, future-non-CC clients).
- session_recall now folds three new candidate sources on top of the
existing JSON sessions / consolidated decisions:
- decisions (manual + migrated, last 200 by recency)
- code_areas (manual + migrated, last 200 by recency)
- turn_summaries (last 200 turns, the layer-1 compact index)
Tags include source and session_id so the agent can drill via the
new tools.
- session_timeline(session_id, limit=20) — layer 2. Returns the
session's turn_summaries with rollup + status header.
- session_event(event_id) — layer 3. Returns the raw input/output
payload for one tool_event, with a dedicated "aged out" message
when the payload row was pruned by retention.
- record_decision and record_code_area now dual-write to memory.db
with source='manual'. Prior JSON write path remains active so a
rollback to a previous PR doesn't lose recall coverage. The JSON
write side can be retired once parity is confirmed in production.
Tests (10 new, full suite 350 passed):
- tests/memory/test_mcp_recall.py covers dual-write of decisions and
code_areas, session_timeline with seeded summaries, session_event
payload roundtrip + aged-out message + invalid id, session_recall
surfacing memory.db decisions, and TOOL_NAMES registration.
There was a problem hiding this comment.
Pull request overview
Adds progressive-disclosure “recall” capabilities to the MCP surface by extending session_recall and introducing drill-down tools for session timelines and per-event raw payloads, backed by (and dual-writing into) memory.db.
Changes:
- Extend MCP tool registry + dispatcher to include new
session_timeline(session_id, limit)andsession_event(event_id)tools. - Add
memory.dbconnection initialization in the MCP server and dual-writerecord_decision/record_code_areainto SQLite. - Extend
session_recallcandidate gathering to includememory.dbdecisions, code areas, and turn summaries; add direct handler tests.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 6 comments.
| File | Description |
|---|---|
src/context_engine/integration/mcp_server.py |
Adds memory.db wiring, registers/dispatches new tools, implements timeline/event handlers, and extends recall candidate sources. |
tests/memory/test_mcp_recall.py |
Adds unit tests covering dual-write behavior and the new timeline/event recall handlers. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| return [TextContent(type="text", text="decision is required.")] | ||
| self._session_capture.record_decision(self._session_id, decision, reason) | ||
| self._persist_current_session() | ||
| # Dual-write into memory.db so the new recall path (FTS5) sees it too. |
There was a problem hiding this comment.
This comment says the dual-write is so the "FTS5" recall path sees it, but session_recall currently ranks candidates via embeddings/cosine similarity and does not query the FTS tables. Please update the comment to reflect the actual mechanism (or switch recall to use the FTS index).
| # Dual-write into memory.db so the new recall path (FTS5) sees it too. | |
| # Mirror manual decisions into memory.db as well so they remain | |
| # available to the database-backed memory flows alongside the | |
| # session-capture copy used by session_recall. |
| # Fold in memory.db rows: decisions/code_areas (manual or migrated), | ||
| # plus the index of turn_summaries. Tagged with [layer:..|sid:..] so | ||
| # the agent knows where to drill from session_recall results — those | ||
| # tags map onto session_timeline / session_event drill-downs. | ||
| if self._memory_conn is not None: | ||
| try: | ||
| for row in self._memory_conn.execute( | ||
| "SELECT decision, reason, source, session_id " | ||
| "FROM decisions ORDER BY created_at_epoch DESC LIMIT 200" | ||
| ): | ||
| text = ( | ||
| f"[decision src={row['source']}|sid:{row['session_id'] or '-'}] " | ||
| f"{row['decision']} — {row['reason']}" | ||
| ) | ||
| if text not in seen: | ||
| seen.add(text) | ||
| candidates.append(text) | ||
| for row in self._memory_conn.execute( | ||
| "SELECT file_path, description, source, session_id " | ||
| "FROM code_areas ORDER BY created_at_epoch DESC LIMIT 200" | ||
| ): | ||
| text = ( | ||
| f"[code_area src={row['source']}|sid:{row['session_id'] or '-'}] " | ||
| f"{row['file_path']} — {row['description']}" | ||
| ) | ||
| if text not in seen: | ||
| seen.add(text) | ||
| candidates.append(text) | ||
| for row in self._memory_conn.execute( | ||
| "SELECT session_id, prompt_number, summary " | ||
| "FROM turn_summaries ORDER BY created_at_epoch DESC LIMIT 200" | ||
| ): |
There was a problem hiding this comment.
_search_sessions now unconditionally adds up to 600 additional memory.db candidates (decisions + code_areas + turn_summaries) and then calls embedder.embed_query() for every candidate. This increases worst-case recall latency and CPU substantially for large histories. Consider using the existing FTS5 indexes to prefilter candidates by topic (or otherwise reducing candidate count) before embedding/ranking.
| session_id = (args.get("session_id") or "").strip() | ||
| limit = int(args.get("limit") or 20) | ||
| if not session_id: | ||
| return [TextContent(type="text", text="session_id is required.")] | ||
| if self._memory_conn is None: | ||
| return [TextContent(type="text", text="Memory store not available.")] |
There was a problem hiding this comment.
limit = int(args.get('limit') or 20) will raise ValueError for non-integer inputs and also allows negative/very large limits, which can crash the handler (when called directly) or produce unbounded output. Consider parsing with try/except and clamping to a sane range (e.g., 1..200), similar to _clamp_top_k.
| meta = self._memory_conn.execute( | ||
| "SELECT project, started_at, ended_at, status, prompt_count, " | ||
| "rollup_summary FROM sessions WHERE id = ?", | ||
| (session_id,), | ||
| ).fetchone() |
There was a problem hiding this comment.
The sessions metadata query (SELECT project, started_at, ... FROM sessions) is executed outside the try/except. If the DB is unavailable/corrupt or schema is mid-migration, this will raise and bubble up as a generic tool failure. Consider wrapping this query in the same error handling as the turn_summaries query and returning a consistent, user-facing error message.
| row = self._memory_conn.execute( | ||
| "SELECT te.tool_name, te.session_id, te.prompt_number, te.created_at, " | ||
| "p.raw_input, p.raw_output FROM tool_events te " | ||
| "LEFT JOIN tool_event_payloads p ON p.id = te.payload_id " | ||
| "WHERE te.id = ?", | ||
| (event_id,), | ||
| ).fetchone() |
There was a problem hiding this comment.
_handle_session_event executes the SQL query without a try/except. Any sqlite error (locked DB, corruption, missing table during partial init) will raise and be caught only by the outer call_tool handler, returning a generic "Tool session_event failed" message. Consider catching DB exceptions here and returning a specific "event query failed" response (matching _handle_session_timeline behavior).
| row = self._memory_conn.execute( | |
| "SELECT te.tool_name, te.session_id, te.prompt_number, te.created_at, " | |
| "p.raw_input, p.raw_output FROM tool_events te " | |
| "LEFT JOIN tool_event_payloads p ON p.id = te.payload_id " | |
| "WHERE te.id = ?", | |
| (event_id,), | |
| ).fetchone() | |
| try: | |
| row = self._memory_conn.execute( | |
| "SELECT te.tool_name, te.session_id, te.prompt_number, te.created_at, " | |
| "p.raw_input, p.raw_output FROM tool_events te " | |
| "LEFT JOIN tool_event_payloads p ON p.id = te.payload_id " | |
| "WHERE te.id = ?", | |
| (event_id,), | |
| ).fetchone() | |
| except Exception as exc: | |
| return [TextContent(type="text", text=f"event query failed: {exc}")] |
| body = ( | ||
| f"event {event_id} · {row['tool_name']} · session {row['session_id']} · " | ||
| f"turn {row['prompt_number']} · {row['created_at']}\n\n" | ||
| f"input:\n{row['raw_input']}\n\n" | ||
| f"output:\n{row['raw_output']}" |
There was a problem hiding this comment.
tool_event_payloads.raw_output is nullable, but this formatter interpolates it directly. When raw_output is NULL, the response will contain the literal string None, which is misleading. Consider normalizing NULL to an empty string or an explicit placeholder (e.g., "") before formatting.
| body = ( | |
| f"event {event_id} · {row['tool_name']} · session {row['session_id']} · " | |
| f"turn {row['prompt_number']} · {row['created_at']}\n\n" | |
| f"input:\n{row['raw_input']}\n\n" | |
| f"output:\n{row['raw_output']}" | |
| raw_input = row["raw_input"] if row["raw_input"] is not None else "<no input>" | |
| raw_output = row["raw_output"] if row["raw_output"] is not None else "<no output>" | |
| body = ( | |
| f"event {event_id} · {row['tool_name']} · session {row['session_id']} · " | |
| f"turn {row['prompt_number']} · {row['created_at']}\n\n" | |
| f"input:\n{raw_input}\n\n" | |
| f"output:\n{raw_output}" |
Consolidates the 11 actionable items from the 5-PR memory stack into one
fixup commit on the unified ai-memory branch.
migrate.py
- Archive *before* mark-imported + commit, with rollback on zip failure
so a failed archive no longer leaves files stuck imported-but-not-archived.
- Preserve session_id linkage from decisions_log.json when the referenced
session already exists (was unconditionally NULL).
- Memoise _session_exists per-file (constant per archive entry).
- Use `timestamp is not None` so legacy 0/0.0 timestamps keep their original
ordering instead of being stamped to "now".
compressor.py
- Drop unused `Any` import.
- Cap raw_input at _TOOL_INPUT_CHAR_CAP (4 KB) before json.loads so multi-MB
patch payloads don't stall the compression worker.
- Yield the asyncio loop after every drained item, with a short breath every
5 items, so a backlog doesn't monopolise mcp.run_stdio().
mcp_server.py
- Use FTS5 MATCH on decisions_fts / turn_summaries_fts (and a LIKE filter on
code_areas) to prefilter recall candidates instead of embedding the latest
600 rows on every session_recall call.
- Clamp `limit` in session_timeline to 1..200 with a helper that handles bad
input cleanly.
- Wrap the sessions metadata query and the session_event payload query in
try/except so DB errors return a specific message.
- Handle NULL raw_input/raw_output in session_event without rendering the
string "None".
- Update the dual-write comment to reflect that recall now goes through FTS5.
design spec
- Drop the inaccurate "Eight tables" sentence; describe the support and
FTS5 tables that ship in v1 instead.
Tests: 43/43 memory + dashboard pass; 344/345 across the full suite (the
single failure is test_ollama_client hitting httpx.ReadTimeout because no
Ollama is running in this environment — unrelated).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
PR 4 of 5 — Recall
Stacked on PR 3 (compress). Base: `feature/memory-pr3-compress`.
Implements the 3-layer progressive disclosure on the MCP surface.
Compatibility
Test plan