diff --git a/headroom/cli/memory.py b/headroom/cli/memory.py index 91e717c94..51a1fd1f9 100644 --- a/headroom/cli/memory.py +++ b/headroom/cli/memory.py @@ -66,6 +66,44 @@ def get_store(db_path: str) -> SQLiteMemoryStore: return SQLiteMemoryStore(db_path) +def _resolve_memory(store: SQLiteMemoryStore, memory_id: str) -> Memory: + """Resolve an exact or unambiguous partial memory ID.""" + memory = asyncio.run(store.get(memory_id)) + if memory is not None: + return memory + + memories = asyncio.run(store.query(MemoryFilter(limit=10000, include_superseded=True))) + matches = [candidate for candidate in memories if candidate.id.startswith(memory_id)] + if not matches: + raise ValueError(f"Memory not found: {memory_id}") + if len(matches) > 1: + raise ValueError( + f"Ambiguous ID '{memory_id}'. Matches: {[memory.id[:8] for memory in matches]}" + ) + return matches[0] + + +async def _apply_supersession_repair( + db_path: str, + old_memory_id: str, + new_memory_id: str, + vector_dimension: int, +) -> tuple[Memory, Memory]: + """Apply a repair through LocalBackend so indexes and cache are refreshed.""" + from ..memory.backends.local import LocalBackend, LocalBackendConfig + + backend = LocalBackend( + LocalBackendConfig( + db_path=db_path, + vector_dimension=vector_dimension, + ) + ) + try: + return await backend.detach_supersession(old_memory_id, new_memory_id) + finally: + await backend.close() + + def get_scope_label(memory: Memory) -> str: """Get a human-readable scope label for a memory.""" if memory.turn_id is not None: @@ -255,6 +293,7 @@ def memory(ctx: click.Context) -> None: headroom memory stats Show memory statistics headroom memory edit --content ... Edit a memory's content headroom memory delete Delete a memory + headroom memory repair-supersession Repair one lineage edge headroom memory prune --older-than 30d Delete memories older than 30 days headroom memory purge --confirm Delete ALL memories headroom memory export --output file.json Export all memories to JSON @@ -595,6 +634,81 @@ def edit_memory( sys.exit(1) +@memory.command("repair-supersession") +@db_path_option +@click.argument("old_memory_id", type=str) +@click.argument("new_memory_id", type=str) +@click.option( + "--apply", + "apply_change", + is_flag=True, + help="Apply the repair. Stop any proxy using this database first.", +) +@click.pass_context +def repair_supersession( + ctx: click.Context, + db_path: str, + old_memory_id: str, + new_memory_id: str, + apply_change: bool, +) -> None: + """Detach one incorrect OLD_ID -> NEW_ID supersession edge. + + The command validates both reciprocal lineage pointers and changes no + neighboring edges. It is a dry run unless ``--apply`` is provided. + + \b + Examples: + headroom memory repair-supersession OLD_ID NEW_ID + headroom memory repair-supersession OLD_ID NEW_ID --apply + """ + _ = ctx + store = get_store(db_path) + + try: + old_memory = _resolve_memory(store, old_memory_id) + new_memory = _resolve_memory(store, new_memory_id) + + if old_memory.superseded_by != new_memory.id or new_memory.supersedes != old_memory.id: + raise ValueError( + f"Memories {old_memory.id} and {new_memory.id} do not form " + "a reciprocal supersession edge" + ) + + click.echo("\nSupersession repair preview:") + click.echo( + f" Restore old memory: {old_memory.id} " + f"({truncate(old_memory.content.replace(chr(10), ' '), 60)})" + ) + click.echo( + f" Detach new memory: {new_memory.id} " + f"({truncate(new_memory.content.replace(chr(10), ' '), 60)})" + ) + click.echo(" Clear: old.valid_until, old.superseded_by, new.supersedes") + + if not apply_change: + print_warning("DRY RUN: No changes made. Re-run with --apply to repair this edge.") + return + + embedding = ( + old_memory.embedding if old_memory.embedding is not None else new_memory.embedding + ) + vector_dimension = len(embedding) if embedding is not None else 384 + asyncio.run( + _apply_supersession_repair( + db_path, + old_memory.id, + new_memory.id, + vector_dimension, + ) + ) + print_success(f"Detached supersession edge {old_memory.id[:8]} -> {new_memory.id[:8]}.") + + except Exception as e: + print_error(f"Failed to repair supersession: {e}") + sys.exit(1) + + @memory.command("delete") @db_path_option @click.argument("memory_ids", type=str, nargs=-1, required=True) diff --git a/headroom/memory/adapters/sqlite.py b/headroom/memory/adapters/sqlite.py index c1057b4fc..3b779bd91 100644 --- a/headroom/memory/adapters/sqlite.py +++ b/headroom/memory/adapters/sqlite.py @@ -699,6 +699,54 @@ class SQLiteMemoryStore: return new_memory + async def detach_supersession( + self, + old_memory_id: str, + new_memory_id: str, + ) -> tuple[Memory, Memory]: + """Atomically detach one verified supersession edge. + + This is an explicit repair operation. It never infers identity from + content or embedding similarity and leaves neighboring chain edges + untouched. + """ + if old_memory_id == new_memory_id: + raise ValueError("A memory cannot supersede itself") + + with self._get_conn() as conn: + conn.execute("BEGIN IMMEDIATE") + rows = conn.execute( + "SELECT * FROM memories WHERE id IN (?, ?)", + (old_memory_id, new_memory_id), + ).fetchall() + memories = {row["id"]: self._row_to_memory(row) for row in rows} + old_memory = memories.get(old_memory_id) + new_memory = memories.get(new_memory_id) + + if old_memory is None: + raise ValueError(f"Memory {old_memory_id} not found") + if new_memory is None: + raise ValueError(f"Memory {new_memory_id} not found") + if old_memory.superseded_by != new_memory_id or new_memory.supersedes != old_memory_id: + raise ValueError( + f"Memories {old_memory_id} and {new_memory_id} do not form " + "a reciprocal supersession edge" + ) + + conn.execute( + "UPDATE memories SET valid_until = NULL, superseded_by = NULL WHERE id = ?", + (old_memory_id,), + ) + conn.execute( + "UPDATE memories SET supersedes = NULL WHERE id = ?", + (new_memory_id,), + ) + + old_memory.valid_until = None + old_memory.superseded_by = None + new_memory.supersedes = None + return old_memory, new_memory + async def get_history( self, memory_id: str, diff --git a/headroom/memory/backends/local.py b/headroom/memory/backends/local.py index 757ccdbed..1dca91b8e 100644 --- a/headroom/memory/backends/local.py +++ b/headroom/memory/backends/local.py @@ -561,6 +561,19 @@ class LocalBackend: return new_memory + async def detach_supersession( + self, + old_memory_id: str, + new_memory_id: str, + ) -> tuple[Memory, Memory]: + """Detach one explicit supersession edge and refresh indexes.""" + await self._ensure_initialized() + assert self._hierarchical_memory is not None + return await self._hierarchical_memory.detach_supersession( + old_memory_id, + new_memory_id, + ) + async def delete_memory( self, memory_id: str, diff --git a/headroom/memory/core.py b/headroom/memory/core.py index c5b0ccf9f..0265b8e12 100644 --- a/headroom/memory/core.py +++ b/headroom/memory/core.py @@ -585,6 +585,39 @@ class HierarchicalMemory: logger.debug(f"Superseded memory {old_memory_id} with {new_memory.id}") return new_memory + async def detach_supersession( + self, + old_memory_id: str, + new_memory_id: str, + ) -> tuple[Memory, Memory]: + """Detach one explicit, reciprocal supersession edge. + + The store performs the atomic lineage repair. Both affected memories + are then re-indexed so the restored old memory is immediately + searchable and index metadata reflects the repaired lineage. + """ + old_memory, new_memory = await self._store.detach_supersession( + old_memory_id, + new_memory_id, + ) + + for memory in (old_memory, new_memory): + if memory.embedding is not None: + await self._vector_index.index(memory) + await self._index_for_text_search(memory) + + if self._cache is not None: + memory_ids = [old_memory.id, new_memory.id] + await self._cache.invalidate_batch(memory_ids) + await self._cache.put_batch([old_memory, new_memory]) + + logger.info( + "Detached supersession edge %s -> %s", + old_memory_id, + new_memory_id, + ) + return old_memory, new_memory + async def get_history( self, memory_id: str, diff --git a/headroom/memory/ports.py b/headroom/memory/ports.py index 7fd0b7bad..48aa185ca 100644 --- a/headroom/memory/ports.py +++ b/headroom/memory/ports.py @@ -397,6 +397,23 @@ class MemoryStore(Protocol): """ ... + async def detach_supersession( + self, + old_memory_id: str, + new_memory_id: str, + ) -> tuple[Memory, Memory]: + """Detach one explicit edge from a supersession chain. + + The two memories must form a reciprocal direct edge: + ``old.superseded_by == new.id`` and ``new.supersedes == old.id``. + The old memory becomes current again while all other chain edges + remain unchanged. + + Returns: + The updated ``(old_memory, new_memory)`` pair. + """ + ... + async def get_history( self, memory_id: str, diff --git a/tests/test_memory/test_supersession_repair.py b/tests/test_memory/test_supersession_repair.py new file mode 100644 index 000000000..dd4c40663 --- /dev/null +++ b/tests/test_memory/test_supersession_repair.py @@ -0,0 +1,174 @@ +"""Tests for explicit supersession-edge repair.""" + +from __future__ import annotations + +import asyncio +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import numpy as np +import pytest +from click.testing import CliRunner + +import headroom.cli.memory as memory_cli +from headroom.cli.main import main +from headroom.memory.adapters.sqlite import SQLiteMemoryStore +from headroom.memory.core import HierarchicalMemory +from headroom.memory.models import Memory +from headroom.memory.ports import MemoryFilter + + +@pytest.mark.asyncio +async def test_store_detaches_only_the_requested_supersession_edge(tmp_path) -> None: + store = SQLiteMemoryStore(tmp_path / "memory.db") + first = Memory(content="First fact", user_id="alice") + await store.save(first) + second = await store.supersede( + first.id, + Memory(content="Second fact", user_id="alice"), + ) + third = await store.supersede( + second.id, + Memory(content="Third fact", user_id="alice"), + ) + + repaired_first, repaired_second = await store.detach_supersession(first.id, second.id) + + assert repaired_first.valid_until is None + assert repaired_first.superseded_by is None + assert repaired_second.supersedes is None + assert repaired_second.superseded_by == third.id + assert repaired_second.valid_until is not None + assert third.supersedes == second.id + + current = await store.query(MemoryFilter(user_id="alice")) + assert {memory.id for memory in current} == {first.id, third.id} + + +@pytest.mark.asyncio +async def test_store_rejects_non_reciprocal_repair_without_mutation(tmp_path) -> None: + store = SQLiteMemoryStore(tmp_path / "memory.db") + first = Memory(content="First fact", user_id="alice") + unrelated = Memory(content="Unrelated fact", user_id="alice") + await store.save_batch([first, unrelated]) + + with pytest.raises(ValueError, match="reciprocal supersession edge"): + await store.detach_supersession(first.id, unrelated.id) + + assert (await store.get(first.id)).is_current + assert (await store.get(unrelated.id)).is_current + + +@pytest.mark.asyncio +async def test_core_reindexes_and_refreshes_cache_after_repair(tmp_path) -> None: + store = SQLiteMemoryStore(tmp_path / "memory.db") + first = Memory( + content="First fact", + user_id="alice", + embedding=np.array([1.0, 0.0], dtype=np.float32), + ) + await store.save(first) + second = await store.supersede( + first.id, + Memory( + content="Second fact", + user_id="alice", + embedding=np.array([0.0, 1.0], dtype=np.float32), + ), + ) + vector_index = SimpleNamespace(index=AsyncMock()) + text_index = SimpleNamespace(index_memory=AsyncMock()) + cache = SimpleNamespace(invalidate_batch=AsyncMock(), put_batch=AsyncMock()) + system = HierarchicalMemory( + store=store, + vector_index=vector_index, + text_index=text_index, + embedder=SimpleNamespace(), + cache=cache, + ) + + repaired = await system.detach_supersession(first.id, second.id) + + assert [call.args[0].id for call in vector_index.index.await_args_list] == [ + first.id, + second.id, + ] + assert [call.args[0].id for call in text_index.index_memory.await_args_list] == [ + first.id, + second.id, + ] + cache.invalidate_batch.assert_awaited_once_with([first.id, second.id]) + cache.put_batch.assert_awaited_once_with(list(repaired)) + + +def _seed_chain(db_path) -> tuple[Memory, Memory]: + async def seed() -> tuple[Memory, Memory]: + store = SQLiteMemoryStore(db_path) + first = Memory(content="Keep Python preference", user_id="alice") + await store.save(first) + second = await store.supersede( + first.id, + Memory(content="Keep dark mode preference", user_id="alice"), + ) + return first, second + + return asyncio.run(seed()) + + +def test_cli_repair_is_dry_run_by_default(tmp_path) -> None: + db_path = tmp_path / "memory.db" + first, second = _seed_chain(db_path) + + result = CliRunner().invoke( + main, + [ + "memory", + "repair-supersession", + first.id[:8], + second.id[:8], + "--db-path", + str(db_path), + ], + ) + + assert result.exit_code == 0, result.output + assert "DRY RUN" in result.output + store = SQLiteMemoryStore(db_path) + assert asyncio.run(store.get(first.id)).superseded_by == second.id + assert asyncio.run(store.get(second.id)).supersedes == first.id + + +def test_cli_repair_requires_apply_and_uses_full_ids(tmp_path, monkeypatch) -> None: + db_path = tmp_path / "memory.db" + first, second = _seed_chain(db_path) + applied: list[tuple[str, str, int]] = [] + + async def fake_apply( + db_path_arg: str, + old_memory_id: str, + new_memory_id: str, + vector_dimension: int, + ) -> tuple[Memory, Memory]: + applied.append((old_memory_id, new_memory_id, vector_dimension)) + store = SQLiteMemoryStore(db_path_arg) + return await store.detach_supersession(old_memory_id, new_memory_id) + + monkeypatch.setattr(memory_cli, "_apply_supersession_repair", fake_apply) + result = CliRunner().invoke( + main, + [ + "memory", + "repair-supersession", + first.id[:8], + second.id[:8], + "--db-path", + str(db_path), + "--apply", + ], + ) + + assert result.exit_code == 0, result.output + assert applied == [(first.id, second.id, 384)] + store = SQLiteMemoryStore(db_path) + assert asyncio.run(store.get(first.id)).is_current + assert asyncio.run(store.get(second.id)).supersedes is None