mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-27 14:17:10 -04:00
9e38905a7d
## Description `SQLiteMemoryStore._build_query_conditions()` dropped `turn_id` when a query specified `session_id` and `turn_id` without also specifying `agent_id`. That made a single-turn scope return every memory in the session, and `count()` uses the same helper. ## Fix - Apply `agent_id` and `turn_id` as independent narrowing predicates inside the `session_id` branch. - Preserve the existing `agent_id`-only behavior. - Add direct query-condition regression tests for turn-only, agent+turn, and agent-only scopes. - Merge current `main` to refresh mergeability and stale lint results. ## Testing ```text uvx ruff@0.15.17 check headroom/memory/adapters/sqlite.py tests/test_memory/test_query_conditions.py headroom/memory/factory.py All checks passed! uvx ruff@0.15.17 format --check headroom/memory/adapters/sqlite.py tests/test_memory/test_query_conditions.py headroom/memory/factory.py 3 files already formatted git diff --check headroomlabs/main...HEAD # no output uv run --extra dev python -m pytest tests/test_memory/test_query_conditions.py -q 3 passed ``` ## Review Readiness - [x] Ready for review - [x] Regression tests added - [x] CHANGELOG updated Co-authored-by: JerrettDavis <mxjerrett@gmail.com> Co-authored-by: Tejas Chopra <chopratejas@gmail.com>
40 lines
1.3 KiB
Python
40 lines
1.3 KiB
Python
"""SQLiteMemoryStore._build_query_conditions scope filtering.
|
|
|
|
`_build_query_conditions` only reads the filter, so it is exercised directly via
|
|
``object.__new__`` (no DB, no embedder).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from headroom.memory.adapters.sqlite import SQLiteMemoryStore
|
|
from headroom.memory.ports import MemoryFilter
|
|
|
|
|
|
def _conditions(**kwargs) -> tuple[list[str], list]:
|
|
store = object.__new__(SQLiteMemoryStore)
|
|
return store._build_query_conditions(MemoryFilter(**kwargs))
|
|
|
|
|
|
def test_turn_id_is_applied_without_agent_id():
|
|
"""A (user, session, turn) filter without agent_id must still narrow to the
|
|
turn — previously the turn_id condition was nested inside the agent_id block
|
|
and silently dropped, returning the whole session."""
|
|
conditions, params = _conditions(user_id="u", session_id="s", turn_id="t")
|
|
|
|
assert "turn_id = ?" in conditions
|
|
assert "t" in params
|
|
|
|
|
|
def test_agent_id_and_turn_id_both_applied():
|
|
conditions, params = _conditions(user_id="u", session_id="s", agent_id="a", turn_id="t")
|
|
|
|
assert "agent_id = ?" in conditions
|
|
assert "turn_id = ?" in conditions
|
|
assert "a" in params and "t" in params
|
|
|
|
|
|
def test_agent_id_only_still_applied():
|
|
conditions, _ = _conditions(user_id="u", session_id="s", agent_id="a")
|
|
|
|
assert "agent_id = ?" in conditions
|
|
assert "turn_id = ?" not in conditions
|