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