Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 8 additions & 6 deletions headroom/memory/traffic_learner.py
Original file line number Diff line number Diff line change
Expand Up @@ -743,7 +743,7 @@ def _collect_all_patterns(self) -> list[ExtractedPattern]:
db_path = _resolve_backend_db_path(self._backend)
if db_path is not None and db_path.exists():
try:
persisted = _load_persisted_patterns_from_sqlite(db_path)
persisted = _load_persisted_patterns_from_sqlite(db_path, self._user_id)
except Exception as e:
logger.debug("Reading persisted traffic patterns failed: %s", e)
persisted = []
Expand Down Expand Up @@ -1473,7 +1473,8 @@ def _read() -> list[tuple[str, str, str]]:
try:
rows = conn.execute(
"SELECT id, content, metadata FROM memories "
"WHERE json_extract(metadata, '$.source') = 'traffic_learner' "
"WHERE user_id = ? "
"AND json_extract(metadata, '$.source') = 'traffic_learner' "
# Hydrate at most dedup_window rows so a large persisted
# history does not start the in-memory dedup maps oversized
# (they are trimmed to dedup_window in steady state). Keep the
Expand All @@ -1482,7 +1483,7 @@ def _read() -> list[tuple[str, str, str]]:
# as a deterministic tie-break.
"ORDER BY json_extract(metadata, '$.last_seen_at') DESC, id DESC "
"LIMIT ?",
(self._dedup_window,),
(self._user_id, self._dedup_window),
).fetchall()
except sqlite3.DatabaseError as e:
logger.warning("Traffic learner hydrate: failed to read %s: %s", db_path, e)
Expand Down Expand Up @@ -1909,8 +1910,8 @@ def _resolve_backend_db_path(backend: Any) -> Path | None:
return Path(db_path)


def _load_persisted_patterns_from_sqlite(db_path: Path) -> list[ExtractedPattern]:
"""Read traffic_learner rows from memory.db, dedupe, return patterns.
def _load_persisted_patterns_from_sqlite(db_path: Path, user_id: str) -> list[ExtractedPattern]:
"""Read ``user_id``'s traffic_learner rows from memory.db, dedupe, return patterns.

Uses a direct read-only SQLite connection — we don't go through the
backend's vector search because we want all rows, not semantically
Expand All @@ -1928,7 +1929,8 @@ def _load_persisted_patterns_from_sqlite(db_path: Path) -> list[ExtractedPattern
rows = conn.execute(
"SELECT content, metadata, entity_refs, importance, created_at "
"FROM memories "
"WHERE json_extract(metadata, '$.source') = 'traffic_learner'"
"WHERE user_id = ? AND json_extract(metadata, '$.source') = 'traffic_learner'",
(user_id,),
).fetchall()
except sqlite3.DatabaseError as e:
logger.warning("Traffic learner: failed to read persisted patterns %s: %s", db_path, e)
Expand Down
95 changes: 84 additions & 11 deletions tests/test_memory/test_traffic_learner.py
Original file line number Diff line number Diff line change
Expand Up @@ -771,6 +771,7 @@ def _make_db(self, tmp_path, rows: list[dict]):
conn.execute(
"CREATE TABLE memories ("
"id TEXT PRIMARY KEY, content TEXT NOT NULL, "
"user_id TEXT NOT NULL DEFAULT 'default', "
"metadata TEXT NOT NULL DEFAULT '{}', "
"entity_refs TEXT NOT NULL DEFAULT '[]', "
"importance REAL NOT NULL DEFAULT 0.5, "
Expand Down Expand Up @@ -816,7 +817,7 @@ def test_dedupes_by_content_and_sums_evidence(self, tmp_path):
},
],
)
patterns = _load_persisted_patterns_from_sqlite(db)
patterns = _load_persisted_patterns_from_sqlite(db, "default")
assert len(patterns) == 1
assert patterns[0].evidence_count == 5
assert patterns[0].category == PatternCategory.ERROR_RECOVERY
Expand All @@ -838,7 +839,7 @@ def test_skips_non_traffic_rows(self, tmp_path):
},
],
)
patterns = _load_persisted_patterns_from_sqlite(db)
patterns = _load_persisted_patterns_from_sqlite(db, "default")
assert len(patterns) == 1
assert patterns[0].content == "From traffic"

Expand All @@ -856,7 +857,7 @@ def test_reads_importance_column(self, tmp_path):
},
],
)
patterns = _load_persisted_patterns_from_sqlite(db)
patterns = _load_persisted_patterns_from_sqlite(db, "default")
assert len(patterns) == 1
assert patterns[0].importance == 0.85

Expand All @@ -870,7 +871,7 @@ def test_skips_unknown_category(self, tmp_path):
},
],
)
assert _load_persisted_patterns_from_sqlite(db) == []
assert _load_persisted_patterns_from_sqlite(db, "default") == []


# =============================================================================
Expand Down Expand Up @@ -1567,6 +1568,74 @@ async def test_cancelled_save_survives_restart(self, tmp_path):
assert rows[0][2]["evidence_count"] == 6


class TestUserScopedPersistence:
"""Learners sharing one memory.db only read their own user's rows."""

@pytest.mark.asyncio
async def test_hydration_ignores_other_users_rows(self, tmp_path, monkeypatch):
import json
import sqlite3

monkeypatch.setenv("HOME", str(tmp_path / "home"))
(tmp_path / "home").mkdir()
db = tmp_path / "memory.db"
_init_db(db)
backend = _FakeBackend(db)
content = "Use uv for Python packages."
await backend.save_memory(
content=content,
user_id="alice",
importance=0.7,
metadata={
"source": "traffic_learner",
"category": "preference",
"evidence_count": 3,
"last_seen_at": "2026-02-01T00:00:00+00:00",
},
)
learner = TrafficLearner(backend=backend, user_id="bob", min_evidence=3, dedup_window=1)
await learner.start()
try:
for _ in range(3):
await learner._accumulate(
ExtractedPattern(
category=PatternCategory.PREFERENCE, content=content, importance=0.7
)
)
await _wait_for_saved(learner, 2, db)
finally:
await learner.stop()

with sqlite3.connect(db) as conn:
rows = conn.execute("SELECT user_id, metadata FROM memories").fetchall()
evidence = {user_id: json.loads(meta)["evidence_count"] for user_id, meta in rows}
assert evidence == {"alice": 3, "bob": 3}

@pytest.mark.asyncio
async def test_collected_patterns_exclude_other_users(self, tmp_path):
db = tmp_path / "memory.db"
_init_db(db)
backend = _FakeBackend(db)
for user_id, content in (
("alice", "Alice-only deployment procedure."),
("bob", "Bob's package workflow."),
):
await backend.save_memory(
content=content,
user_id=user_id,
importance=0.7,
metadata={
"source": "traffic_learner",
"category": "preference",
"evidence_count": 5,
},
)

learner = TrafficLearner(backend=backend, user_id="bob")
contents = [p.content for p in learner._collect_all_patterns()]
assert contents == ["Bob's package workflow."]


class TestEvidencePersistence:
@pytest.mark.asyncio
@pytest.mark.parametrize("min_evidence", [1, 2, 5])
Expand Down Expand Up @@ -2023,7 +2092,7 @@ async def test_merges_db_and_accumulator(self, tmp_path):
},
)

learner = TrafficLearner(backend=backend, min_evidence=1)
learner = TrafficLearner(backend=backend, user_id="t", min_evidence=1)
# Same content in accumulator with count=2; hash matches.
p = ExtractedPattern(
category=PatternCategory.ENVIRONMENT,
Expand Down Expand Up @@ -2596,6 +2665,7 @@ def _make_db(self, tmp_path, rows: list[dict]):
conn.execute(
"CREATE TABLE memories ("
"id TEXT PRIMARY KEY, content TEXT NOT NULL, "
"user_id TEXT NOT NULL DEFAULT 'default', "
"metadata TEXT NOT NULL DEFAULT '{}', "
"entity_refs TEXT NOT NULL DEFAULT '[]', "
"importance REAL NOT NULL DEFAULT 0.5, "
Expand Down Expand Up @@ -2635,7 +2705,7 @@ def test_reads_timestamps_from_metadata(self, tmp_path):
}
],
)
patterns = _load_persisted_patterns_from_sqlite(db)
patterns = _load_persisted_patterns_from_sqlite(db, "default")
assert len(patterns) == 1
p = patterns[0]
assert p.first_seen_at is not None
Expand All @@ -2659,7 +2729,7 @@ def test_falls_back_to_created_at(self, tmp_path):
}
],
)
patterns = _load_persisted_patterns_from_sqlite(db)
patterns = _load_persisted_patterns_from_sqlite(db, "default")
assert len(patterns) == 1
assert patterns[0].first_seen_at is not None
assert patterns[0].first_seen_at.month == 3
Expand Down Expand Up @@ -2695,7 +2765,7 @@ def test_collision_merges_timestamps_max_last_min_first(self, tmp_path):
},
],
)
patterns = _load_persisted_patterns_from_sqlite(db)
patterns = _load_persisted_patterns_from_sqlite(db, "default")
assert len(patterns) == 1
p = patterns[0]
assert p.evidence_count == 5
Expand All @@ -2714,6 +2784,7 @@ def test_non_numeric_importance_falls_back_to_default(self, tmp_path):
conn.execute(
"CREATE TABLE memories ("
"id TEXT PRIMARY KEY, content TEXT NOT NULL, "
"user_id TEXT NOT NULL DEFAULT 'default', "
"metadata TEXT NOT NULL DEFAULT '{}', "
"entity_refs TEXT NOT NULL DEFAULT '[]', "
"importance TEXT, "
Expand All @@ -2736,7 +2807,7 @@ def test_non_numeric_importance_falls_back_to_default(self, tmp_path):
)
conn.commit()
conn.close()
patterns = _load_persisted_patterns_from_sqlite(db)
patterns = _load_persisted_patterns_from_sqlite(db, "default")
assert len(patterns) == 1
assert patterns[0].importance == 0.5

Expand All @@ -2749,20 +2820,22 @@ def test_malformed_metadata_json_skipped_gracefully(self, tmp_path):
conn.execute(
"CREATE TABLE memories ("
"id TEXT PRIMARY KEY, content TEXT NOT NULL, "
"user_id TEXT NOT NULL DEFAULT 'default', "
"metadata TEXT NOT NULL DEFAULT '{}', "
"entity_refs TEXT NOT NULL DEFAULT '[]', "
"importance REAL NOT NULL DEFAULT 0.5, "
"created_at TEXT)"
)
# Invalid JSON in metadata
conn.execute(
"INSERT INTO memories VALUES (?,?,?,?,?,?)",
"INSERT INTO memories (id, content, metadata, entity_refs, importance, created_at) "
"VALUES (?,?,?,?,?,?)",
("0", "bullet", "{not json", "[]", 0.5, None),
)
conn.commit()
conn.close()
# Should not raise — the row is simply skipped (no recognizable category).
patterns = _load_persisted_patterns_from_sqlite(db)
patterns = _load_persisted_patterns_from_sqlite(db, "default")
assert patterns == []


Expand Down
Loading