brooklyn!
642 lines
28 KiB
Python
642 lines
28 KiB
Python
"""Tests for SessionDB.get_messages(include_compacted=...).
|
|
|
|
In-place compaction archives earlier turns as ``active=0, compacted=1`` rows
|
|
that are durable display history, not soft-deleted rows. A transcript read
|
|
that drops them silently cuts the user-visible conversation off at the
|
|
compaction boundary (#80680): the UI exhausts its active-only window, "Show
|
|
earlier messages" disappears, and earlier turns become unreachable even
|
|
though they are still on disk.
|
|
|
|
``include_compacted=True`` must surface those rows while still excluding
|
|
soft-deleted Undo/Rewind rows (``active=0, compacted=0``) — that remains the
|
|
job of ``include_inactive`` (audit / debug reads).
|
|
"""
|
|
|
|
import json
|
|
import sqlite3
|
|
import tracemalloc
|
|
|
|
import pytest
|
|
|
|
from agent.context_compressor import (
|
|
HISTORICAL_TASK_HEADING,
|
|
SUMMARY_PREFIX,
|
|
_MERGED_PRIOR_CONTEXT_HEADER,
|
|
_MERGED_SUMMARY_DELIMITER,
|
|
_SUMMARY_END_MARKER,
|
|
)
|
|
from hermes_state import SessionDB
|
|
|
|
|
|
@pytest.fixture
|
|
def db(tmp_path):
|
|
return SessionDB(tmp_path / "state.db")
|
|
|
|
|
|
def _seed(db, sid="s1"):
|
|
"""Session with 4 archived (compacted) turns + 2 live turns + 1 rewound row."""
|
|
db.create_session(sid, source="cli")
|
|
old = [
|
|
{"role": "user", "content": "old q1"},
|
|
{"role": "assistant", "content": "old a1"},
|
|
{"role": "user", "content": "old q2"},
|
|
{"role": "assistant", "content": "old a2"},
|
|
]
|
|
db.append_messages_batch(sid, old)
|
|
db.archive_and_compact(
|
|
sid,
|
|
[
|
|
{"role": "assistant", "content": "summary of old turns"},
|
|
{"role": "user", "content": "live q1"},
|
|
{"role": "assistant", "content": "live a1"},
|
|
],
|
|
)
|
|
# Soft-delete the last live user turn (active=0, compacted=0) so every
|
|
# row class is present: active=1/compacted=0, active=0/compacted=1,
|
|
# active=0/compacted=0. rewind_to_message requires a user target.
|
|
live = db.get_messages(sid)
|
|
user_msg = next(m for m in reversed(live) if m["role"] == "user")
|
|
db.rewind_to_message(sid, user_msg["id"])
|
|
return db
|
|
|
|
|
|
def _row_ids(db, sid, **kwargs):
|
|
return [m["id"] for m in db.get_messages(sid, **kwargs)]
|
|
|
|
|
|
class TestIncludeCompacted:
|
|
def test_default_returns_only_active_rows(self, db):
|
|
"""Regression guard: the default read must not change behaviour."""
|
|
sid = "s1"
|
|
db = _seed(db, sid)
|
|
msgs = db.get_messages(sid)
|
|
assert all(m["active"] for m in msgs)
|
|
# Only the compaction summary survived (the rewind soft-deleted
|
|
# the live user turn AND everything after it); the 4 archived rows
|
|
# stay hidden.
|
|
assert len(msgs) == 1
|
|
|
|
def test_include_compacted_surfaces_archived_rows(self, db):
|
|
sid = "s1"
|
|
db = _seed(db, sid)
|
|
msgs = db.get_messages(sid, include_compacted=True)
|
|
# 4 archived + 1 live (the summary); the 2 rewound rows are excluded.
|
|
assert len(msgs) == 5
|
|
assert all(m["active"] or m["compacted"] for m in msgs)
|
|
# Archived rows are the oldest — they come first in insertion order.
|
|
assert msgs[0]["content"] == "old q1"
|
|
assert msgs[-1]["content"] == "summary of old turns"
|
|
|
|
def test_include_compacted_excludes_soft_deleted_rows(self, db):
|
|
"""Undo/Rewind rows (active=0, compacted=0) stay hidden."""
|
|
sid = "s1"
|
|
db = _seed(db, sid)
|
|
msgs = db.get_messages(sid, include_compacted=True)
|
|
assert not any(not m["active"] and not m["compacted"] for m in msgs)
|
|
|
|
@pytest.mark.parametrize("read_only", [False, True])
|
|
@pytest.mark.parametrize("latest", [False, True])
|
|
@pytest.mark.parametrize("limit,offset", [(3, 1), (0, 0), (None, 2), (3, 100), (-1, 0), (3, -1)])
|
|
def test_include_inactive_compacted_paging_keeps_rewound_rows(self, tmp_path, read_only, latest, limit, offset):
|
|
path = tmp_path / "audit.db"
|
|
db = _seed(SessionDB(path))
|
|
expected = db.get_messages("s1", include_inactive=True)
|
|
db._execute_write(lambda conn: conn.execute(
|
|
"UPDATE messages SET display_order = NULL, display_identity = NULL"))
|
|
if read_only:
|
|
db.close()
|
|
db = SessionDB(path, read_only=True)
|
|
try:
|
|
all_rows = db.get_messages("s1", include_compacted=True, include_inactive=True)
|
|
assert all_rows == expected
|
|
page = db.get_messages("s1", include_compacted=True, include_inactive=True,
|
|
limit=limit, offset=offset, latest=latest)
|
|
assert page == (expected[::-1][offset:][:limit][::-1] if latest else expected[offset:][:limit])
|
|
finally:
|
|
db.close()
|
|
|
|
def test_include_inactive_still_returns_everything(self, db):
|
|
"""Audit semantics are unchanged: include_inactive wins."""
|
|
sid = "s1"
|
|
db = _seed(db, sid)
|
|
msgs = db.get_messages(sid, include_inactive=True)
|
|
assert len(msgs) == 7 # 4 archived + 1 live + 2 rewound
|
|
|
|
def test_latest_page_with_compacted_rows(self, db):
|
|
"""latest=True pages back from the newest message, still in order."""
|
|
sid = "s1"
|
|
db = _seed(db, sid)
|
|
ids = _row_ids(db, sid, include_compacted=True, latest=True)
|
|
all_ids = _row_ids(db, sid, include_compacted=True)
|
|
# The whole display history fits one page; latest pages are returned
|
|
# in chronological order (offset measured back from the newest row).
|
|
assert ids == all_ids
|
|
# A bounded page still lands on the newest rows.
|
|
tail = db.get_messages(sid, include_compacted=True, latest=True, limit=3)
|
|
assert [m["id"] for m in tail] == all_ids[-3:]
|
|
|
|
def test_pagination_with_compacted_rows(self, db):
|
|
"""limit/offset pages over the combined display history."""
|
|
sid = "s1"
|
|
db = _seed(db, sid)
|
|
page = db.get_messages(sid, include_compacted=True, limit=3, offset=2)
|
|
all_ids = _row_ids(db, sid, include_compacted=True)
|
|
assert [m["id"] for m in page] == all_ids[2:5]
|
|
|
|
|
|
class TestDisplayDedupe:
|
|
"""Compaction epochs copy the protected tail into each new generation, so
|
|
the same logical message exists as several rows (identical
|
|
role/content/timestamp). The display read must surface it exactly once.
|
|
"""
|
|
|
|
def _copy_tail_as_new_generation(self, db, sid, ids):
|
|
"""Simulate one compaction epoch: duplicate rows as active=0,
|
|
compacted=1 with the SAME content and timestamp (the real
|
|
copy-protected-tail behaviour)."""
|
|
|
|
def _do(conn):
|
|
placeholders = ",".join("?" * len(ids))
|
|
conn.execute(
|
|
f"""
|
|
INSERT INTO messages
|
|
(session_id, role, content, tool_call_id, tool_calls,
|
|
tool_name, timestamp, active, compacted)
|
|
SELECT session_id, role, content, tool_call_id, tool_calls,
|
|
tool_name, timestamp, 0, 1
|
|
FROM messages
|
|
WHERE session_id = ? AND id IN ({placeholders})
|
|
""",
|
|
[sid, *ids],
|
|
)
|
|
|
|
db._execute_write(_do)
|
|
|
|
@pytest.mark.parametrize("read_only", [False, True])
|
|
def test_legacy_page_retains_only_bounded_payloads(self, tmp_path, read_only):
|
|
"""Benjamin Brumbaugh's PR #106838: backfill and fallback retain identities, not payloads."""
|
|
path = tmp_path / "legacy-large.db"
|
|
db = SessionDB(path)
|
|
sid = "legacy-large"
|
|
payload_size = 2_000_000
|
|
db.create_session(sid, source="desktop")
|
|
db.append_messages_batch(
|
|
sid, [{"role": "assistant", "content": f"small-{i}"} for i in range(1_001)],
|
|
chunk_rows=500)
|
|
first_id = db.get_messages(sid, limit=1)[0]["id"]
|
|
self._copy_tail_as_new_generation(db, sid, [first_id])
|
|
db.append_messages_batch(sid, [
|
|
{"role": "assistant", "content": chr(65 + i) * payload_size} for i in range(12)])
|
|
db._execute_write(lambda conn: conn.execute(
|
|
"UPDATE messages SET display_order = NULL, display_identity = NULL WHERE session_id = ?", (sid,)))
|
|
if read_only:
|
|
db.close()
|
|
db = SessionDB(path, read_only=True)
|
|
try:
|
|
tracemalloc.start()
|
|
try:
|
|
page = db.get_messages(sid, include_compacted=True, latest=True, limit=1)
|
|
_, peak = tracemalloc.get_traced_memory()
|
|
finally:
|
|
tracemalloc.stop()
|
|
assert page[0]["content"] == "L" * payload_size
|
|
assert peak < payload_size * 5
|
|
if not read_only:
|
|
missing = db._read_one(
|
|
"SELECT COUNT(*) FROM messages WHERE display_order IS NULL OR display_identity IS NULL")
|
|
assert missing is not None and missing[0] == 0
|
|
assert [row[0] for row in db._read_all(
|
|
"SELECT display_order FROM messages WHERE content = ? ORDER BY id", ("small-0",))] == [first_id, first_id]
|
|
finally:
|
|
db.close()
|
|
|
|
@pytest.mark.parametrize("journal_mode", [
|
|
pytest.param("wal", marks=pytest.mark.requires_wal),
|
|
"delete",
|
|
])
|
|
def test_legacy_page_keeps_one_snapshot_during_rewind(self, tmp_path, monkeypatch, journal_mode):
|
|
"""WAL commits beside the snapshot; rollback journaling waits for its release."""
|
|
monkeypatch.setattr("hermes_state_wal.resolve_journal_mode", lambda: journal_mode)
|
|
path = tmp_path / "snapshot.db"
|
|
writer = SessionDB(path)
|
|
assert writer._read_one("PRAGMA journal_mode")[0] == journal_mode
|
|
writer.create_session("snapshot", source="desktop")
|
|
writer.append_message("snapshot", "assistant", "visible-at-scan")
|
|
writer._execute_write(lambda conn: conn.execute(
|
|
"UPDATE messages SET display_order = NULL, display_identity = NULL"))
|
|
writer._conn.execute("PRAGMA busy_timeout = 0")
|
|
reader = SessionDB(path, read_only=True)
|
|
identity = reader._display_identity
|
|
|
|
def rewind():
|
|
writer._execute_write(lambda conn: conn.execute(
|
|
"UPDATE messages SET content = 'rewound', active = 0, compacted = 0"), patience_s=0)
|
|
|
|
def rewind_after_identity(key):
|
|
result = identity(key)
|
|
if journal_mode == "wal":
|
|
rewind()
|
|
else:
|
|
# This callback runs inside the reader. Waiting for its own
|
|
# transaction to end would deadlock on rollback-journal builds.
|
|
with pytest.raises(sqlite3.OperationalError, match="locked"):
|
|
rewind()
|
|
return result
|
|
|
|
monkeypatch.setattr(reader, "_display_identity", rewind_after_identity)
|
|
try:
|
|
page = reader.get_messages("snapshot", include_compacted=True, latest=True, limit=1)
|
|
assert [(row["active"], row["content"]) for row in page] == [(1, "visible-at-scan")]
|
|
assert reader._conn is not None
|
|
assert not reader._conn.in_transaction
|
|
monkeypatch.setattr(reader, "_display_identity", identity)
|
|
if journal_mode == "delete":
|
|
rewind()
|
|
assert reader.get_messages("snapshot", include_compacted=True) == []
|
|
finally:
|
|
reader.close()
|
|
writer.close()
|
|
|
|
def test_legacy_page_preserves_aborted_transaction_error(self, tmp_path, monkeypatch):
|
|
path = tmp_path / "read-error.db"
|
|
writer = SessionDB(path)
|
|
writer.create_session("error", source="desktop")
|
|
writer.append_message("error", "assistant", "message")
|
|
writer._execute_write(lambda conn: conn.execute(
|
|
"UPDATE messages SET display_order = NULL, display_identity = NULL"))
|
|
writer.close()
|
|
reader = SessionDB(path, read_only=True)
|
|
identity = reader._display_identity
|
|
connection = reader._conn
|
|
assert connection is not None
|
|
|
|
def abort_then_fail(key):
|
|
connection.execute("ROLLBACK")
|
|
raise sqlite3.OperationalError("disk I/O error")
|
|
|
|
monkeypatch.setattr(reader, "_display_identity", abort_then_fail)
|
|
try:
|
|
with pytest.raises(sqlite3.OperationalError, match="disk I/O error"):
|
|
reader.get_messages("error", include_compacted=True, limit=1)
|
|
assert not connection.in_transaction
|
|
monkeypatch.setattr(reader, "_display_identity", identity)
|
|
assert reader.get_messages("error", include_compacted=True)[0]["content"] == "message"
|
|
finally:
|
|
reader.close()
|
|
|
|
def test_display_paging_and_append_work_is_bounded(self, db):
|
|
"""Page and identity-lookup work scale with the page, not the transcript: 10x rows
|
|
must not cost 10x SQLite VM steps (the pre-index read deduped the whole session)."""
|
|
for sid, count in (("small", 2_000), ("large", 20_000)):
|
|
db.create_session(sid, source="desktop")
|
|
db.append_messages_batch(
|
|
sid,
|
|
[{"role": "assistant", "content": f"row-{index}"} for index in range(count)],
|
|
chunk_rows=500,
|
|
)
|
|
|
|
db._wal_active = False
|
|
|
|
def progress_steps(sid):
|
|
callbacks = 0
|
|
|
|
def progress():
|
|
nonlocal callbacks
|
|
callbacks += 1
|
|
return 0
|
|
|
|
db._conn.set_progress_handler(progress, 100)
|
|
try:
|
|
page = db.get_messages(
|
|
sid, include_compacted=True, latest=True, limit=120)
|
|
finally:
|
|
db._conn.set_progress_handler(None, 0)
|
|
assert len(page) == 120
|
|
return callbacks
|
|
|
|
small_steps = progress_steps("small")
|
|
large_steps = progress_steps("large")
|
|
assert large_steps < small_steps * 3
|
|
|
|
def seed(sid, count):
|
|
db.create_session(sid, source="desktop")
|
|
db._execute_write(lambda conn: conn.executemany(
|
|
"INSERT INTO messages (session_id, role, content, timestamp, display_order) "
|
|
"VALUES (?, 'assistant', ?, ?, ?)",
|
|
[(sid, f"row-{index}", 100_000.0, index + 1) for index in range(count)],
|
|
))
|
|
|
|
seed("write-small", 2_000)
|
|
seed("write-large", 20_000)
|
|
db._wal_active = False
|
|
|
|
def append_steps(sid):
|
|
callbacks = 0
|
|
|
|
def progress():
|
|
nonlocal callbacks
|
|
callbacks += 1
|
|
return 0
|
|
|
|
db._conn.set_progress_handler(progress, 10)
|
|
try:
|
|
db.append_message(sid, role="user", content="fresh", timestamp=100_000.0)
|
|
finally:
|
|
db._conn.set_progress_handler(None, 0)
|
|
return callbacks
|
|
|
|
small_steps = append_steps("write-small")
|
|
large_steps = append_steps("write-large")
|
|
assert large_steps < small_steps * 3
|
|
|
|
@pytest.mark.parametrize("read_only", [False, True])
|
|
def test_composite_handoff_keeps_live_turn_identity_and_first_position(self, db, read_only):
|
|
sid = "composite"
|
|
db.create_session(sid, source="desktop")
|
|
original_id = db.append_message(sid, role="user", content="live ask", timestamp=100.0)
|
|
later_id = db.append_message(sid, role="assistant", content="later answer", timestamp=200.0)
|
|
db._execute_write(lambda conn: conn.execute(
|
|
"UPDATE messages SET active = 0, compacted = 1 WHERE session_id = ?", (sid,)))
|
|
carrier = (
|
|
f"{_MERGED_PRIOR_CONTEXT_HEADER}\n"
|
|
"live ask\n\n"
|
|
f"{_MERGED_SUMMARY_DELIMITER}\n\n"
|
|
f"{SUMMARY_PREFIX}\n\n"
|
|
f"{HISTORICAL_TASK_HEADING}\nold work\n\n"
|
|
f"{_SUMMARY_END_MARKER}"
|
|
)
|
|
carrier_id = db.append_message(sid, role="user", content=carrier, timestamp=100.0)
|
|
# Simulate rows written before display_order existed; lazy backfill must use
|
|
# the same normalized user-turn key as the historical Python projection.
|
|
db._execute_write(lambda conn: conn.execute(
|
|
"UPDATE messages SET display_order = NULL WHERE session_id = ?", (sid,)))
|
|
|
|
reader = SessionDB(db.db_path, read_only=True) if read_only else db
|
|
try:
|
|
messages = reader.get_messages(sid, include_compacted=True)
|
|
finally:
|
|
if read_only:
|
|
reader.close()
|
|
|
|
assert [message["id"] for message in messages] == [carrier_id, later_id]
|
|
assert messages[0]["content"] == carrier
|
|
assert original_id not in [message["id"] for message in messages]
|
|
# Index columns never escape the message dict (BLOB identity is not JSON-serializable).
|
|
assert not {"display_identity", "display_order"} & set(messages[0])
|
|
json.dumps(messages)
|
|
# SQLite stores -0.0 and 0.0 as one value; the hashed identity must agree with that.
|
|
db.append_message(sid, role="assistant", content="same", timestamp=-0.0)
|
|
newest_id = db.append_message(sid, role="assistant", content="same", timestamp=0.0)
|
|
assert [m["id"] for m in db.get_messages(sid, include_compacted=True)][-1:] == [newest_id]
|
|
assert len([m for m in db.get_messages(sid, include_compacted=True) if m["content"] == "same"]) == 1
|
|
|
|
def test_display_metadata_update_does_not_invalidate_display_identity(self, tmp_path):
|
|
# Install the pre-narrowing trigger (``display_metadata`` in UPDATE OF, no WHEN) as an existing
|
|
# store would carry it; the reopen must replace it, since IF NOT EXISTS alone never would.
|
|
path = tmp_path / "state.db"
|
|
SessionDB(path).close()
|
|
conn = sqlite3.connect(path)
|
|
conn.execute("DROP TRIGGER IF EXISTS messages_display_identity_update")
|
|
conn.execute(
|
|
"CREATE TRIGGER messages_display_identity_update AFTER UPDATE OF role, content, timestamp, "
|
|
"tool_call_id, tool_calls, tool_name, display_kind, display_metadata ON messages "
|
|
"BEGIN UPDATE messages SET display_identity = NULL, display_order = NULL WHERE id = new.id; END")
|
|
conn.commit()
|
|
conn.close()
|
|
db = SessionDB(path)
|
|
trigger_sql = db._conn.execute(
|
|
"SELECT sql FROM sqlite_master WHERE name = 'messages_display_identity_update'").fetchone()[0]
|
|
assert "display_metadata" not in trigger_sql and "display_kind" in trigger_sql
|
|
|
|
sid = "metadata"
|
|
db.create_session(sid, source="desktop")
|
|
row_ids = [
|
|
db.append_message(sid, role="assistant", content=f"row-{index}", timestamp=100.0)
|
|
for index in range(3)
|
|
]
|
|
before = db._conn.execute(
|
|
"SELECT id, display_order, display_identity FROM messages WHERE session_id = ? ORDER BY id",
|
|
(sid,),
|
|
).fetchall()
|
|
|
|
changes = db._conn.total_changes
|
|
db.set_message_reaction(sid, row_ids[1], "👍", author="user")
|
|
assert db._conn.total_changes - changes == 1
|
|
assert db._conn.execute(
|
|
"SELECT id, display_order, display_identity FROM messages WHERE session_id = ? ORDER BY id",
|
|
(sid,),
|
|
).fetchall() == before
|
|
|
|
changes = db._conn.total_changes
|
|
db.get_messages(sid, include_compacted=True)
|
|
assert db._conn.total_changes == changes
|
|
|
|
def test_display_backfill_rewrites_only_changed_identity_group(self, db):
|
|
sid = "identity-update"
|
|
db.create_session(sid, source="desktop")
|
|
first = db.append_message(sid, role="assistant", content="same", timestamp=100.0)
|
|
db.append_message(sid, role="assistant", content="same", timestamp=100.0)
|
|
unaffected = db.append_message(sid, role="assistant", content="other", timestamp=100.0)
|
|
unaffected_before = db._conn.execute(
|
|
"SELECT display_order, display_identity FROM messages WHERE id = ?", (unaffected,)
|
|
).fetchone()
|
|
db._execute_write(lambda conn: conn.execute(
|
|
"UPDATE messages SET content = ? WHERE id = ?", ('"changed"', first)))
|
|
|
|
changes = db._conn.total_changes
|
|
db.get_messages(sid, include_compacted=True)
|
|
assert db._conn.total_changes - changes == 2
|
|
assert db._conn.execute(
|
|
"SELECT COUNT(*) FROM messages WHERE session_id = ? "
|
|
"AND (display_order IS NULL OR display_identity IS NULL)",
|
|
(sid,),
|
|
).fetchone()[0] == 0
|
|
assert db._conn.execute(
|
|
"SELECT display_order, display_identity FROM messages WHERE id = ?", (unaffected,)
|
|
).fetchone() == unaffected_before
|
|
|
|
def test_legacy_store_is_readable_then_lazily_migrated(self, tmp_path):
|
|
path = tmp_path / "legacy.db"
|
|
writer = SessionDB(path)
|
|
writer.create_session("legacy", source="desktop")
|
|
writer.append_message("legacy", role="assistant", content="same", timestamp=1.0)
|
|
writer.append_message("legacy", role="assistant", content="same", timestamp=1.0)
|
|
new_store_missing = writer._read_one(
|
|
"SELECT COUNT(*) FROM messages "
|
|
"WHERE display_order IS NULL OR display_identity IS NULL")
|
|
assert new_store_missing is not None and new_store_missing[0] == 0
|
|
writer.close()
|
|
|
|
conn = sqlite3.connect(path)
|
|
conn.execute("DROP TRIGGER IF EXISTS messages_display_order_insert")
|
|
conn.execute("DROP TRIGGER IF EXISTS messages_display_visibility_update")
|
|
conn.execute("DROP TRIGGER IF EXISTS messages_display_identity_update")
|
|
conn.execute("DROP TRIGGER IF EXISTS messages_display_identity_delete")
|
|
conn.execute("DROP INDEX IF EXISTS idx_messages_display_page")
|
|
conn.execute("DROP INDEX IF EXISTS idx_messages_display_backfill")
|
|
conn.execute("DROP INDEX IF EXISTS idx_messages_display_identity")
|
|
conn.execute("DROP INDEX IF EXISTS idx_messages_session_id")
|
|
columns = {row[1] for row in conn.execute("PRAGMA table_info(messages)")}
|
|
for column in ("display_order", "display_identity"):
|
|
if column in columns:
|
|
conn.execute(f"ALTER TABLE messages DROP COLUMN {column}")
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
reader = SessionDB(path, read_only=True)
|
|
try:
|
|
legacy_messages = reader.get_messages("legacy", include_compacted=True)
|
|
assert len(legacy_messages) == 1
|
|
json.dumps(legacy_messages)
|
|
assert "display_order" not in {
|
|
row[1] for row in reader._conn.execute("PRAGMA table_info(messages)")}
|
|
finally:
|
|
reader.close()
|
|
|
|
migrated = SessionDB(path)
|
|
try:
|
|
assert len(migrated.get_messages("legacy", include_compacted=True)) == 1
|
|
columns = {row[1] for row in migrated._conn.execute("PRAGMA table_info(messages)")}
|
|
assert {"display_order", "display_identity"} <= columns
|
|
assert migrated._conn.execute(
|
|
"SELECT COUNT(*) FROM messages "
|
|
"WHERE display_order IS NULL OR display_identity IS NULL").fetchone()[0] == 0
|
|
finally:
|
|
migrated.close()
|
|
|
|
def test_copied_protected_tail_is_surfaced_once(self, db):
|
|
"""A message copied across compaction epochs appears exactly once."""
|
|
sid = "s1"
|
|
db.create_session(sid, source="cli")
|
|
db.append_messages_batch(
|
|
sid,
|
|
[
|
|
{"role": "user", "content": "turn 1"},
|
|
{"role": "assistant", "content": "answer 1"},
|
|
],
|
|
)
|
|
orig = _row_ids(db, sid)
|
|
self._copy_tail_as_new_generation(db, sid, orig)
|
|
msgs = db.get_messages(sid, include_compacted=True)
|
|
# 2 logical messages, not 4 (the copies are duplicates).
|
|
assert len(msgs) == 2
|
|
assert [m["content"] for m in msgs] == ["turn 1", "answer 1"]
|
|
|
|
def test_new_generation_copy_keeps_original_chronological_position(self, db):
|
|
"""A protected-tail copy inserted after a newer message stays in its
|
|
original position in the display read (C, A, B regression)."""
|
|
sid = "s1"
|
|
db.create_session(sid, source="cli")
|
|
db.append_messages_batch(
|
|
sid,
|
|
[
|
|
{"role": "assistant", "content": "A", "timestamp": 100.0},
|
|
{"role": "assistant", "content": "B", "timestamp": 200.0},
|
|
],
|
|
)
|
|
original = _row_ids(db, sid)
|
|
db._execute_write(
|
|
lambda conn: conn.execute(
|
|
"UPDATE messages SET active = 0, compacted = 1 WHERE session_id = ?",
|
|
[sid],
|
|
)
|
|
)
|
|
db.append_message(sid, role="user", content="C", timestamp=300.0)
|
|
self._copy_tail_as_new_generation(db, sid, original)
|
|
|
|
msgs = db.get_messages(sid, include_compacted=True)
|
|
|
|
assert [m["content"] for m in msgs] == ["A", "B", "C"]
|
|
|
|
def test_dedupe_prefers_live_row_then_newest_generation(self, db):
|
|
"""When generations conflict, the live row wins; otherwise the newest
|
|
generation (highest id) wins."""
|
|
sid = "s1"
|
|
db.create_session(sid, source="cli")
|
|
db.append_messages_batch(sid, [{"role": "user", "content": "dup q"}])
|
|
gen1 = _row_ids(db, sid)
|
|
self._copy_tail_as_new_generation(db, sid, gen1) # compacted copy
|
|
msgs = db.get_messages(sid, include_compacted=True)
|
|
assert len(msgs) == 1
|
|
assert msgs[0]["active"] == 1 # live row wins
|
|
|
|
# Archive the live row and copy again: the newest compacted copy wins.
|
|
db._execute_write(
|
|
lambda conn: conn.execute(
|
|
"UPDATE messages SET active = 0, compacted = 1 WHERE session_id = ?",
|
|
[sid],
|
|
)
|
|
)
|
|
self._copy_tail_as_new_generation(db, sid, gen1)
|
|
newest_id = max(m["id"] for m in db.get_messages(sid, include_inactive=True))
|
|
msgs = db.get_messages(sid, include_compacted=True)
|
|
assert len(msgs) == 1
|
|
assert msgs[0]["id"] == newest_id # newest generation wins
|
|
|
|
def test_dedupe_applies_before_paging(self, db):
|
|
"""Deduping happens over the full display set, not per page, so
|
|
offset paging never surfaces a duplicate."""
|
|
sid = "s1"
|
|
db.create_session(sid, source="cli")
|
|
db.append_messages_batch(
|
|
sid,
|
|
[
|
|
{"role": "user", "content": "q1"},
|
|
{"role": "assistant", "content": "a1"},
|
|
{"role": "user", "content": "q2"},
|
|
{"role": "assistant", "content": "a2"},
|
|
],
|
|
)
|
|
orig = _row_ids(db, sid)
|
|
self._copy_tail_as_new_generation(db, sid, orig)
|
|
all_ids = _row_ids(db, sid, include_compacted=True)
|
|
assert len(all_ids) == 4 # deduped, no copies
|
|
# Paginate past where the copies would have landed.
|
|
page = db.get_messages(sid, include_compacted=True, limit=2, offset=2)
|
|
assert [m["id"] for m in page] == all_ids[2:]
|
|
assert len(page) == 2
|
|
|
|
def test_distinct_tool_calls_with_same_content_are_not_merged(self, db):
|
|
"""Two real tool messages that happen to share role/content/timestamp
|
|
must stay separate: the dedupe key includes the tool fields, so only
|
|
genuine compaction copies (which copy those fields verbatim) collapse.
|
|
"""
|
|
sid = "s1"
|
|
db.create_session(sid, source="cli")
|
|
|
|
def _seed_tool_rows(conn):
|
|
ts = 1700000000.0
|
|
for cid in ("call-1", "call-2"):
|
|
conn.execute(
|
|
"INSERT INTO messages (session_id, role, content, tool_call_id,"
|
|
" tool_name, timestamp, active, compacted)"
|
|
" VALUES (?, ?, ?, ?, ?, ?, 1, 0)",
|
|
(sid, "tool", "identical result", cid, "search", ts),
|
|
)
|
|
|
|
db._execute_write(_seed_tool_rows)
|
|
msgs = db.get_messages(sid, include_compacted=True)
|
|
assert len(msgs) == 2
|
|
assert {m["tool_call_id"] for m in msgs} == {"call-1", "call-2"}
|
|
|
|
def test_compaction_copies_of_tool_messages_still_collapse(self, db):
|
|
"""Tool rows copied by a compaction epoch (identical tool fields) are
|
|
deduped like any other message, not split by the widened key."""
|
|
sid = "s1"
|
|
db.create_session(sid, source="cli")
|
|
|
|
def _seed_tool_row(conn):
|
|
conn.execute(
|
|
"INSERT INTO messages (session_id, role, content, tool_call_id,"
|
|
" tool_name, timestamp, active, compacted)"
|
|
" VALUES (?, ?, ?, ?, ?, ?, 1, 0)",
|
|
(sid, "tool", "result", "call-1", "search", 1700000000.0),
|
|
)
|
|
|
|
db._execute_write(_seed_tool_row)
|
|
orig = _row_ids(db, sid)
|
|
self._copy_tail_as_new_generation(db, sid, orig)
|
|
msgs = db.get_messages(sid, include_compacted=True)
|
|
assert len(msgs) == 1
|
|
assert msgs[0]["tool_call_id"] == "call-1"
|