akadmin/headroom
1
0
Fork 0
mirror of https://github.com/headroomlabs-ai/headroom.git synced 2026-08-27 14:17:10 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

feat(memory): add explicit supersession repair (#2217)

Browse source 
## Description

Add an explicit, reviewable way to detach one incorrect supersession
edge while preserving both memories and all neighboring version history.

Closes #2216

## Type of Change

- [ ] Bug fix (non-breaking change that fixes an issue)
- [x] New feature (non-breaking change that adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to change)
- [ ] Documentation update
- [ ] Performance improvement
- [ ] Code refactoring (no functional changes)

## Changes Made

- Add an atomic SQLite `detach_supersession(old_id, new_id)` primitive
that requires reciprocal direct lineage.
- Restore only the old memory's validity and clear only the selected
edge.
- Re-index both affected memories and refresh cache state through
`HierarchicalMemory`.
- Expose the operation through `LocalBackend`.
- Add `headroom memory repair-supersession OLD_ID NEW_ID`, dry-run by
default with explicit `--apply`.
- Resolve unambiguous partial IDs for preview but pass full IDs to the
mutation.
- Add chain-locality, rejection, index/cache, dry-run, and apply-path
tests.

## Testing

- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check .`)
- [ ] Type checking passes (`mypy headroom`)
- [x] New tests added for new functionality
- [ ] Manual testing performed

### Test Output

```text
$ uv run --with pytest --with pytest-asyncio --with numpy --with fastapi --with httpx pytest \
    tests/test_memory/test_supersession_repair.py \
    tests/test_memory/test_hierarchical.py::TestSQLiteMemoryStore \
    tests/test_cli/test_main_help_version.py -q
22 passed

$ uv run --with ruff ruff check <touched Python files>
All checks passed!

$ uv run --with ruff ruff format --check <touched Python files>
6 files already formatted
```

## Real Behavior Proof

- Environment: Python 3.13, temporary SQLite databases, synthetic two-
and three-version chains
- Exact command / steps: run the focused test set above
- Observed result: detaching `v1 -> v2` restores `v1` as current, leaves
`v2 -> v3` intact, re-indexes both records, refreshes cache, and keeps
CLI preview read-only until `--apply`
- Not tested: live proxy process, external MemoryStore/VectorIndex
plugins, full repository suite

## Review Readiness

- [x] I have performed a self-review
- [ ] This PR is ready for human review

## Checklist

- [x] My code follows the project's style guidelines
- [x] I have performed a self-review of my code
- [x] I have commented my code, particularly in hard-to-understand areas
- [ ] I have made corresponding changes to the documentation
- [ ] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my
feature works
- [x] New and existing unit tests pass locally with my changes
- [ ] I have updated the CHANGELOG.md if applicable

## Screenshots (if applicable)

N/A

## Additional Notes

This is intentionally separate from #2188: that PR prevents new false
edges, while this PR repairs historical data.

The CLI help requires stopping any proxy that is actively using the same
database before `--apply`, because another process can retain an old
in-memory index snapshot. External backend semantics and stronger
cross-store rollback behavior are left visible for maintainer review
before this Draft is marked ready.

Co-authored-by: JerrettDavis <mxjerrett@gmail.com>
This commit is contained in:
Chester 2026-07-16 02:17:17 +08:00 • committed by GitHub
parent 528517cff8
commit ce52b30c8f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 399 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

114
headroom/cli/memory.py
View file

@ -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 <id> --content ... Edit a memory's content
headroom memory delete <id> Delete a memory
headroom memory repair-supersession <old-id> <new-id> 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)

48
headroom/memory/adapters/sqlite.py
View file

@ -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,

13
headroom/memory/backends/local.py
View file

@ -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,

33
headroom/memory/core.py
View file

@ -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,

17
headroom/memory/ports.py
View file

@ -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,

174
tests/test_memory/test_supersession_repair.py Normal file
View file

@ -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