nanobot/tests/agent/test_consolidator.py

"""Tests for the lightweight Consolidator — append-only to HISTORY.md."""

from dataclasses import replace
from unittest.mock import AsyncMock, MagicMock

import pytest

from nanobot.agent.memory import (
    _ARCHIVE_SUMMARY_MAX_CHARS,
    Consolidator,
    MemoryStore,
)
from nanobot.providers.base import (
    GenerationSettings,
    LLMResponse,
    ProviderConversationState,
    ToolCallRequest,
)
from nanobot.runtime_context import (
    RUNTIME_CONTEXT_HISTORY_META,
    RuntimeContextBlock,
    append_runtime_context,
)
from nanobot.session.keys import UNIFIED_SESSION_KEY, remember_last_channel
from nanobot.session.manager import Session
from nanobot.utils.llm_runtime import LLMRuntime
from nanobot.utils.prompt_templates import render_template


@pytest.fixture
def store(tmp_path):
    return MemoryStore(tmp_path)


@pytest.fixture
def mock_provider():
    p = MagicMock()
    p.chat_with_retry = AsyncMock()
    p.generation = GenerationSettings(max_tokens=100)
    return p


@pytest.fixture
def runtime(mock_provider):
    return LLMRuntime.capture(
        mock_provider,
        "test-model",
        context_window_tokens=1000,
    )


@pytest.fixture
def consolidator(store):
    sessions = MagicMock()
    sessions.save = MagicMock()
    # When maybe_consolidate_by_tokens refreshes the session reference via
    # get_or_create(session.key), it should get back the same object the test
    # passed in.  Store sessions by key so the lookup is transparent.
    _session_cache: dict[str, MagicMock] = {}
    sessions.get_or_create = MagicMock(side_effect=lambda key: _session_cache.get(key, MagicMock()))
    sessions._session_cache = _session_cache
    return Consolidator(
        store=store,
        sessions=sessions,
        build_messages=MagicMock(return_value=[]),
        get_tool_definitions=MagicMock(return_value=[]),
    )


def _tool_round(call_id: str) -> list[dict]:
    return [
        {
            "role": "assistant",
            "content": None,
            "tool_calls": [
                {"id": call_id, "type": "function", "function": {"name": "x", "arguments": "{}"}}
            ],
        },
        {"role": "tool", "tool_call_id": call_id, "name": "x", "content": "ok"},
    ]


def _provider_state() -> ProviderConversationState:
    return ProviderConversationState(
        kind="openai_responses",
        provider="openai:test",
        model="test-model",
        version=1,
        payload={"items": []},
    )


def _build_test_messages(**kwargs):
    return [
        {"role": "system", "content": "system prompt"},
        *kwargs["history"],
        {"role": "user", "content": kwargs["current_message"]},
    ]


async def _archive(consolidator, messages, runtime, *, session_key="test:session"):
    return await consolidator.archive(
        messages,
        runtime=runtime,
        session_key=session_key,
        request_messages=_build_test_messages(
            history=messages,
            current_message="consolidate",
        ),
        request_tools=[],
    )


class TestConsolidatorSummarize:
    def test_format_messages_keeps_media_only_user_turn(self):
        path = "/home/user/.nanobot/media/websocket/clip.mp4"

        formatted = MemoryStore._format_messages([
            {
                "role": "user",
                "content": "",
                "media": [path],
                "timestamp": "2026-07-27",
            }
        ])

        assert formatted == f"[2026-07-27] USER: [image: {path}]"

    async def test_archive_uses_captured_generation(
        self, consolidator, mock_provider, runtime
    ):
        admitted = replace(
            runtime,
            generation=GenerationSettings(
                temperature=0.25,
                max_tokens=321,
                reasoning_effort="medium",
            ),
        )
        mock_provider.generation = GenerationSettings(
            temperature=0.9,
            max_tokens=999,
            reasoning_effort="high",
        )
        mock_provider.chat_with_retry.return_value = MagicMock(
            content="Summary.",
            finish_reason="stop",
        )

        await _archive(consolidator, [{"role": "user", "content": "hello"}], admitted)

        call = mock_provider.chat_with_retry.call_args.kwargs
        assert call["model"] == admitted.model
        assert call["temperature"] == 0.25
        assert call["max_tokens"] == 321
        assert call["reasoning_effort"] == "medium"

    async def test_summarize_appends_to_history(
        self, consolidator, mock_provider, store, runtime
    ):
        """Consolidator should call LLM to summarize, then append to HISTORY.md."""
        mock_provider.chat_with_retry.return_value = MagicMock(
            content="User fixed a bug in the auth module."
        )
        messages = [
            {"role": "user", "content": "fix the auth bug"},
            {"role": "assistant", "content": "Done, fixed the race condition."},
        ]
        result = await _archive(consolidator, messages, runtime)
        assert result == "User fixed a bug in the auth module."
        entries = store.read_unprocessed_history(since_cursor=0)
        assert len(entries) == 1

    async def test_summarize_appends_session_key_to_history(
        self,
        consolidator,
        mock_provider,
        store,
        runtime,
    ):
        mock_provider.chat_with_retry.return_value = MagicMock(
            content="User fixed a bug in the auth module.",
            finish_reason="stop",
        )
        messages = [{"role": "user", "content": "fix the auth bug"}]

        await _archive(
            consolidator,
            messages,
            runtime,
            session_key="telegram:chat-1",
        )

        entries = store.read_unprocessed_history(since_cursor=0)
        assert entries[0]["session_key"] == "telegram:chat-1"

    async def test_summarize_raw_dumps_on_llm_failure(
        self, consolidator, mock_provider, store, runtime
    ):
        """On LLM failure, raw-dump messages to HISTORY.md."""
        mock_provider.chat_with_retry.side_effect = Exception("API error")
        messages = [{"role": "user", "content": "hello"}]
        result = await _archive(consolidator, messages, runtime)
        assert result is None  # no summary on raw dump fallback
        entries = store.read_unprocessed_history(since_cursor=0)
        assert len(entries) == 1
        assert "[RAW]" in entries[0]["content"]

    async def test_raw_dump_fallback_appends_session_key(
        self,
        consolidator,
        mock_provider,
        store,
        runtime,
    ):
        mock_provider.chat_with_retry.side_effect = Exception("API error")
        messages = [{"role": "user", "content": "hello"}]

        await _archive(
            consolidator,
            messages,
            runtime,
            session_key="slack:chat-2",
        )

        entries = store.read_unprocessed_history(since_cursor=0)
        assert entries[0]["session_key"] == "slack:chat-2"

    async def test_summarize_skips_empty_messages(self, consolidator, runtime):
        result = await _archive(consolidator, [], runtime)
        assert result is None


class TestConsolidatorPromptContract:
    def test_archive_prompt_outputs_attribute_tags_without_missing_context_claims(self):
        prompt = render_template("agent/consolidator_archive.md", strip=True, archive_count=4)

        assert "SNIP" in prompt
        assert "final 4 conversation messages" in prompt
        for mark in ("[permanent]", "[durable]", "[ephemeral]", "[correction]", "[skip]"):
            assert mark in prompt
        assert "check context below" not in prompt.lower()
        assert "Do not output facts already present in the system prompt's Recent History" in prompt
        assert "Do not mark something [skip] merely because it might already exist" in prompt

class TestConsolidatorArchiveErrorHandling:
    """archive() must fall back when the LLM does not complete its overview.

    Error responses include overloaded / quota failures from #3244; length
    responses contain a partial overview that is likewise unsafe to persist.
    """

    @pytest.mark.parametrize("finish_reason", ["error", "length"])
    async def test_archive_falls_back_on_incomplete_finish_reason(
        self,
        consolidator,
        mock_provider,
        store,
        runtime,
        finish_reason: str,
    ):
        """Incomplete LLM output should trigger raw_archive, not persist partial text."""
        invalid_output = f"INVALID_{finish_reason.upper()}_OUTPUT"
        mock_provider.chat_with_retry.return_value = MagicMock(
            content=invalid_output,
            finish_reason=finish_reason,
        )
        messages = [
            {"role": "user", "content": "fix the auth bug"},
            {"role": "assistant", "content": "Done, fixed the race condition."},
        ]
        result = await _archive(consolidator, messages, runtime)
        assert result is None
        entries = store.read_unprocessed_history(since_cursor=0)
        assert len(entries) == 1
        assert "[RAW]" in entries[0]["content"]
        assert invalid_output not in entries[0]["content"]

    async def test_archive_preserves_summary_on_success(
        self, consolidator, mock_provider, store, runtime
    ):
        """Normal LLM response should still produce a proper summary entry."""
        mock_provider.chat_with_retry.return_value = MagicMock(
            content="User fixed a bug in the auth module.",
            finish_reason="stop",
        )
        messages = [
            {"role": "user", "content": "fix the auth bug"},
            {"role": "assistant", "content": "Done."},
        ]
        result = await _archive(consolidator, messages, runtime)
        assert result == "User fixed a bug in the auth module."
        entries = store.read_unprocessed_history(since_cursor=0)
        assert len(entries) == 1
        assert "[RAW]" not in entries[0]["content"]

    async def test_archive_propagates_history_write_failure(
        self, consolidator, mock_provider, runtime
    ):
        mock_provider.chat_with_retry.return_value = MagicMock(
            content="Summary.",
            finish_reason="stop",
        )
        consolidator.store.append_history = MagicMock(side_effect=OSError("disk full"))
        consolidator.store.raw_archive = MagicMock()

        with pytest.raises(OSError, match="disk full"):
            await _archive(
                consolidator,
                [{"role": "user", "content": "important"}],
                runtime,
            )

        consolidator.store.raw_archive.assert_not_called()

    async def test_archive_propagates_template_failure_without_raw_archive(
        self, consolidator, mock_provider, runtime, monkeypatch
    ):
        runtime = replace(runtime, context_window_tokens=128_000)
        consolidator.store.raw_archive = MagicMock()
        monkeypatch.setattr(
            "nanobot.agent.memory.render_template",
            MagicMock(side_effect=RuntimeError("template failed")),
        )
        session = Session(key="test:template")
        session.add_message("user", "important")

        with pytest.raises(RuntimeError, match="template failed"):
            await consolidator.archive_session(
                session,
                archive_end=len(session.messages),
                runtime=runtime,
            )

        mock_provider.chat_with_retry.assert_not_awaited()
        consolidator.store.raw_archive.assert_not_called()


class TestConsolidatorTokenBudget:
    async def test_prompt_below_threshold_does_not_consolidate(
        self, consolidator, runtime
    ):
        """No consolidation when tokens are within budget."""
        session = MagicMock()
        session.last_consolidated = 0
        session.messages = [{"role": "user", "content": "hi"}]
        session.key = "test:key"
        consolidator.sessions._session_cache[session.key] = session
        consolidator.estimate_session_prompt_tokens = MagicMock(return_value=(100, "tiktoken"))
        consolidator.archive_session = AsyncMock(return_value=True)
        await consolidator.maybe_consolidate_by_tokens(session, runtime=runtime)
        consolidator.archive_session.assert_not_called()

    async def test_token_estimation_failure_propagates(self, consolidator, runtime):
        session = Session(key="test:estimate-failure")
        session.add_message("user", "hello")
        consolidator.sessions._session_cache[session.key] = session
        consolidator.estimate_session_prompt_tokens = MagicMock(
            side_effect=RuntimeError("counter failed")
        )

        with pytest.raises(RuntimeError, match="counter failed"):
            await consolidator.maybe_consolidate_by_tokens(session, runtime=runtime)

    async def test_estimate_uses_full_unconsolidated_tail(self, consolidator, runtime):
        """Consolidation pressure must account for the full unarchived tail."""
        session = Session(key="test:full-tail")
        for i in range(160):
            session.add_message("user", f"msg-{i}")

        captured: dict[str, list[dict]] = {}

        def build_messages(**kwargs):
            captured["history"] = kwargs["history"]
            return kwargs["history"]

        consolidator._build_messages = build_messages

        consolidator.estimate_session_prompt_tokens(session, runtime=runtime)

        assert len(captured["history"]) == 160
        assert captured["history"][0]["content"].endswith("msg-0")

    async def test_estimate_includes_recent_archived_replay(self, consolidator, runtime):
        session = Session(key="test:archived-replay")
        for i in range(10):
            session.add_message("user", f"msg-{i}")
        session.last_consolidated = len(session.messages)

        captured: dict[str, list[dict]] = {}

        def build_messages(**kwargs):
            captured["history"] = kwargs["history"]
            return kwargs["history"]

        consolidator._build_messages = build_messages

        consolidator.estimate_session_prompt_tokens(session, runtime=runtime)

        assert len(captured["history"]) == 8
        assert captured["history"][0]["content"] == "msg-2"

    async def test_token_overflow_appends_prompt_to_replay_prefix(
        self,
        consolidator,
        mock_provider,
        runtime,
    ):
        consolidator._SAFETY_BUFFER = 0
        session = Session(key="test:token-prefix")
        session.provider_state = _provider_state()
        session.messages = [
            {
                "role": "user" if i in {0, 50, 61} else "assistant",
                "content": f"m{i}",
            }
            for i in range(70)
        ]
        consolidator.sessions._session_cache[session.key] = session
        consolidator.estimate_session_prompt_tokens = MagicMock(
            side_effect=[(1200, "tiktoken"), (400, "tiktoken")]
        )
        consolidator.pick_consolidation_boundary = MagicMock(return_value=(50, 800))
        consolidator._build_messages = MagicMock(side_effect=_build_test_messages)
        mock_provider.estimate_prompt_tokens.return_value = (100, "test-counter")
        mock_provider.chat_with_retry.return_value = LLMResponse(
            content="Token overflow summary.",
            finish_reason="stop",
        )

        await consolidator.maybe_consolidate_by_tokens(session, runtime=runtime)

        request = mock_provider.chat_with_retry.await_args.kwargs
        assert [message["content"] for message in request["messages"][1:-1]] == [
            f"m{i}" for i in range(50)
        ]
        assert "final 50 conversation messages" in request["messages"][-1]["content"]
        assert request["tools"] == []
        assert request["tool_choice"] == "none"
        assert session.last_consolidated == 50
        assert session.provider_state is None

    async def test_raw_archive_fallback_advances_last_consolidated(
        self, consolidator, runtime
    ):
        """When archive() falls back to raw-archive (LLM failed), the cursor
        must still advance. Otherwise the same chunk gets raw-archived again
        on every subsequent maybe_consolidate_by_tokens() call, spamming
        duplicate [RAW] entries into history.jsonl."""
        consolidator._SAFETY_BUFFER = 0
        session = MagicMock()
        session.last_consolidated = 0
        session.key = "test:key"
        session.messages = [
            {"role": "user" if i in {0, 50} else "assistant", "content": f"m{i}"}
            for i in range(70)
        ]
        session.metadata = {}
        consolidator.sessions._session_cache[session.key] = session
        consolidator.estimate_session_prompt_tokens = MagicMock(
            side_effect=[(1200, "tiktoken"), (400, "tiktoken")]
        )
        # LLM consolidation fails after raw_archive fires.
        consolidator.archive_session = AsyncMock(return_value=None)

        await consolidator.maybe_consolidate_by_tokens(session, runtime=runtime)

        consolidator.archive_session.assert_awaited_once()
        # The chunk is considered "materialized" (as a raw-archive breadcrumb),
        # so last_consolidated must have moved past it.
        assert session.last_consolidated == 50

    async def test_raw_archive_fallback_breaks_round_loop(
        self, consolidator, runtime
    ):
        """A degraded LLM should not trigger more archive() calls within the
        same maybe_consolidate_by_tokens invocation — bail after one fallback."""
        consolidator._SAFETY_BUFFER = 0
        session = MagicMock()
        session.last_consolidated = 0
        session.key = "test:key"
        session.messages = [
            {"role": "user" if i in {0, 20, 40, 60} else "assistant", "content": f"m{i}"}
            for i in range(70)
        ]
        session.metadata = {}
        consolidator.sessions._session_cache[session.key] = session
        # Keep estimates high so the loop would otherwise run multiple rounds.
        consolidator.estimate_session_prompt_tokens = MagicMock(
            return_value=(1200, "tiktoken")
        )
        consolidator.archive_session = AsyncMock(return_value=None)

        await consolidator.maybe_consolidate_by_tokens(session, runtime=runtime)

        # Exactly one fallback per call — not _MAX_CONSOLIDATION_ROUNDS.
        assert consolidator.archive_session.await_count == 1

    async def test_boundary_respected_when_no_intermediate_user_turn(
        self, consolidator, runtime
    ):
        """When boundary points past a long tool chain, the full chunk is archived."""
        consolidator._SAFETY_BUFFER = 0
        session = MagicMock()
        session.last_consolidated = 0
        session.key = "test:key"
        session.messages = [
            {
                "role": "user" if i in {0, 61} else "assistant",
                "content": f"m{i}",
            }
            for i in range(70)
        ]
        consolidator.sessions._session_cache[session.key] = session
        consolidator.estimate_session_prompt_tokens = MagicMock(
            side_effect=[(1200, "tiktoken"), (400, "tiktoken")]
        )
        consolidator.archive_session = AsyncMock(return_value=True)

        await consolidator.maybe_consolidate_by_tokens(session, runtime=runtime)

        consolidator.archive_session.assert_awaited_once()
        # pick_consolidation_boundary finds the only boundary at idx=61
        assert session.last_consolidated == 61


class TestCompactIdleSession:
    """Idle compaction tests."""

    @pytest.fixture
    def runtime(self, mock_provider):
        """Exercise the structured idle-consolidation path by default."""
        return LLMRuntime.capture(
            mock_provider,
            "test-model",
            context_window_tokens=128_000,
        )

    @pytest.fixture
    def real_consolidator(self, store, mock_provider):
        """Create a Consolidator with a real SessionManager (not a mock)."""
        from nanobot.session.manager import SessionManager

        sessions = SessionManager(store.workspace)
        return Consolidator(
            store=store,
            sessions=sessions,
            build_messages=MagicMock(side_effect=_build_test_messages),
            get_tool_definitions=MagicMock(return_value=[]),
        )

    @pytest.mark.asyncio
    async def test_archives_full_tail_preserves_messages_and_replays_recent_suffix(
        self, real_consolidator, mock_provider, runtime
    ):
        mock_provider.chat_with_retry.return_value = MagicMock(
            content="Summary of old conversation.", finish_reason="stop"
        )
        sessions = real_consolidator.sessions
        session = sessions.get_or_create("cli:test")
        session.provider_state = _provider_state()
        old_ts = session.updated_at
        for i in range(20):
            session.add_message("user", f"user msg {i}")
            session.add_message("assistant", f"assistant msg {i}")
        session.updated_at = old_ts
        sessions.save(session)

        result = await real_consolidator.compact_idle_session(
            "cli:test", runtime=runtime, max_suffix=8
        )
        assert result == "Summary of old conversation."

        sessions.invalidate("cli:test")
        reloaded = sessions.get_or_create("cli:test")
        assert len(reloaded.messages) == 40
        assert reloaded.messages[0]["content"] == "user msg 0"
        assert reloaded.last_consolidated == 40
        assert reloaded.provider_state is None
        visible = reloaded.get_history(max_messages=40)
        assert len(visible) == 8
        assert visible[0]["content"] == "user msg 16"
        assert visible[-1]["content"] == "assistant msg 19"
        meta = reloaded.metadata.get("_last_summary")
        assert meta is not None
        assert meta["text"] == "Summary of old conversation."
        assert "last_active" in meta
        assert reloaded.updated_at == old_ts

    @pytest.mark.asyncio
    async def test_short_idle_session_archives_once(
        self, real_consolidator, mock_provider, store, runtime
    ):
        mock_provider.chat_with_retry.return_value = MagicMock(
            content="Short summary.", finish_reason="stop"
        )
        sessions = real_consolidator.sessions
        session = sessions.get_or_create("cli:short")
        session.add_message("user", "hello")
        session.add_message("assistant", "hi")
        sessions.save(session)

        first = await real_consolidator.compact_idle_session("cli:short", runtime=runtime)
        second = await real_consolidator.compact_idle_session("cli:short", runtime=runtime)

        assert first == "Short summary."
        assert second == ""
        mock_provider.chat_with_retry.assert_awaited_once()
        assert len(store.read_unprocessed_history(since_cursor=0)) == 1
        reloaded = sessions.get_or_create("cli:short")
        assert reloaded.last_consolidated == 2
        assert [message["content"] for message in reloaded.get_history()] == ["hello", "hi"]

    @pytest.mark.asyncio
    async def test_new_messages_advance_existing_archive_progress(
        self, real_consolidator, mock_provider, runtime
    ):
        mock_provider.chat_with_retry.return_value = MagicMock(
            content="Summary.", finish_reason="stop"
        )
        sessions = real_consolidator.sessions
        session = sessions.get_or_create("cli:incremental")
        session.add_message("user", "first user")
        session.add_message("assistant", "first assistant")
        sessions.save(session)

        await real_consolidator.compact_idle_session("cli:incremental", runtime=runtime)
        current = sessions.get_or_create("cli:incremental")
        current.add_message("user", "second user")
        current.add_message("assistant", "second assistant")
        sessions.save(current)
        await real_consolidator.compact_idle_session("cli:incremental", runtime=runtime)

        assert mock_provider.chat_with_retry.await_count == 2
        latest_messages = mock_provider.chat_with_retry.await_args_list[-1].kwargs["messages"]
        assert [message["content"] for message in latest_messages[1:5]] == [
            "first user",
            "first assistant",
            "second user",
            "second assistant",
        ]
        assert "final 2 conversation messages" in latest_messages[-1]["content"]
        assert sessions.get_or_create("cli:incremental").last_consolidated == 4

    @pytest.mark.asyncio
    async def test_concurrent_append_remains_unarchived(
        self, real_consolidator, mock_provider, runtime
    ):
        sessions = real_consolidator.sessions
        session = sessions.get_or_create("cli:concurrent")
        session.add_message("user", "captured user")
        session.add_message("assistant", "captured assistant")
        sessions.save(session)

        async def append_during_archive(**_kwargs):
            current = sessions.get_or_create("cli:concurrent")
            current.add_message("user", "late user")
            current.add_message("assistant", "late assistant")
            return LLMResponse(content="Summary.", finish_reason="stop")

        mock_provider.chat_with_retry.side_effect = append_during_archive

        await real_consolidator.compact_idle_session("cli:concurrent", runtime=runtime)

        reloaded = sessions.get_or_create("cli:concurrent")
        assert len(reloaded.messages) == 4
        assert reloaded.last_consolidated == 2

    @pytest.mark.asyncio
    async def test_summarizes_retained_suffix_not_just_dropped_prefix(
        self, real_consolidator, mock_provider, runtime
    ):
        """idleCompact must summarize over the full unconsolidated tail, including
        the recent suffix it retains. Otherwise a late user correction / final
        result that lands in the kept suffix is excluded from the persisted
        summary, leaving a stale wrong conclusion in history. Regression for #4264."""
        mock_provider.chat_with_retry.return_value = MagicMock(
            content="Summary.", finish_reason="stop"
        )
        sessions = real_consolidator.sessions
        session = sessions.get_or_create("cli:correction")
        for i in range(18):
            session.add_message("user", f"user msg {i}")
            session.add_message("assistant", f"assistant msg {i}")
        # Final correction exchange lands inside the retained max_suffix window.
        session.add_message("user", "no, that's wrong, use approach B")
        session.add_message("assistant", "CORRECTED_FINAL_RESULT_alpha")
        sessions.save(session)

        await real_consolidator.compact_idle_session(
            "cli:correction", runtime=runtime, max_suffix=8
        )

        sent_messages = mock_provider.chat_with_retry.call_args.kwargs["messages"]
        assert any(
            message.get("content") == "CORRECTED_FINAL_RESULT_alpha"
            for message in sent_messages
        )

    @pytest.mark.asyncio
    async def test_raw_dumps_full_archive_batch_on_llm_failure(
        self, real_consolidator, mock_provider, store, runtime
    ):
        """The fallback covers the same full range as successful idle archival."""
        mock_provider.chat_with_retry.side_effect = RuntimeError("LLM unavailable")
        sessions = real_consolidator.sessions
        session = sessions.get_or_create("cli:rawdrop")
        for i in range(18):
            session.add_message("user", f"user msg {i}")
            session.add_message("assistant", f"assistant msg {i}")
        session.add_message("user", "final user follow-up")
        session.add_message("assistant", "RETAINED_SUFFIX_marker")
        sessions.save(session)

        await real_consolidator.compact_idle_session(
            "cli:rawdrop", runtime=runtime, max_suffix=8
        )

        raw = "\n".join(e["content"] for e in store.read_unprocessed_history(since_cursor=0))
        assert "[RAW]" in raw
        assert "user msg 0" in raw
        assert "RETAINED_SUFFIX_marker" in raw
        reloaded = sessions.get_or_create("cli:rawdrop")
        assert len(reloaded.messages) == 38
        assert reloaded.messages[-1]["content"] == "RETAINED_SUFFIX_marker"

    @pytest.mark.asyncio
    async def test_idle_compact_writes_session_key_to_history(
        self,
        real_consolidator,
        mock_provider,
        store,
        runtime,
    ):
        mock_provider.chat_with_retry.return_value = MagicMock(
            content="Summary of old conversation.", finish_reason="stop"
        )
        session = real_consolidator.sessions.get_or_create("cli:test")
        for i in range(10):
            session.add_message("user", f"user msg {i}")
            session.add_message("assistant", f"assistant msg {i}")
        real_consolidator.sessions.save(session)

        await real_consolidator.compact_idle_session(
            "cli:test", runtime=runtime, max_suffix=4
        )

        entries = store.read_unprocessed_history(since_cursor=0)
        assert entries[0]["session_key"] == "cli:test"

    @pytest.mark.asyncio
    async def test_empty_session_does_not_refresh_timestamp(
        self, real_consolidator, runtime
    ):
        """Empty session with old updated_at does not look active after compaction."""
        from datetime import datetime, timedelta

        sessions = real_consolidator.sessions
        session = sessions.get_or_create("cli:empty")
        old_ts = datetime.now() - timedelta(hours=2)
        session.updated_at = old_ts
        sessions.save(session)

        result = await real_consolidator.compact_idle_session(
            "cli:empty", runtime=runtime
        )
        assert result == ""

        reloaded = sessions.get_or_create("cli:empty")
        assert reloaded.updated_at == old_ts
        assert reloaded.metadata == {}

    @pytest.mark.asyncio
    async def test_nothing_summary_not_stored(
        self, real_consolidator, mock_provider, runtime
    ):
        """LLM returns '(nothing)' → neither history nor metadata stores it."""
        mock_provider.chat_with_retry.return_value = MagicMock(
            content="(nothing)", finish_reason="stop"
        )
        sessions = real_consolidator.sessions
        session = sessions.get_or_create("cli:nothing")
        for i in range(10):
            session.add_message("user", f"u{i}")
            session.add_message("assistant", f"a{i}")
        sessions.save(session)

        result = await real_consolidator.compact_idle_session(
            "cli:nothing", runtime=runtime, max_suffix=4
        )
        assert result == "(nothing)"

        reloaded = sessions.get_or_create("cli:nothing")
        assert "_last_summary" not in reloaded.metadata
        assert real_consolidator.store.read_unprocessed_history(0) == []

    @pytest.mark.asyncio
    async def test_llm_failure_preserves_history_but_advances_replay_boundary(
        self, real_consolidator, mock_provider, store, runtime
    ):
        mock_provider.chat_with_retry.side_effect = RuntimeError("LLM unavailable")
        sessions = real_consolidator.sessions
        session = sessions.get_or_create("cli:fail")
        for i in range(10):
            session.add_message("user", f"u{i}")
            session.add_message("assistant", f"a{i}")
        sessions.save(session)

        result = await real_consolidator.compact_idle_session(
            "cli:fail", runtime=runtime, max_suffix=4
        )
        assert result is None

        # raw_archive should have been called (history.jsonl gets an entry)
        entries = store.read_unprocessed_history(since_cursor=0)
        assert any("[RAW]" in e["content"] for e in entries)

        reloaded = sessions.get_or_create("cli:fail")
        assert len(reloaded.messages) == 20
        assert reloaded.messages[0]["content"] == "u0"
        assert reloaded.last_consolidated == 20
        assert [m["content"] for m in reloaded.get_history(max_messages=20)] == [
            "u6",
            "a6",
            "u7",
            "a7",
            "u8",
            "a8",
            "u9",
            "a9",
        ]

    @pytest.mark.asyncio
    async def test_respects_last_consolidated(
        self, real_consolidator, mock_provider, runtime
    ):
        """30 turns with last_consolidated=50 → only unconsolidated tail considered."""
        mock_provider.chat_with_retry.return_value = MagicMock(
            content="Tail summary.", finish_reason="stop"
        )
        sessions = real_consolidator.sessions
        session = sessions.get_or_create("cli:offset")
        for i in range(30):
            session.add_message("user", f"u{i}")
            session.add_message("assistant", f"a{i}")
        session.last_consolidated = 50  # Only 10 messages unconsolidated
        sessions.save(session)

        result = await real_consolidator.compact_idle_session(
            "cli:offset", runtime=runtime, max_suffix=4
        )
        assert result == "Tail summary."
        reloaded = sessions.get_or_create("cli:offset")
        assert len(reloaded.messages) == 60
        assert reloaded.last_consolidated == 60

        # Verify only the unconsolidated tail was processed:
        # All 10 unconsolidated messages (50-59) are archived exactly once.
        archived_call = mock_provider.chat_with_retry.call_args
        sent_messages = archived_call.kwargs["messages"]
        sent_content = [message.get("content") for message in sent_messages]
        # The ordinary replay prefix contributes recent context, while the
        # temporary instruction limits the new overview to the unarchived tail.
        assert "u0" not in sent_content
        assert "u26" in sent_content
        assert "final 10 conversation messages" in sent_messages[-1]["content"]

    @pytest.mark.asyncio
    async def test_full_archive_keeps_extended_legal_replay_suffix(
        self,
        real_consolidator,
        mock_provider,
        runtime,
    ):
        mock_provider.chat_with_retry.return_value = MagicMock(
            content="Tail summary.", finish_reason="stop"
        )
        sessions = real_consolidator.sessions
        session = sessions.get_or_create("cli:noncontiguous")
        for i in range(15):
            session.add_message("user", f"user-{i:02d}")
        for i in range(10):
            session.add_message("assistant", f"assistant-{i:02d}")
        sessions.save(session)

        result = await real_consolidator.compact_idle_session(
            "cli:noncontiguous", runtime=runtime, max_suffix=6
        )
        assert result == "Tail summary."

        reloaded = sessions.get_or_create("cli:noncontiguous")
        assert len(reloaded.messages) == 25
        assert reloaded.last_consolidated == 25
        assert [m["content"] for m in reloaded.get_history(max_messages=25)] == [
            "user-14",
            "assistant-00",
            "assistant-01",
            "assistant-02",
            "assistant-03",
            "assistant-04",
            "assistant-05",
            "assistant-06",
            "assistant-07",
            "assistant-08",
            "assistant-09",
        ]

        # #4264: idle compaction now summarizes the full unconsolidated tail, so
        # the dropped head (user-00) and retained suffix (user-14 through
        # assistant-09) are all summarized.
        archived_call = mock_provider.chat_with_retry.call_args
        sent_content = [message.get("content") for message in archived_call.kwargs["messages"]]
        assert "user-00" in sent_content
        assert "assistant-09" in sent_content
        assert "user-14" in sent_content

    @pytest.mark.asyncio
    async def test_preserves_tool_history_and_persists_only_overview(
        self,
        real_consolidator,
        mock_provider,
        store,
        runtime,
    ):
        tools = [{"type": "function", "function": {"name": "lookup"}}]
        real_consolidator._get_tool_definitions.return_value = tools
        mock_provider.chat_with_retry.return_value = LLMResponse(
            content="Overview from the temporary turn.",
            finish_reason="stop",
        )
        sessions = real_consolidator.sessions
        session = sessions.get_or_create("cli:tool-history")
        session.add_message("user", "look this up")
        session.messages.extend(_tool_round("call-1"))
        session.add_message("assistant", "final answer")
        sessions.save(session)

        result = await real_consolidator.compact_idle_session(
            "cli:tool-history",
            runtime=runtime,
        )

        assert result == "Overview from the temporary turn."
        call = mock_provider.chat_with_retry.call_args.kwargs
        sent_messages = call["messages"]
        assert [message["role"] for message in sent_messages] == [
            "system",
            "user",
            "assistant",
            "tool",
            "assistant",
            "user",
        ]
        assert sent_messages[2]["tool_calls"][0]["id"] == "call-1"
        assert "final 4 conversation messages" in sent_messages[-1]["content"]
        assert call["tools"] == tools
        assert call["tool_choice"] == "none"

        reloaded = sessions.get_or_create("cli:tool-history")
        assert len(reloaded.messages) == 4
        assert reloaded.messages[-1]["content"] == "final answer"
        assert all(
            "memory overview" not in str(message.get("content", "")).lower()
            for message in reloaded.messages
        )
        entries = store.read_unprocessed_history(since_cursor=0)
        assert [entry["content"] for entry in entries] == [
            "Overview from the temporary turn."
        ]

    @pytest.mark.asyncio
    async def test_tool_call_response_uses_raw_fallback(
        self,
        real_consolidator,
        mock_provider,
        store,
        runtime,
    ):
        mock_provider.chat_with_retry.return_value = LLMResponse(
            content=None,
            tool_calls=[ToolCallRequest(id="call-1", name="lookup", arguments={})],
            finish_reason="tool_calls",
        )
        sessions = real_consolidator.sessions
        session = sessions.get_or_create("cli:unexpected-tool")
        session.add_message("user", "remember this")
        session.add_message("assistant", "important answer")
        sessions.save(session)

        result = await real_consolidator.compact_idle_session(
            "cli:unexpected-tool",
            runtime=runtime,
        )

        assert result is None
        entries = store.read_unprocessed_history(since_cursor=0)
        assert len(entries) == 1
        assert entries[0]["content"].startswith("[RAW] ")
        assert "important answer" in entries[0]["content"]
        assert sessions.get_or_create("cli:unexpected-tool").last_consolidated == 2

    @pytest.mark.asyncio
    async def test_empty_response_uses_raw_fallback(
        self,
        real_consolidator,
        mock_provider,
        store,
        runtime,
    ):
        mock_provider.chat_with_retry.return_value = LLMResponse(
            content="",
            finish_reason="stop",
        )
        sessions = real_consolidator.sessions
        session = sessions.get_or_create("cli:empty-summary")
        session.add_message("user", "remember this")
        session.add_message("assistant", "important answer")
        sessions.save(session)

        result = await real_consolidator.compact_idle_session(
            "cli:empty-summary",
            runtime=runtime,
        )

        assert result is None
        entries = store.read_unprocessed_history(since_cursor=0)
        assert len(entries) == 1
        assert entries[0]["content"].startswith("[RAW] ")
        assert "important answer" in entries[0]["content"]
        assert sessions.get_or_create("cli:empty-summary").last_consolidated == 2

    @pytest.mark.asyncio
    async def test_oversized_prefix_raw_archives_without_flattened_llm_retry(
        self,
        real_consolidator,
        mock_provider,
        store,
        runtime,
    ):
        runtime = replace(runtime, context_window_tokens=1_000)
        sessions = real_consolidator.sessions
        session = sessions.get_or_create("sdk:oversized")
        session.add_message("user", "x" * 100_000)
        sessions.save(session)

        result = await real_consolidator.compact_idle_session(
            "sdk:oversized",
            runtime=runtime,
        )

        assert result is None
        mock_provider.chat_with_retry.assert_not_awaited()
        entries = store.read_unprocessed_history(since_cursor=0)
        assert len(entries) == 1
        assert entries[0]["content"].startswith("[RAW] ")
        assert sessions.get_or_create("sdk:oversized").last_consolidated == 1

    @pytest.mark.asyncio
    async def test_incremental_scope_counts_only_model_visible_messages(
        self,
        real_consolidator,
        mock_provider,
        runtime,
    ):
        mock_provider.chat_with_retry.return_value = LLMResponse(
            content="Summary.",
            finish_reason="stop",
        )
        sessions = real_consolidator.sessions
        session = sessions.get_or_create("cli:commands")
        session.add_message("user", "already archived user")
        session.add_message("assistant", "already archived answer")
        session.last_consolidated = 2
        session.add_message("user", "/status", _command=True)
        session.add_message("assistant", "status output", _command=True)
        session.add_message("user", "new user")
        session.add_message("assistant", "new answer")
        sessions.save(session)

        await real_consolidator.compact_idle_session(
            "cli:commands",
            runtime=runtime,
        )

        sent = mock_provider.chat_with_retry.call_args.kwargs["messages"]
        assert [message.get("content") for message in sent[1:-1]] == [
            "already archived user",
            "already archived answer",
            "new user",
            "new answer",
        ]
        assert "final 2 conversation messages" in sent[-1]["content"]

    @pytest.mark.asyncio
    async def test_reuses_real_prefix_for_unified_session_workspace(
        self,
        loop_factory,
        mock_provider,
        tmp_path,
    ):
        project = tmp_path / "project"
        project.mkdir()
        (tmp_path / "AGENTS.md").write_text("GLOBAL_WORKSPACE_MARKER", encoding="utf-8")
        (project / "AGENTS.md").write_text("PROJECT_WORKSPACE_MARKER", encoding="utf-8")
        loop = loop_factory(provider=mock_provider, unified_session=True)
        runtime = loop.llm_runtime()
        runtime.provider.chat_with_retry.return_value = LLMResponse(
            content="Summary.",
            finish_reason="stop",
        )
        session = loop.sessions.get_or_create(UNIFIED_SESSION_KEY)
        remember_last_channel(session.metadata, "websocket", "scope")
        session.metadata["workspace_scope"] = {
            "project_path": str(project),
            "access_mode": "restricted",
        }
        session.add_message("user", "project question")
        session.add_message("assistant", "project answer")
        loop.sessions.save(session)
        ordinary_messages = loop.context.build_messages(
            history=session.get_history(max_messages=0),
            current_message="next project question",
            channel="websocket",
            workspace=project,
            session_key=session.key,
            unified_session=True,
        )

        await loop.consolidator.compact_idle_session(
            session.key,
            runtime=runtime,
        )

        sent_messages = runtime.provider.chat_with_retry.call_args.kwargs["messages"]
        assert sent_messages[:-1] == ordinary_messages[:-1]
        assert "final 2 conversation messages" in sent_messages[-1]["content"]
        system = sent_messages[0]["content"]
        assert "PROJECT_WORKSPACE_MARKER" in system
        assert "GLOBAL_WORKSPACE_MARKER" not in system

    @pytest.mark.asyncio
    async def test_acquires_consolidation_lock(
        self, real_consolidator, mock_provider, runtime
    ):
        """Verify lock is held during execution."""
        import asyncio

        # Use a slow LLM response to ensure the lock is held while we check
        started = asyncio.Event()
        release_chat = asyncio.Event()

        async def slow_chat(**kwargs):
            started.set()
            await release_chat.wait()
            return LLMResponse(content="Summary.", finish_reason="stop")

        mock_provider.chat_with_retry = slow_chat

        sessions = real_consolidator.sessions
        session = sessions.get_or_create("cli:lock")
        for i in range(10):
            session.add_message("user", f"u{i}")
            session.add_message("assistant", f"a{i}")
        sessions.save(session)

        lock = real_consolidator.get_lock("cli:lock")
        assert not lock.locked()

        task = asyncio.ensure_future(
            real_consolidator.compact_idle_session(
                "cli:lock", runtime=runtime, max_suffix=4
            )
        )
        await started.wait()
        assert lock.locked()
        release_chat.set()
        await task
        assert not lock.locked()


class TestConsolidatorSessionRefresh:
    """Background consolidation must detect stale session references."""

    @pytest.mark.asyncio
    async def test_reloads_before_empty_session_guard(self, tmp_path):
        """A stale empty reference must not skip a non-empty cached session."""
        from nanobot.agent.memory import Consolidator, MemoryStore
        from nanobot.session.manager import Session, SessionManager

        store = MemoryStore(tmp_path)
        provider = MagicMock()
        provider.chat_with_retry = AsyncMock(
            return_value=MagicMock(content="summary", finish_reason="stop")
        )
        provider.generation = GenerationSettings(max_tokens=4096)
        provider.estimate_prompt_tokens = MagicMock(return_value=(10, "test"))
        runtime = LLMRuntime.capture(
            provider,
            "test-model",
            context_window_tokens=128_000,
        )
        sessions = SessionManager(tmp_path)
        consolidator = Consolidator(
            store=store,
            sessions=sessions,
            build_messages=MagicMock(return_value=[]),
            get_tool_definitions=MagicMock(return_value=[]),
        )

        fresh = sessions.get_or_create("cli:test")
        fresh.add_message("user", "fresh message")
        sessions.save(fresh)
        stale_empty = Session(key="cli:test")

        seen: dict[str, Session] = {}

        def estimate(session: Session, *, runtime):
            seen["session"] = session
            return 10, "test"

        consolidator.estimate_session_prompt_tokens = MagicMock(side_effect=estimate)

        await consolidator.maybe_consolidate_by_tokens(
            stale_empty,
            runtime=runtime,
        )

        assert seen["session"] is fresh

    @pytest.mark.asyncio
    async def test_reloads_stale_session_after_compact(self, tmp_path):
        """After compact_idle_session replaces the session, a concurrent
        maybe_consolidate_by_tokens with the old reference should use the
        fresh session from cache instead of overwriting."""
        from nanobot.agent.memory import Consolidator, MemoryStore
        from nanobot.session.manager import SessionManager

        store = MemoryStore(tmp_path)
        provider = MagicMock()
        provider.chat_with_retry = AsyncMock(
            return_value=MagicMock(content="summary", finish_reason="stop")
        )
        provider.generation = GenerationSettings(max_tokens=4096)
        provider.estimate_prompt_tokens = MagicMock(return_value=(10, "test"))
        runtime = LLMRuntime.capture(
            provider,
            "test-model",
            context_window_tokens=128_000,
        )
        sessions = SessionManager(tmp_path)
        consolidator = Consolidator(
            store=store,
            sessions=sessions,
            build_messages=MagicMock(return_value=[]),
            get_tool_definitions=MagicMock(return_value=[]),
        )

        # Populate session with many messages
        session = sessions.get_or_create("cli:test")
        for i in range(20):
            session.add_message("user", f"u{i}")
            session.add_message("assistant", f"a{i}")
        sessions.save(session)

        # Simulate: background consolidation captures old reference
        old_ref = session

        await consolidator.compact_idle_session(
            "cli:test",
            runtime=runtime,
            max_suffix=8,
        )

        await consolidator.maybe_consolidate_by_tokens(
            old_ref,
            runtime=runtime,
        )

        session_after = sessions.get_or_create("cli:test")
        assert len(session_after.messages) == 40
        assert session_after.last_consolidated == 40
        assert len(session_after.get_history(max_messages=40)) == 8


class TestRawArchiveTruncation:
    """raw_archive() must cap entry size to avoid bloating history.jsonl."""

    def test_raw_archive_truncates_large_content(self, store):
        """Large messages should be truncated to _RAW_ARCHIVE_MAX_CHARS."""
        big = "x" * 50_000
        messages = [{"role": "user", "content": big}]
        store.raw_archive(messages)
        entries = store.read_unprocessed_history(since_cursor=0)
        assert len(entries) == 1
        assert len(entries[0]["content"]) < 50_000
        assert "[RAW]" in entries[0]["content"]

    def test_raw_archive_preserves_small_content(self, store):
        """Small messages should not be truncated."""
        messages = [{"role": "user", "content": "hello"}]
        store.raw_archive(messages)
        entries = store.read_unprocessed_history(since_cursor=0)
        assert len(entries) == 1
        assert "hello" in entries[0]["content"]

    def test_raw_archive_excludes_model_only_runtime_context(self, store):
        content, marker = append_runtime_context(
            "ship the feature",
            [RuntimeContextBlock(source="goal", content="host-only goal guidance")],
        )

        store.raw_archive([{
            "role": "user",
            "content": content,
            RUNTIME_CONTEXT_HISTORY_META: marker,
        }])

        entry = store.read_unprocessed_history(since_cursor=0)[0]["content"]
        assert "ship the feature" in entry
        assert "host-only goal guidance" not in entry

    def test_raw_archive_preserves_session_key(self, store):
        messages = [{"role": "user", "content": "hello"}]
        store.raw_archive(messages, session_key="websocket:chat-1")
        entries = store.read_unprocessed_history(since_cursor=0)
        assert entries[0]["session_key"] == "websocket:chat-1"

    def test_raw_archive_custom_max_chars(self, store):
        """max_chars parameter should override default limit."""
        messages = [{"role": "user", "content": "a" * 200}]
        store.raw_archive(messages, max_chars=100)
        entries = store.read_unprocessed_history(since_cursor=0)
        assert len(entries[0]["content"]) < 200


class TestArchivePersistence:
    async def test_oversized_summary_is_capped_before_append(
        self, consolidator, mock_provider, store, runtime
    ):
        """A pathologically large LLM summary must not land full-length in
        history.jsonl — that would re-open the #3412 bloat vector from the
        *success* path instead of the fallback path."""
        mock_provider.chat_with_retry.return_value = MagicMock(
            content="S" * (_ARCHIVE_SUMMARY_MAX_CHARS * 10),
            finish_reason="stop",
        )
        await _archive(
            consolidator,
            [{"role": "user", "content": "hi"}],
            runtime,
        )

        entry = store.read_unprocessed_history(since_cursor=0)[0]
        assert len(entry["content"]) <= _ARCHIVE_SUMMARY_MAX_CHARS + 50