HERMES-GIELIE/tests/test_issue2223_compression_no_rename.py

"""
Tests for #2223: compression session rotation must not destroy session history.

The previous implementation renamed old_sid.json → new_sid.json during context
compression, destroying the only persistent copy of the uncompressed history
before the new session had been saved.  If the summariser also failed, the user
was left with zero recoverable messages.

The fix preserves old_sid.json and creates new_sid.json as a fresh file, setting
parent_session_id to link the lineage.
"""
import json
import pathlib
import textwrap
import threading

import pytest

STREAMING = pathlib.Path(__file__).resolve().parents[1] / "api" / "streaming.py"
streaming_src = STREAMING.read_text(encoding="utf-8")


# ── Structural checks ────────────────────────────────────────────────────────


class TestNoRenameDuringCompression:
    """The destructive old_path.rename(new_path) call must be removed."""

    def test_rename_call_removed(self):
        """old_path.rename(new_path) must not appear in the compression rotation block."""
        # The old code had: if old_path.exists() and not new_path.exists(): old_path.rename(new_path)
        # That line must be gone.
        assert "old_path.rename(new_path)" not in streaming_src, (
            "old_path.rename(new_path) still present — compression rotation "
            "still destroys session history (#2223)"
        )

    def test_parent_session_id_stamped_on_continuation(self):
        """The continuation session must carry parent_session_id linking to old_sid."""
        assert "s.parent_session_id = old_sid" in streaming_src, (
            "parent_session_id not stamped on continuation session (#2223)"
        )

    def test_old_session_preservation_logic_exists(self):
        """There must be logic to preserve the pre-compression session file."""
        assert "Preserved pre-compression session" in streaming_src, (
            "Pre-compression session preservation logging not found (#2223)"
        )


class TestMergePreservesHistory:
    """_merge_display_messages_after_agent_result must preserve all previous
    display messages when compression returns only a marker."""

    @pytest.fixture
    def merge(self):
        from api.streaming import _merge_display_messages_after_agent_result
        return _merge_display_messages_after_agent_result

    def test_marker_only_preserves_all_previous(self, merge):
        """When result is just a compression-failure marker, previous display survives."""
        previous_display = [
            {"role": "user", "content": f"msg{i}"} for i in range(100)
        ] + [
            {"role": "assistant", "content": f"reply{i}"} for i in range(100)
        ]
        previous_context = list(previous_display)
        marker = {
            "role": "user",
            "content": (
                "Summary generation was unavailable. 200 message(s) were removed "
                "to free context space but could not be summarized."
            ),
        }
        result = [marker, {"role": "user", "content": "continue"}, {"role": "assistant", "content": "ok"}]

        merged = merge(previous_display, previous_context, result, "continue")

        # All 200 original messages must survive.
        assert len(merged) >= 200
        for i in range(100):
            assert merged[i]["content"] == f"msg{i}"

    def test_empty_result_preserves_all_previous(self, merge):
        """If result_messages is empty, previous display is returned unchanged."""
        previous_display = [
            {"role": "user", "content": "hello"},
            {"role": "assistant", "content": "hi"},
        ]
        previous_context = list(previous_display)

        merged = merge(previous_display, previous_context, [], "test")

        assert merged == previous_display

    def test_none_result_preserves_all_previous(self, merge):
        """If result_messages is None, previous display is returned unchanged."""
        previous_display = [
            {"role": "user", "content": "hello"},
            {"role": "assistant", "content": "hi"},
        ]
        previous_context = list(previous_display)

        merged = merge(previous_display, previous_context, None, "test")

        assert merged == previous_display