"""Tests for context window management: estimate_tokens, group_into_turns,
compact_messages, drop_middle_turns, clamp_output_tokens, and ContextOverflowError."""

import copy
from types import SimpleNamespace
from unittest.mock import patch

import pytest

from swival.agent import (
    estimate_tokens,
    group_into_turns,
    compact_messages,
    compact_tool_result,
    is_pinned,
    score_turn,
    drop_middle_turns,
    aggressive_drop_turns,
    compact_context,
    compact_to_budget,
    CompactionContext,
    COMPACTION_AGGRESSIVE,
    COMPACTION_COMPACT_MESSAGES,
    COMPACTION_DROP_MIDDLE,
    COMPACTION_DROP_TOOLS,
    COMPACTION_GC_SCAFFOLDING,
    COMPACTION_STRIP_REASONING,
    _gc_scaffolding,
    _has_spent_scaffolding,
    _prompt_budget,
    GOAL_RECAP_PREFIX,
    MIN_OUTPUT_TOKENS,
    _emergency_truncate,
    summarize_turns,
    _RECAP_PREFIX,
    CompactionState,
    MAX_CHECKPOINTS,
    MAX_CHECKPOINT_TOKENS,
    clamp_output_tokens,
    ContextOverflowError,
    call_llm,
    _fix_orphaned_tool_calls,
    _run_terminal_floor_ladder,
    TerminalAttemptResult,
    _TERMINAL_FLOOR_BUDGETS,
    _looks_like_context_overflow,
    _parse_context_limit,
    _terminal_floor_eligible,
    AdaptiveContextBudget,
    _CONTEXT_EXHAUSTED_FALLBACK,
    _CONTEXT_EXHAUSTED_REASON,
    REACTIVE_BUDGET_BACKOFF,
)
from swival.report import AgentError


# ---------------------------------------------------------------------------
# Helpers to build messages
# ---------------------------------------------------------------------------


def _sys(content):
    return {"role": "system", "content": content}


def _user(content):
    return {"role": "user", "content": content}


def _assistant(content):
    return {"role": "assistant", "content": content}


def _assistant_tc(tool_calls):
    """Assistant message with tool_calls (list of (id, name, args_json))."""
    tcs = [
        SimpleNamespace(id=tc_id, function=SimpleNamespace(name=name, arguments=args))
        for tc_id, name, args in tool_calls
    ]
    return SimpleNamespace(role="assistant", content=None, tool_calls=tcs)


def _tool(tc_id, content):
    return {"role": "tool", "tool_call_id": tc_id, "content": content}


# ---------------------------------------------------------------------------
# estimate_tokens
# ---------------------------------------------------------------------------


class TestEstimateTokens:
    def test_basic(self):
        msgs = [_user("hello world")]
        count = estimate_tokens(msgs)
        assert count > 0

    def test_includes_tool_calls(self):
        msgs_plain = [_assistant("hello")]
        msgs_tc = [_assistant_tc([("tc1", "read_file", '{"path": "foo.txt"}')])]
        # The tool call version should have more tokens than an empty-content message
        estimate_tokens(msgs_plain)
        count_tc = estimate_tokens(msgs_tc)
        assert count_tc > 4  # More than just per-message overhead

    def test_includes_reasoning_content(self):
        msgs_plain = [_assistant("hello")]
        msgs_reasoning = [
            {
                "role": "assistant",
                "content": "hello",
                "reasoning_content": "internal scratch " * 200,
            }
        ]
        assert estimate_tokens(msgs_reasoning) > estimate_tokens(msgs_plain)

    def test_tools_schema_counted(self):
        msgs = [_user("hi")]
        tools = [
            {"type": "function", "function": {"name": "read_file", "parameters": {}}}
        ]
        count_no_tools = estimate_tokens(msgs)
        count_with_tools = estimate_tokens(msgs, tools)
        assert count_with_tools > count_no_tools

    def test_empty_messages(self):
        assert estimate_tokens([]) == 0

    def test_none_content(self):
        # Assistant messages with tool_calls often have content=None
        msgs = [{"role": "assistant", "content": None}]
        count = estimate_tokens(msgs)
        assert count == 4  # Just per-message overhead

    def test_dict_tool_calls_counted(self):
        """Tool calls in dict-shaped messages should be counted too."""
        msgs_no_tc = [{"role": "assistant", "content": None}]
        msgs_with_tc = [
            {
                "role": "assistant",
                "content": None,
                "tool_calls": [
                    {
                        "id": "tc1",
                        "function": {
                            "name": "read_file",
                            "arguments": '{"path": "foo.txt"}',
                        },
                    }
                ],
            }
        ]
        count_no_tc = estimate_tokens(msgs_no_tc)
        count_with_tc = estimate_tokens(msgs_with_tc)
        assert count_with_tc > count_no_tc


# ---------------------------------------------------------------------------
# group_into_turns
# ---------------------------------------------------------------------------


class TestGroupIntoTurns:
    def test_basic(self):
        msgs = [_sys("sys"), _user("q"), _assistant("a")]
        turns = group_into_turns(msgs)
        assert len(turns) == 3
        assert all(len(t) == 1 for t in turns)

    def test_tool_calls(self):
        tc = _assistant_tc([("tc1", "read_file", "{}"), ("tc2", "grep", "{}")])
        tr1 = _tool("tc1", "content1")
        tr2 = _tool("tc2", "content2")
        msgs = [_sys("sys"), _user("q"), tc, tr1, tr2, _assistant("done")]
        turns = group_into_turns(msgs)
        assert len(turns) == 4  # sys, user, (tc+tr1+tr2), assistant
        assert len(turns[2]) == 3  # assistant + 2 tool results

    def test_partial_orphaned_tool_result(self):
        # A tool result without a preceding assistant with matching tool_calls
        # should be kept as a standalone turn (defensive)
        orphan = _tool("tc_orphan", "data")
        msgs = [_user("q"), orphan]
        turns = group_into_turns(msgs)
        assert len(turns) == 2
        assert turns[1] == [orphan]


# ---------------------------------------------------------------------------
# compact_messages
# ---------------------------------------------------------------------------


class TestCompactMessages:
    def test_truncates_large_results(self):
        tc = _assistant_tc([("tc1", "read_file", "{}")])
        big_content = "x" * 2000
        tr = _tool("tc1", big_content)
        # Add another turn after so the tool turn is not in the last 2
        msgs = [_sys("sys"), _user("q"), tc, tr, _assistant("mid"), _assistant("done")]
        result = compact_messages(msgs)
        # Find the tool result
        tool_msgs = [
            m
            for m in result
            if (m.get("role") if isinstance(m, dict) else None) == "tool"
        ]
        assert len(tool_msgs) == 1
        assert tool_msgs[0]["content"].startswith("[read_file:")
        assert "compacted" in tool_msgs[0]["content"]

    def test_preserves_recent_turns(self):
        """Last 2 turns should not be compacted."""
        tc1 = _assistant_tc([("tc1", "f", "{}")])
        tr1 = _tool("tc1", "x" * 2000)
        tc2 = _assistant_tc([("tc2", "f", "{}")])
        tr2 = _tool("tc2", "y" * 2000)
        msgs = [_sys("s"), _user("q"), tc1, tr1, tc2, tr2]
        result = compact_messages(msgs)
        # tc2+tr2 is the last turn, tc1+tr1 is second-to-last
        # Both are in the last 2 turns, so neither should be compacted
        tool_msgs = [
            m
            for m in result
            if (m.get("role") if isinstance(m, dict) else None) == "tool"
        ]
        for tm in tool_msgs:
            assert not tm["content"].startswith("[compacted")

    def test_preserves_turn_atomicity(self):
        tc = _assistant_tc([("tc1", "read_file", "{}"), ("tc2", "grep", "{}")])
        tr1 = _tool("tc1", "x" * 2000)
        tr2 = _tool("tc2", "short")
        # Ensure this turn is not in the last 2
        msgs = [_sys("s"), _user("q"), tc, tr1, tr2, _assistant("a"), _assistant("b")]
        result = compact_messages(msgs)
        # Both the assistant with tool_calls and tool results should still be present
        turns = group_into_turns(result)
        # Find the turn with tool calls
        tc_turn = [t for t in turns if len(t) > 1]
        assert len(tc_turn) == 1
        assert len(tc_turn[0]) == 3  # assistant + 2 tool results


# ---------------------------------------------------------------------------
# compact_context
# ---------------------------------------------------------------------------


class TestCompactContext:
    def test_strips_reasoning_payloads(self):
        tc = _assistant_tc([("tc1", "read_file", '{"file_path": "a.py"}')])
        tc.reasoning_content = "thinking " * 1000
        msgs = [_sys("s"), _user("q"), tc, _tool("tc1", "ok"), _assistant("done")]

        result = compact_context(
            CompactionContext(
                messages=msgs,
                tools=None,
                context_length=10_000,
                max_output_tokens=1000,
                attempted_strategies=(COMPACTION_COMPACT_MESSAGES,),
                model_id="generic-model",
            )
        )

        assert result.strategy == COMPACTION_STRIP_REASONING
        assert tc.reasoning_content is None
        assert result.tokens_after < result.tokens_before

    def test_keeps_required_reasoning_placeholder(self):
        tc = _assistant_tc([("tc1", "read_file", '{"file_path": "a.py"}')])
        tc.reasoning_content = "thinking " * 1000
        msgs = [_sys("s"), _user("q"), tc, _tool("tc1", "ok"), _assistant("done")]

        compact_context(
            CompactionContext(
                messages=msgs,
                tools=None,
                context_length=10_000,
                max_output_tokens=1000,
                attempted_strategies=(COMPACTION_COMPACT_MESSAGES,),
                model_id="deepseek-v4",
            )
        )

        assert tc.reasoning_content == " "

    def test_drop_tools_is_request_local(self):
        tools = [
            {"type": "function", "function": {"name": "read_file", "parameters": {}}}
        ]
        msgs = [_sys("s"), _user("q")]

        result = compact_context(
            CompactionContext(
                messages=msgs,
                tools=tools,
                context_length=100,
                max_output_tokens=20,
                attempted_strategies=(
                    COMPACTION_COMPACT_MESSAGES,
                    COMPACTION_DROP_MIDDLE,
                    COMPACTION_AGGRESSIVE,
                ),
            )
        )

        assert result.strategy == COMPACTION_DROP_TOOLS
        assert result.tools is None
        assert tools
        assert result.history_mutated is False


# ---------------------------------------------------------------------------
# compact_tool_result
# ---------------------------------------------------------------------------


class TestCompactToolResult:
    def test_short_content_unchanged(self):
        content = "short result"
        assert (
            compact_tool_result("read_file", {"file_path": "f.py"}, content) == content
        )

    def test_read_file(self):
        content = "line\n" * 500  # >1000 chars, 500 newlines
        result = compact_tool_result("read_file", {"file_path": "src/app.py"}, content)
        assert result.startswith("[read_file: src/app.py,")
        assert "500 lines" in result
        assert "compacted" in result

    def test_grep(self):
        content = "match\n" * 300
        result = compact_tool_result(
            "grep", {"pattern": "TODO", "path": "src/"}, content
        )
        assert result.startswith("[grep: 'TODO' in src/,")
        assert "~300 matches" in result
        assert "compacted" in result

    def test_list_files(self):
        content = "file.py\n" * 200
        result = compact_tool_result(
            "list_files", {"pattern": "*.py", "path": "/project"}, content
        )
        assert result.startswith("[list_files: '*.py' in /project,")
        assert "~200 entries" in result

    def test_run_command(self):
        content = "output " * 200  # >1000 chars
        result = compact_tool_result(
            "run_command", {"command": ["pytest", "-v"]}, content
        )
        assert "[run_command: `pytest -v`" in result
        assert "first 200 chars" in result
        assert "last 200 chars" in result

    def test_run_command_string_cmd(self):
        content = "x" * 2000
        result = compact_tool_result("run_command", {"command": "ls -la"}, content)
        assert "`ls -la`" in result

    def test_run_shell_command(self):
        content = "output " * 200  # >1000 chars
        result = compact_tool_result(
            "run_shell_command", {"command": "ls -la | head"}, content
        )
        assert "[run_shell_command: `ls -la | head`" in result
        assert "first 200 chars" in result
        assert "last 200 chars" in result

    def test_run_shell_command_array_fallback(self):
        content = "x" * 2000
        result = compact_tool_result(
            "run_shell_command", {"command": ["echo", "hi"]}, content
        )
        assert "[run_shell_command: `echo hi`" in result

    def test_fetch_url(self):
        content = "page content " * 200
        result = compact_tool_result(
            "fetch_url", {"url": "https://example.com"}, content
        )
        assert "[fetch_url: https://example.com," in result
        assert "chars" in result
        assert "compacted" in result

    def test_unknown_tool(self):
        content = "x" * 2000
        result = compact_tool_result("some_new_tool", {}, content)
        assert "[some_new_tool:" in result
        assert "2000" in result

    def test_missing_args_keys(self):
        content = "x" * 2000
        result = compact_tool_result("read_file", {}, content)
        assert "[read_file: ?," in result

    def test_none_args(self):
        content = "x" * 2000
        result = compact_tool_result("grep", None, content)
        assert "[grep:" in result
        assert "compacted" in result

    def test_exactly_1000_chars_unchanged(self):
        content = "x" * 1000
        assert (
            compact_tool_result("read_file", {"file_path": "f.py"}, content) == content
        )

    def test_1001_chars_compacted(self):
        content = "x" * 1001
        result = compact_tool_result("read_file", {"file_path": "f.py"}, content)
        assert result != content
        assert "compacted" in result

    def test_read_multiple_files_batch_list(self):
        content = "x" * 2000
        result = compact_tool_result(
            "read_multiple_files",
            {"files": [{"file_path": "a.py"}, {"file_path": "b.py"}]},
            content,
        )
        assert result.startswith("[read_multiple_files: a.py, b.py,")
        assert "compacted" in result

    def test_read_multiple_files_bare_string(self):
        content = "x" * 2000
        result = compact_tool_result(
            "read_multiple_files",
            {"files": "x.py"},
            content,
        )
        assert result.startswith("[read_multiple_files: x.py,")
        assert "compacted" in result

    def test_read_multiple_files_mixed_entries(self):
        content = "x" * 2000
        result = compact_tool_result(
            "read_multiple_files",
            {"files": [{"file_path": "a.py"}, "b.py", 42]},
            content,
        )
        assert "a.py" in result
        assert "b.py" in result
        assert "?" in result
        assert "compacted" in result

    def test_outline_single_file(self):
        content = "x" * 2000
        result = compact_tool_result("outline", {"file_path": "src/app.py"}, content)
        assert result == "[outline: src/app.py — compacted]"

    def test_outline_batch_list(self):
        content = "x" * 2000
        result = compact_tool_result(
            "outline",
            {"files": [{"file_path": "a.py"}, {"file_path": "b.py"}]},
            content,
        )
        assert result.startswith("[outline: a.py, b.py,")
        assert "compacted" in result

    def test_outline_batch_bare_string(self):
        content = "x" * 2000
        result = compact_tool_result("outline", {"files": "x.py"}, content)
        assert result.startswith("[outline: x.py,")
        assert "compacted" in result

    def test_outline_batch_mixed_entries(self):
        content = "x" * 2000
        result = compact_tool_result(
            "outline",
            {"files": [{"file_path": "a.py"}, "b.py", 42]},
            content,
        )
        assert "a.py" in result
        assert "b.py" in result
        assert "?" in result
        assert "compacted" in result

    def test_mcp_tool_compacted_with_head(self):
        content = "abcdefgh" * 250  # 2000 chars
        result = compact_tool_result("mcp__server__tool", {}, content)
        assert result.startswith("[mcp__server__tool: 2000 chars")
        assert "compacted" in result
        assert "First 300 chars" in result
        assert content[:300] in result

    def test_mcp_tool_exactly_1000_unchanged(self):
        content = "x" * 1000
        assert compact_tool_result("mcp__server__tool", {}, content) == content

    def test_mcp_tool_1001_compacted(self):
        content = "y" * 1001
        result = compact_tool_result("mcp__server__tool", {}, content)
        assert result != content
        assert "mcp__server__tool" in result
        assert "compacted" in result

    def test_compact_messages_uses_structured_summaries(self):
        """compact_messages should produce per-tool summaries, not generic ones."""
        tc = _assistant_tc([("tc1", "grep", '{"pattern": "error", "path": "logs/"}')])
        big_content = "match: error found\n" * 100  # >1000 chars
        tr = _tool("tc1", big_content)
        msgs = [_sys("sys"), _user("q"), tc, tr, _assistant("mid"), _assistant("done")]
        result = compact_messages(msgs)
        tool_msgs = [
            m
            for m in result
            if (m.get("role") if isinstance(m, dict) else None) == "tool"
        ]
        assert len(tool_msgs) == 1
        assert tool_msgs[0]["content"].startswith("[grep:")
        assert "'error'" in tool_msgs[0]["content"]
        assert "logs/" in tool_msgs[0]["content"]

    def test_grep_compaction_with_context(self):
        """Compaction uses the header count, not newline count, with context."""
        # Simulate grep output with context_lines — many newlines, but only 2 matches
        # Pad with enough context lines to exceed the 1000-char compaction threshold
        ctx_lines = "\n".join(f"  Line {i}: {'x' * 80}" for i in range(50))
        content = (
            f"Found 2 matches\n\nfile.py:\n{ctx_lines}\n"
            "  Line 100: match1  <<<\n"
            "  --\n"
            "  Line 200: match2  <<<\n"
        )
        assert len(content) > 1000  # ensure compaction triggers
        result = compact_tool_result("grep", {"pattern": "match", "path": "."}, content)
        assert "~2 matches" in result


# ---------------------------------------------------------------------------
# is_pinned / score_turn
# ---------------------------------------------------------------------------


class TestIsPinned:
    def test_user_turn_is_pinned(self):
        assert is_pinned([_user("hello")]) is True

    def test_assistant_turn_not_pinned(self):
        assert is_pinned([_assistant("response")]) is False

    def test_tool_turn_not_pinned(self):
        tc = _assistant_tc([("tc1", "read_file", "{}")])
        tr = _tool("tc1", "content")
        assert is_pinned([tc, tr]) is False

    def test_system_turn_not_pinned(self):
        assert is_pinned([_sys("system prompt")]) is False


class TestScoreTurn:
    def test_error_content_scores_high(self):
        tc = _assistant_tc([("tc1", "run_command", "{}")])
        tr = _tool("tc1", "error: command failed")
        score = score_turn([tc, tr])
        assert score >= 3

    def test_file_edit_scores_high(self):
        tc = _assistant_tc([("tc1", "edit_file", "{}")])
        tr = _tool("tc1", "ok")
        score = score_turn([tc, tr])
        assert score >= 5

    def test_write_file_scores_high(self):
        tc = _assistant_tc([("tc1", "write_file", "{}")])
        tr = _tool("tc1", "ok")
        score = score_turn([tc, tr])
        assert score >= 5

    def test_read_file_scores_low(self):
        tc = _assistant_tc([("tc1", "read_file", "{}")])
        tr = _tool("tc1", "file contents here")
        score = score_turn([tc, tr])
        assert score == 0

    def test_plain_assistant_scores_zero(self):
        assert score_turn([_assistant("just thinking")]) == 0

    def test_error_in_assistant_content(self):
        score = score_turn([_assistant("I encountered an error in the code")])
        assert score >= 3

    def test_combined_scores_accumulate(self):
        # A turn with both an error and a file edit should score higher
        tc = _assistant_tc([("tc1", "edit_file", "{}")])
        tr = _tool("tc1", "error: partial write failed")
        score = score_turn([tc, tr])
        assert score >= 8  # 5 (edit) + 3 (error)


# ---------------------------------------------------------------------------
# drop_middle_turns
# ---------------------------------------------------------------------------


class TestDropMiddleTurns:
    def test_keeps_boundaries(self):
        tc1 = _assistant_tc([("tc1", "f", "{}")])
        tr1 = _tool("tc1", "result1")
        tc2 = _assistant_tc([("tc2", "f", "{}")])
        tr2 = _tool("tc2", "result2")
        tc3 = _assistant_tc([("tc3", "f", "{}")])
        tr3 = _tool("tc3", "result3")
        tc4 = _assistant_tc([("tc4", "f", "{}")])
        tr4 = _tool("tc4", "result4")
        msgs = [_sys("sys"), _user("q"), tc1, tr1, tc2, tr2, tc3, tr3, tc4, tr4]
        result = drop_middle_turns(msgs)
        # Should have: sys, user, splice marker, last 3 turns (tc2+tr2, tc3+tr3, tc4+tr4)
        roles = []
        for m in result:
            r = m.get("role") if isinstance(m, dict) else getattr(m, "role", None)
            roles.append(r)
        assert roles[0] == "system"
        assert roles[1] == "user"
        assert roles[2] == "user"  # splice marker
        assert "[context compacted" in result[2]["content"]

    def test_no_system(self):
        """Works correctly when there's no system message."""
        tc1 = _assistant_tc([("tc1", "f", "{}")])
        tr1 = _tool("tc1", "r1")
        tc2 = _assistant_tc([("tc2", "f", "{}")])
        tr2 = _tool("tc2", "r2")
        tc3 = _assistant_tc([("tc3", "f", "{}")])
        tr3 = _tool("tc3", "r3")
        tc4 = _assistant_tc([("tc4", "f", "{}")])
        tr4 = _tool("tc4", "r4")
        msgs = [_user("q"), tc1, tr1, tc2, tr2, tc3, tr3, tc4, tr4]
        result = drop_middle_turns(msgs)
        # Leading block is just the user message
        assert result[0]["role"] == "user"
        assert result[0]["content"] == "q"
        # Splice marker
        assert "[context compacted" in result[1]["content"]

    def test_preserves_turn_atomicity(self):
        tc1 = _assistant_tc([("tc1", "f", "{}"), ("tc1b", "g", "{}")])
        tr1a = _tool("tc1", "r1")
        tr1b = _tool("tc1b", "r1b")
        tc2 = _assistant_tc([("tc2", "f", "{}")])
        tr2 = _tool("tc2", "r2")
        tc3 = _assistant_tc([("tc3", "f", "{}")])
        tr3 = _tool("tc3", "r3")
        tc4 = _assistant_tc([("tc4", "f", "{}")])
        tr4 = _tool("tc4", "r4")
        msgs = [_sys("s"), _user("q"), tc1, tr1a, tr1b, tc2, tr2, tc3, tr3, tc4, tr4]
        result = drop_middle_turns(msgs)
        # Verify no orphaned tool results
        _validate_tool_pairing(result)

    def test_small_history(self):
        """When history is too small for a middle, returns unchanged."""
        msgs = [_sys("s"), _user("q"), _assistant("a")]
        result = drop_middle_turns(msgs)
        assert len(result) == 3
        # No splice marker
        for m in result:
            if isinstance(m, dict) and m.get("content"):
                assert "[context compacted" not in m["content"]

    def test_user_turns_never_dropped(self):
        """User turns in the middle must be preserved (pinned)."""
        tc1 = _assistant_tc([("tc1", "read_file", "{}")])
        tr1 = _tool("tc1", "r1")
        user_mid = _user("can you also check bar.py?")
        tc2 = _assistant_tc([("tc2", "read_file", "{}")])
        tr2 = _tool("tc2", "r2")
        tc3 = _assistant_tc([("tc3", "read_file", "{}")])
        tr3 = _tool("tc3", "r3")
        tc4 = _assistant_tc([("tc4", "read_file", "{}")])
        tr4 = _tool("tc4", "r4")
        tc5 = _assistant_tc([("tc5", "read_file", "{}")])
        tr5 = _tool("tc5", "r5")
        msgs = [
            _sys("s"),
            _user("q"),
            tc1,
            tr1,
            user_mid,
            tc2,
            tr2,
            tc3,
            tr3,
            tc4,
            tr4,
            tc5,
            tr5,
        ]
        result = drop_middle_turns(msgs)
        # The mid-conversation user message must survive
        user_contents = [
            m["content"]
            for m in result
            if isinstance(m, dict) and m.get("role") == "user"
        ]
        assert "can you also check bar.py?" in user_contents

    def test_high_scoring_turns_kept(self):
        """Turns with file edits should be kept over plain reads."""
        # Create a history with many middle turns
        tc_read1 = _assistant_tc([("r1", "read_file", "{}")])
        tr_read1 = _tool("r1", "file contents")
        tc_read2 = _assistant_tc([("r2", "read_file", "{}")])
        tr_read2 = _tool("r2", "more contents")
        tc_edit = _assistant_tc([("e1", "edit_file", "{}")])
        tr_edit = _tool("e1", "ok")
        tc_read3 = _assistant_tc([("r3", "read_file", "{}")])
        tr_read3 = _tool("r3", "contents")
        # Tail turns
        tc_tail1 = _assistant_tc([("t1", "read_file", "{}")])
        tr_tail1 = _tool("t1", "t")
        tc_tail2 = _assistant_tc([("t2", "read_file", "{}")])
        tr_tail2 = _tool("t2", "t")
        tc_tail3 = _assistant_tc([("t3", "read_file", "{}")])
        tr_tail3 = _tool("t3", "t")
        msgs = [
            _sys("s"),
            _user("q"),
            tc_read1,
            tr_read1,
            tc_read2,
            tr_read2,
            tc_edit,
            tr_edit,
            tc_read3,
            tr_read3,
            tc_tail1,
            tr_tail1,
            tc_tail2,
            tr_tail2,
            tc_tail3,
            tr_tail3,
        ]
        result = drop_middle_turns(msgs)
        # The edit turn (high score) should still be in the result
        tc_ids_in_result = set()
        for m in result:
            tcs = (
                m.get("tool_calls", None)
                if isinstance(m, dict)
                else getattr(m, "tool_calls", None)
            )
            if tcs:
                for tc in tcs:
                    fn = (
                        tc.function
                        if hasattr(tc, "function")
                        else tc.get("function", {})
                    )
                    fn_name = fn.name if hasattr(fn, "name") else fn.get("name", "")
                    tc_ids_in_result.add(fn_name)
        assert "edit_file" in tc_ids_in_result


# ---------------------------------------------------------------------------
# clamp_output_tokens
# ---------------------------------------------------------------------------


class TestClampOutputTokens:
    def test_basic_clamping(self):
        msgs = [_user("hello " * 100)]  # Should be a decent number of tokens
        # With a tight context_length, output should be clamped
        result = clamp_output_tokens(msgs, None, 200, 16384)
        assert result < 16384
        assert result > 0

    def test_none_context_length(self):
        result = clamp_output_tokens([_user("hi")], None, None, 16384)
        assert result == 16384

    def test_available_below_minimum_raises(self):
        # When prompt leaves fewer than MIN_OUTPUT_TOKENS, raise ContextOverflowError
        msgs = [_user("x " * 10000)]
        with pytest.raises(ContextOverflowError):
            clamp_output_tokens(msgs, None, 10, 16384)

    def test_no_clamping_when_room(self):
        msgs = [_user("hi")]
        result = clamp_output_tokens(msgs, None, 100000, 16384)
        assert result == 16384

    def test_none_max_output_passes_through(self):
        result = clamp_output_tokens([_user("hi")], None, 100000, None)
        assert result is None

    def test_none_max_output_and_none_context(self):
        result = clamp_output_tokens([_user("hi")], None, None, None)
        assert result is None


# ---------------------------------------------------------------------------
# Integration: compacted messages valid for API
# ---------------------------------------------------------------------------


def _validate_tool_pairing(messages):
    """Validate that every tool result has a matching tool_call_id in a preceding assistant message."""
    # Collect all tool_call_ids from assistant messages
    available_tc_ids = set()
    for m in messages:
        role = m.get("role") if isinstance(m, dict) else getattr(m, "role", None)
        if role == "assistant":
            tcs = (
                m.get("tool_calls", None)
                if isinstance(m, dict)
                else getattr(m, "tool_calls", None)
            )
            if tcs:
                for tc in tcs:
                    tc_id = tc.id if hasattr(tc, "id") else tc["id"]
                    available_tc_ids.add(tc_id)
        elif role == "tool":
            tc_id = (
                m.get("tool_call_id")
                if isinstance(m, dict)
                else getattr(m, "tool_call_id", None)
            )
            assert tc_id in available_tc_ids, (
                f"Orphaned tool result with tool_call_id={tc_id}"
            )


class TestIntegration:
    def _build_realistic_history(self):
        """Build a realistic message sequence with multiple tool turns."""
        msgs = [
            _sys("You are a helpful assistant."),
            _user("Read foo.txt and bar.txt"),
        ]
        # Turn 1: read_file foo.txt
        tc1 = _assistant_tc([("tc1", "read_file", '{"path": "foo.txt"}')])
        tr1 = _tool("tc1", "contents of foo " * 100)
        msgs.extend([tc1, tr1])
        # Turn 2: read_file bar.txt
        tc2 = _assistant_tc([("tc2", "read_file", '{"path": "bar.txt"}')])
        tr2 = _tool("tc2", "contents of bar " * 100)
        msgs.extend([tc2, tr2])
        # Turn 3: grep
        tc3 = _assistant_tc([("tc3", "grep", '{"pattern": "TODO"}')])
        tr3 = _tool("tc3", "line1: TODO fix\nline2: TODO refactor\n" * 50)
        msgs.extend([tc3, tr3])
        # Turn 4: write_file
        tc4 = _assistant_tc(
            [("tc4", "write_file", '{"path": "out.txt", "content": "done"}')]
        )
        tr4 = _tool("tc4", "ok")
        msgs.extend([tc4, tr4])
        # Final assistant
        msgs.append(_assistant("I've completed the task."))
        return msgs

    def test_compact_then_valid_for_api(self):
        msgs = self._build_realistic_history()
        result = compact_messages(msgs)
        _validate_tool_pairing(result)

    def test_drop_then_valid_for_api(self):
        msgs = self._build_realistic_history()
        result = drop_middle_turns(msgs)
        _validate_tool_pairing(result)


# ---------------------------------------------------------------------------
# ContextOverflowError classifier
# ---------------------------------------------------------------------------


class TestContextOverflowClassifier:
    def test_typed_exception(self):
        """call_llm raises ContextOverflowError for litellm.ContextWindowExceededError."""
        import litellm

        with patch("litellm.completion") as mock_comp:
            mock_comp.side_effect = litellm.ContextWindowExceededError(
                message="context length exceeded",
                model="test",
                llm_provider="openai",
            )
            with pytest.raises(ContextOverflowError):
                call_llm(
                    "http://localhost", "model", [], 100, 0.1, 1.0, None, None, False
                )

    def test_bad_request_with_context_keywords(self):
        """call_llm raises ContextOverflowError for BadRequestError with context keywords."""
        import litellm

        with patch("litellm.completion") as mock_comp:
            mock_comp.side_effect = litellm.BadRequestError(
                message="maximum context length exceeded",
                model="test",
                llm_provider="openai",
            )
            with pytest.raises(ContextOverflowError):
                call_llm(
                    "http://localhost", "model", [], 100, 0.1, 1.0, None, None, False
                )

    def test_bad_request_without_context_keywords(self):
        """call_llm raises AgentError for BadRequestError without context keywords."""
        import litellm

        with patch("litellm.completion") as mock_comp:
            mock_comp.side_effect = litellm.BadRequestError(
                message="invalid request format",
                model="test",
                llm_provider="openai",
            )
            with pytest.raises(AgentError):
                call_llm(
                    "http://localhost", "model", [], 100, 0.1, 1.0, None, None, False
                )

    def test_api_error_with_context_keywords(self):
        """call_llm raises ContextOverflowError for APIError with context keywords."""
        import litellm

        with patch("litellm.completion") as mock_comp:
            mock_comp.side_effect = litellm.APIError(
                message="ChatgptException - Your input exceeds the context window of this model.",
                status_code=500,
                model="test",
                llm_provider="openai",
            )
            with pytest.raises(ContextOverflowError):
                call_llm(
                    "http://localhost", "model", [], 100, 0.1, 1.0, None, None, False
                )

    def test_api_error_without_context_keywords(self):
        """call_llm raises AgentError for APIError without context keywords."""
        import litellm

        with patch("litellm.completion") as mock_comp:
            mock_comp.side_effect = litellm.APIError(
                message="internal server error",
                status_code=500,
                model="test",
                llm_provider="openai",
            )
            with pytest.raises(AgentError):
                call_llm(
                    "http://localhost",
                    "model",
                    [],
                    100,
                    0.1,
                    1.0,
                    None,
                    None,
                    False,
                    max_retries=1,
                )

    def test_call_llm_omits_tool_choice_when_tools_none(self):
        """When tools=None, call_llm should not include tool_choice in kwargs."""
        with patch("litellm.completion") as mock_comp:
            mock_comp.return_value = SimpleNamespace(
                choices=[
                    SimpleNamespace(
                        message=SimpleNamespace(content="summary", tool_calls=None),
                        finish_reason="stop",
                    )
                ]
            )
            call_llm(
                "http://localhost",
                "model",
                [{"role": "user", "content": "hi"}],
                100,
                0.1,
                1.0,
                None,
                None,
                False,
            )
            # litellm.completion was called with keyword args
            call_kw = mock_comp.call_args.kwargs
            assert "tool_choice" not in call_kw
            assert "tools" not in call_kw


# ---------------------------------------------------------------------------
# ToolsNotSupportedError
# ---------------------------------------------------------------------------

_DUMMY_TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "dummy",
            "parameters": {"type": "object", "properties": {}},
        },
    }
]


class TestToolsNotSupported:
    """Tests for ToolsNotSupportedError detection in call_llm."""

    def test_hf_function_calling_not_supported_raises(self):
        """call_llm raises ToolsNotSupportedError for HuggingFace models that
        reject function calling."""
        import litellm
        from swival.report import ToolsNotSupportedError

        with patch("litellm.completion") as mock_comp:
            mock_comp.side_effect = litellm.BadRequestError(
                message=(
                    'HuggingfaceException - {"code":400,'
                    '"reason":"INVALID_REQUEST_BODY",'
                    '"message":"model features function calling not support",'
                    '"metadata":{}}'
                ),
                model="huggingface/google/gemma-4-31B-it",
                llm_provider="huggingface",
            )
            with pytest.raises(ToolsNotSupportedError):
                call_llm(
                    "http://localhost",
                    "model",
                    [{"role": "user", "content": "hi"}],
                    100,
                    0.1,
                    1.0,
                    None,
                    _DUMMY_TOOLS,
                    False,
                    max_retries=1,
                )

    def test_tools_not_supported_not_raised_when_tools_none(self):
        """When tools=None, a matching BadRequestError should NOT raise
        ToolsNotSupportedError (it's a different problem)."""
        import litellm

        with patch("litellm.completion") as mock_comp:
            mock_comp.side_effect = litellm.BadRequestError(
                message="model features function calling not support",
                model="test",
                llm_provider="huggingface",
            )
            with pytest.raises(AgentError) as exc_info:
                call_llm(
                    "http://localhost",
                    "model",
                    [{"role": "user", "content": "hi"}],
                    100,
                    0.1,
                    1.0,
                    None,
                    None,
                    False,
                    max_retries=1,
                )
            from swival.report import ToolsNotSupportedError

            assert not isinstance(exc_info.value, ToolsNotSupportedError)

    def test_generic_does_not_support_tools(self):
        """call_llm raises ToolsNotSupportedError for 'does not support tools'."""
        import litellm
        from swival.report import ToolsNotSupportedError

        with patch("litellm.completion") as mock_comp:
            mock_comp.side_effect = litellm.BadRequestError(
                message="This model does not support tools",
                model="test",
                llm_provider="openai",
            )
            with pytest.raises(ToolsNotSupportedError):
                call_llm(
                    "http://localhost",
                    "model",
                    [{"role": "user", "content": "hi"}],
                    100,
                    0.1,
                    1.0,
                    None,
                    _DUMMY_TOOLS,
                    False,
                    max_retries=1,
                )


# ---------------------------------------------------------------------------
# ToolsNotSupportedError — run_agent_loop integration
# ---------------------------------------------------------------------------


class TestToolsNotSupportedLoop:
    """Integration tests: ToolsNotSupportedError fallback in run_agent_loop."""

    @staticmethod
    def _loop_kwargs(tmp_path, **overrides):
        from swival.thinking import ThinkingState
        from swival.todo import TodoState

        defaults = dict(
            api_base="http://127.0.0.1:1234",
            model_id="test-model",
            max_turns=1,
            max_output_tokens=1024,
            temperature=0.5,
            top_p=None,
            seed=None,
            context_length=None,
            base_dir=str(tmp_path),
            thinking_state=ThinkingState(verbose=False),
            resolved_commands={},
            skills_catalog={},
            skill_read_roots=[],
            extra_write_roots=[],
            files_mode="some",
            verbose=False,
            llm_kwargs={"provider": "lmstudio", "api_key": None},
            file_tracker=None,
            todo_state=TodoState(verbose=False),
        )
        defaults.update(overrides)
        return defaults

    def test_max_turns_1_retries_without_tools(self, tmp_path):
        """With max_turns=1 the plain-chat retry must still fire after
        ToolsNotSupportedError — the failed tool-enabled call should NOT
        consume the only turn."""
        from swival.agent import run_agent_loop
        from swival.report import ToolsNotSupportedError as _TNS

        call_count = 0

        def fake_call_llm(*args, **kwargs):
            nonlocal call_count
            call_count += 1
            # First call: has tools → raise
            tools_arg = args[7] if len(args) > 7 else kwargs.get("tools")
            if tools_arg is not None:
                tne = _TNS("function calling not support")
                tne._provider_retries = 0
                raise tne
            # Second call: no tools → succeed
            return (
                SimpleNamespace(
                    content="plain answer", tool_calls=None, role="assistant"
                ),
                "stop",
                [],
                0,
                (0, 0),
            )

        messages = [_sys("system"), _user("hello")]
        with patch("swival.agent.call_llm", side_effect=fake_call_llm):
            answer, exhausted = run_agent_loop(
                messages,
                _DUMMY_TOOLS,
                **self._loop_kwargs(tmp_path, max_turns=1),
            )

        assert answer == "plain answer"
        assert exhausted is False
        assert call_count == 2

    def test_report_records_both_calls(self, tmp_path):
        """The failed tool-enabled call and the successful plain-chat retry
        must both appear in the report."""
        from swival.agent import run_agent_loop
        from swival.report import ReportCollector, ToolsNotSupportedError as _TNS

        def fake_call_llm(*args, **kwargs):
            tools_arg = args[7] if len(args) > 7 else kwargs.get("tools")
            if tools_arg is not None:
                tne = _TNS("function calling not support")
                tne._provider_retries = 0
                raise tne
            return (
                SimpleNamespace(content="ok", tool_calls=None, role="assistant"),
                "stop",
                [],
                0,
                (0, 0),
            )

        report = ReportCollector()
        messages = [_sys("system"), _user("hello")]
        with patch("swival.agent.call_llm", side_effect=fake_call_llm):
            run_agent_loop(
                messages,
                _DUMMY_TOOLS,
                **self._loop_kwargs(tmp_path, max_turns=3, report=report),
            )

        llm_events = [e for e in report.events if e["type"] == "llm_call"]
        assert len(llm_events) == 2
        assert llm_events[0]["finish_reason"] == "tools_not_supported"
        assert llm_events[1]["finish_reason"] == "stop"
        # The successful retry must be tagged as a retry
        assert llm_events[1].get("is_retry") is True
        assert llm_events[1].get("retry_reason") == "drop_tools_unsupported"

    def test_huggingface_non_chat_model_recovers_via_text_generation(self, tmp_path):
        """HF models that reject chat completions should recover after tools drop."""
        import litellm
        from swival.agent import run_agent_loop

        error = litellm.BadRequestError(
            message="The requested model 'google/gemma-4-E4B-it' is not a chat model.",
            model="huggingface/google/gemma-4-E4B-it",
            llm_provider="huggingface",
        )
        client = SimpleNamespace(text_generation=lambda *args, **kwargs: "plain answer")
        info = SimpleNamespace(
            inference="warm",
            inference_provider_mapping=[],
            pipeline_tag="text-generation",
        )
        messages = [_sys("system"), _user("hello")]

        with (
            patch("litellm.completion", side_effect=error),
            patch("huggingface_hub.InferenceClient", return_value=client),
            patch("huggingface_hub.HfApi") as mock_hf_api,
        ):
            mock_hf_api.return_value.model_info.return_value = info
            answer, exhausted = run_agent_loop(
                messages,
                _DUMMY_TOOLS,
                **self._loop_kwargs(
                    tmp_path,
                    api_base=None,
                    model_id="google/gemma-4-E4B-it",
                    llm_kwargs={"provider": "huggingface", "api_key": "hf_test"},
                ),
            )

        assert answer == "plain answer"
        assert exhausted is False

    def test_overflow_then_tools_not_supported_recovers(self, tmp_path):
        """ToolsNotSupportedError discovered during a compaction retry must
        still trigger the tools-drop fallback and eventually succeed."""
        from swival.agent import run_agent_loop, ContextOverflowError
        from swival.report import (
            ReportCollector,
            ToolsNotSupportedError as _TNS,
        )

        call_count = 0

        def fake_call_llm(*args, **kwargs):
            nonlocal call_count
            call_count += 1
            tools_arg = args[7] if len(args) > 7 else kwargs.get("tools")
            if call_count == 1:
                # First call: context overflow (triggers compaction)
                raise ContextOverflowError("context length exceeded")
            if call_count == 2 and tools_arg is not None:
                # Second call (compaction retry, still has tools): tools
                # not supported
                tne = _TNS("function calling not support")
                tne._provider_retries = 0
                raise tne
            # Third call: no tools → succeed
            return (
                SimpleNamespace(content="recovered", tool_calls=None, role="assistant"),
                "stop",
                [],
                0,
                (0, 0),
            )

        report = ReportCollector()
        tc = _assistant_tc([("tc1", "read_file", '{"file_path": "old.py"}')])
        messages = [
            _sys("system"),
            _user("start"),
            tc,
            _tool("tc1", "x" * 5000),
            _assistant("mid"),
            _user("hello"),
        ]
        with patch("swival.agent.call_llm", side_effect=fake_call_llm):
            answer, exhausted = run_agent_loop(
                messages,
                _DUMMY_TOOLS,
                **self._loop_kwargs(tmp_path, max_turns=2, report=report),
            )

        assert answer == "recovered"
        assert exhausted is False
        assert call_count == 3

        llm_events = [e for e in report.events if e["type"] == "llm_call"]
        # Three events: overflow, tools_not_supported, success
        assert len(llm_events) == 3
        assert llm_events[0]["finish_reason"] == "context_overflow"
        assert llm_events[1]["finish_reason"] == "tools_not_supported"
        assert llm_events[2]["finish_reason"] == "stop"
        assert llm_events[2].get("is_retry") is True

    def test_tools_not_supported_then_overflow_uses_compaction_result(self, tmp_path):
        """When tools_not_supported fires first and the no-tools retry
        overflows, the compaction-retry result must be used — the
        _is_tools_retry flag must NOT discard a successful compaction."""
        from swival.agent import run_agent_loop, ContextOverflowError
        from swival.report import (
            ReportCollector,
            ToolsNotSupportedError as _TNS,
        )

        call_count = 0

        def fake_call_llm(*args, **kwargs):
            nonlocal call_count
            call_count += 1
            tools_arg = args[7] if len(args) > 7 else kwargs.get("tools")
            if call_count == 1:
                # First call: tools not supported
                tne = _TNS("function calling not support")
                tne._provider_retries = 0
                raise tne
            if call_count == 2:
                # Second call (no-tools retry): context overflow
                assert tools_arg is None
                raise ContextOverflowError("context length exceeded")
            if call_count == 3:
                # Third call (compaction retry, no tools): succeed
                assert tools_arg is None
                return (
                    SimpleNamespace(
                        content="compaction result",
                        tool_calls=None,
                        role="assistant",
                    ),
                    "stop",
                    [],
                    0,
                    (0, 0),
                )
            # Should not reach here
            return (
                SimpleNamespace(
                    content=f"extra call {call_count}",
                    tool_calls=None,
                    role="assistant",
                ),
                "stop",
                [],
                0,
                (0, 0),
            )

        report = ReportCollector()
        messages = [_sys("system"), _user("hello")]
        with patch("swival.agent.call_llm", side_effect=fake_call_llm):
            answer, exhausted = run_agent_loop(
                messages,
                _DUMMY_TOOLS,
                **self._loop_kwargs(tmp_path, max_turns=2, report=report),
            )

        assert answer == "compaction result"
        assert exhausted is False
        assert call_count == 3  # no spurious fourth call

        llm_events = [e for e in report.events if e["type"] == "llm_call"]
        assert len(llm_events) == 3
        assert llm_events[0]["finish_reason"] == "tools_not_supported"
        # The overflow event should be tagged as a retry (it was the
        # no-tools fallback attempt)
        assert llm_events[1]["finish_reason"] == "context_overflow"
        assert llm_events[1].get("is_retry") is True
        assert llm_events[2]["finish_reason"] == "stop"

    def test_tools_not_supported_then_agent_error_tagged_as_retry(self, tmp_path):
        """When the no-tools retry raises a generic AgentError, the error
        event must be tagged is_retry=True."""
        from swival.agent import run_agent_loop
        from swival.report import (
            AgentError as _AE,
            ReportCollector,
            ToolsNotSupportedError as _TNS,
        )

        call_count = 0

        def fake_call_llm(*args, **kwargs):
            nonlocal call_count
            call_count += 1
            tools_arg = args[7] if len(args) > 7 else kwargs.get("tools")
            if call_count == 1:
                tne = _TNS("function calling not support")
                tne._provider_retries = 0
                raise tne
            # Second call (no-tools retry): fatal error
            assert tools_arg is None
            raise _AE("something else broke")

        report = ReportCollector()
        messages = [_sys("system"), _user("hello")]
        with pytest.raises(_AE, match="something else broke"):
            with patch("swival.agent.call_llm", side_effect=fake_call_llm):
                run_agent_loop(
                    messages,
                    _DUMMY_TOOLS,
                    **self._loop_kwargs(tmp_path, max_turns=2, report=report),
                )

        llm_events = [e for e in report.events if e["type"] == "llm_call"]
        assert len(llm_events) == 2
        assert llm_events[0]["finish_reason"] == "tools_not_supported"
        assert llm_events[1]["finish_reason"] == "error"
        assert llm_events[1].get("is_retry") is True
        assert llm_events[1].get("retry_reason") == "drop_tools_unsupported"

    def test_empty_response_message_no_tools(self, tmp_path):
        """After tools fallback, the empty-response continuation must not
        say 'available tools'."""
        from swival.agent import run_agent_loop
        from swival.report import ToolsNotSupportedError as _TNS

        call_count = 0

        def fake_call_llm(*args, **kwargs):
            nonlocal call_count
            call_count += 1
            if call_count == 1:
                tne = _TNS("function calling not support")
                tne._provider_retries = 0
                raise tne
            if call_count == 2:
                # Return empty response to trigger continuation
                return (
                    SimpleNamespace(content="", tool_calls=None, role="assistant"),
                    "stop",
                    [],
                    0,
                    (0, 0),
                )
            return (
                SimpleNamespace(
                    content="final answer", tool_calls=None, role="assistant"
                ),
                "stop",
                [],
                0,
                (0, 0),
            )

        messages = [_sys("system"), _user("hello")]
        with patch("swival.agent.call_llm", side_effect=fake_call_llm):
            answer, _ = run_agent_loop(
                messages,
                _DUMMY_TOOLS,
                **self._loop_kwargs(tmp_path, max_turns=5),
            )

        assert answer == "final answer"
        # The empty-response continuation should not reference tools
        user_msgs = [m for m in messages if m.get("role") == "user"]
        continuation = user_msgs[-1]["content"]
        assert "available tools" not in continuation
        assert "answer the question directly" in continuation


# ---------------------------------------------------------------------------
# Drop-tools fallback: emergency-truncate retry when server still rejects
# ---------------------------------------------------------------------------


class TestDropToolsEmergencyRetry:
    """When every compaction level *and* the no-tools clamp succeed locally
    but the server still raises ContextOverflowError, the agent must run
    _emergency_truncate and retry instead of giving up.

    Regression for: model rejects clamped prompt because our local tiktoken
    estimate undercounts vs. the model's real tokenizer."""

    @staticmethod
    def _loop_kwargs(tmp_path, **overrides):
        from swival.thinking import ThinkingState
        from swival.todo import TodoState

        defaults = dict(
            api_base="http://127.0.0.1:1234",
            model_id="test-model",
            max_turns=1,
            max_output_tokens=1024,
            temperature=0.5,
            top_p=None,
            seed=None,
            context_length=8000,
            base_dir=str(tmp_path),
            thinking_state=ThinkingState(verbose=False),
            resolved_commands={},
            skills_catalog={},
            skill_read_roots=[],
            extra_write_roots=[],
            files_mode="some",
            verbose=False,
            llm_kwargs={"provider": "lmstudio", "api_key": None},
            file_tracker=None,
            todo_state=TodoState(verbose=False),
        )
        defaults.update(overrides)
        return defaults

    def test_recovers_when_server_rejects_after_drop_tools(self, tmp_path):
        from swival.agent import run_agent_loop

        call_count = 0

        def fake_call_llm(*args, **kwargs):
            nonlocal call_count
            call_count += 1
            tools_arg = args[7] if len(args) > 7 else kwargs.get("tools")
            # The server's real tokenizer counts the tool schemas higher than
            # our estimate, so it rejects the clamped prompt while tools are
            # attached. The reactive net compacts to a budget below the current
            # size, drops the tools for the retry, and the call goes through.
            if tools_arg is not None:
                raise ContextOverflowError(f"too long (call {call_count})")
            return (
                SimpleNamespace(
                    content="recovered after truncation",
                    tool_calls=None,
                    role="assistant",
                ),
                "stop",
                [],
                0,
                (0, 0),
            )

        messages = [_sys("system prompt"), _user("hello")]
        with patch("swival.agent.call_llm", side_effect=fake_call_llm):
            answer, exhausted = run_agent_loop(
                messages,
                _DUMMY_TOOLS,
                **self._loop_kwargs(tmp_path),
            )

        assert answer == "recovered after truncation"
        assert exhausted is False
        # Initial tools call rejected, then one compact-and-retry that drops
        # tools succeeds — no round-trip storm.
        assert call_count == 2

    def test_reactive_carries_dropped_tools_into_next_budget(
        self, tmp_path, monkeypatch
    ):
        """After a round drops tools and the no-tools retry still overflows,
        the next round's budget must be measured against the no-tools prompt
        (so messages shrink) — not re-measured with the full schemas, which
        would just drop tools again and stall."""
        from swival import agent
        from swival.agent import run_agent_loop

        budget_tools = []  # the `tools` arg compact_to_budget saw each round

        def spy_compact_to_budget(messages, tools, **kwargs):
            budget_tools.append(tools)
            if len(budget_tools) == 1:
                # First round: drop the tools for the retry.
                return agent.CompactionResult(
                    messages=messages,
                    tools=None,
                    strategy="drop_tools",
                    description="dropped tools",
                    tokens_before=999,
                    tokens_after=500,
                    history_mutated=False,
                    dropped_tools=True,
                    met_budget=True,
                )
            # Later rounds: shrink the (now no-tools) messages further.
            return agent.CompactionResult(
                messages=messages,
                tools=None,
                strategy="aggressive_drop",
                description="shrank messages",
                tokens_before=500,
                tokens_after=300,
                history_mutated=True,
                dropped_tools=False,
                met_budget=True,
            )

        monkeypatch.setattr(agent, "compact_to_budget", spy_compact_to_budget)

        server_calls = 0

        def fake_call_llm(*args, **kwargs):
            nonlocal server_calls
            server_calls += 1
            tools_arg = args[7] if len(args) > 7 else kwargs.get("tools")
            if server_calls == 1:
                raise ContextOverflowError("initial overflow")
            if server_calls == 2:
                # First reactive retry: tools dropped, still overflows.
                assert tools_arg is None
                raise ContextOverflowError("still too big")
            # Second reactive retry succeeds.
            assert tools_arg is None
            return (
                SimpleNamespace(
                    content="recovered by shrinking messages",
                    tool_calls=None,
                    role="assistant",
                ),
                "stop",
                [],
                0,
                (0, 0),
            )

        messages = [_sys("system prompt"), _user("hello")]
        with patch("swival.agent.call_llm", side_effect=fake_call_llm):
            answer, _ = run_agent_loop(
                messages,
                _DUMMY_TOOLS,
                **self._loop_kwargs(tmp_path),
            )

        assert answer == "recovered by shrinking messages"
        # Round 1 measured against the full tools; round 2 against the dropped
        # (None) tools — the carry-forward fix.
        assert budget_tools[0] is not None
        assert budget_tools[1] is None

    def test_eventual_failure_degrades_to_fallback(self, tmp_path):
        """If even the most aggressive emergency-truncate retry and the terminal
        floor fail, the turn degrades to the local fallback (exhausted, no
        raise) and preserves continue state — it never crashes the loop."""
        from swival.agent import run_agent_loop

        call_count = 0

        def fake_call_llm(*args, **kwargs):
            nonlocal call_count
            call_count += 1
            raise ContextOverflowError(f"too long (call {call_count})")

        messages = [_sys("system prompt"), _user("hello")]
        with patch("swival.agent.call_llm", side_effect=fake_call_llm):
            answer, exhausted = run_agent_loop(
                messages,
                _DUMMY_TOOLS,
                **self._loop_kwargs(tmp_path),
            )

        assert answer == _CONTEXT_EXHAUSTED_FALLBACK
        assert exhausted is True
        # Continue state preserved underneath the synthetic fallback.
        assert (tmp_path / ".swival" / "continue.md").exists()
        assert messages[-1].get("_swival_synthetic") is True
        # Bounded: compact_to_budget collapses the whole ladder into one commit,
        # so the initial call plus a couple of progress-checked retries is all
        # the reactive loop takes; the terminal floor then adds one attempt per
        # fixed budget before the loop gives up. No round-trip storm.
        assert call_count <= 4 + len(_TERMINAL_FLOOR_BUDGETS)


# ---------------------------------------------------------------------------
# summarize_turns
# ---------------------------------------------------------------------------


class TestSummarizeTurns:
    def _make_mock_call_llm(self, content="Summary of dropped turns."):
        """Return a mock call_llm that returns a successful response."""

        def mock_fn(
            *,
            base_url,
            model_id,
            messages,
            max_output_tokens,
            temperature,
            top_p,
            seed,
            tools,
            verbose,
            api_key,
            provider,
            **kwargs,
        ):
            return (
                SimpleNamespace(content=content),
                "stop",
            )

        return mock_fn

    def _make_failing_call_llm(self, exc):
        """Return a mock call_llm that raises the given exception."""

        def mock_fn(
            *,
            base_url,
            model_id,
            messages,
            max_output_tokens,
            temperature,
            top_p,
            seed,
            tools,
            verbose,
            api_key,
            provider,
            **kwargs,
        ):
            raise exc

        return mock_fn

    def _sample_turns(self):
        """Build a list of turns to summarize."""
        tc1 = _assistant_tc([("tc1", "read_file", '{"file_path": "foo.py"}')])
        tr1 = _tool("tc1", "def foo(): pass")
        tc2 = _assistant_tc([("tc2", "grep", '{"pattern": "TODO"}')])
        tr2 = _tool("tc2", "line 1: TODO fix\nline 2: TODO refactor")
        return [[tc1, tr1], [tc2, tr2]]

    def test_successful_summary(self):
        turns = self._sample_turns()
        result = summarize_turns(
            turns,
            self._make_mock_call_llm("The agent read foo.py and searched for TODOs."),
            model_id="test",
            base_url="http://localhost",
            api_key="key",
            top_p=None,
            seed=None,
            provider="lmstudio",
        )
        assert result == "The agent read foo.py and searched for TODOs."

    def test_returns_none_on_context_overflow(self):
        turns = self._sample_turns()
        result = summarize_turns(
            turns,
            self._make_failing_call_llm(ContextOverflowError("overflow")),
            model_id="test",
            base_url="http://localhost",
            api_key="key",
            top_p=None,
            seed=None,
            provider="lmstudio",
        )
        assert result is None

    def test_returns_none_on_timeout(self):
        turns = self._sample_turns()
        result = summarize_turns(
            turns,
            self._make_failing_call_llm(TimeoutError("timed out")),
            model_id="test",
            base_url="http://localhost",
            api_key="key",
            top_p=None,
            seed=None,
            provider="lmstudio",
        )
        assert result is None

    def test_returns_none_on_connection_error(self):
        turns = self._sample_turns()
        result = summarize_turns(
            turns,
            self._make_failing_call_llm(ConnectionError("refused")),
            model_id="test",
            base_url="http://localhost",
            api_key="key",
            top_p=None,
            seed=None,
            provider="lmstudio",
        )
        assert result is None

    def test_returns_none_on_empty_content(self):
        turns = self._sample_turns()
        result = summarize_turns(
            turns,
            self._make_mock_call_llm(""),
            model_id="test",
            base_url="http://localhost",
            api_key="key",
            top_p=None,
            seed=None,
            provider="lmstudio",
        )
        assert result is None

    def test_input_capped_at_8000_chars(self):
        """Large inputs should be truncated before being sent to the model."""
        big_turn = [_assistant("x" * 10000)]
        captured = {}

        def capturing_call_llm(
            *,
            base_url,
            model_id,
            messages,
            max_output_tokens,
            temperature,
            top_p,
            seed,
            tools,
            verbose,
            api_key,
            provider,
            **kwargs,
        ):
            captured["messages"] = messages
            return (SimpleNamespace(content="ok"), "stop")

        summarize_turns(
            [big_turn],
            capturing_call_llm,
            model_id="test",
            base_url="http://localhost",
            api_key="key",
            top_p=None,
            seed=None,
            provider="lmstudio",
        )
        user_content = captured["messages"][1]["content"]
        assert len(user_content) <= 8100  # 8000 + truncation marker

    def test_passes_tools_none(self):
        """summarize_turns should call LLM with tools=None."""
        captured = {}

        def capturing_call_llm(
            *,
            base_url,
            model_id,
            messages,
            max_output_tokens,
            temperature,
            top_p,
            seed,
            tools,
            verbose,
            api_key,
            provider,
            **kwargs,
        ):
            captured["tools"] = tools
            return (SimpleNamespace(content="ok"), "stop")

        summarize_turns(
            self._sample_turns(),
            capturing_call_llm,
            model_id="test",
            base_url="http://localhost",
            api_key="key",
            top_p=None,
            seed=None,
            provider="lmstudio",
        )
        assert captured["tools"] is None


class TestDropMiddleTurnsWithSummary:
    """Tests for AI-powered summarization in drop_middle_turns."""

    def _make_mock_call_llm(self, content="Summary recap."):
        def mock_fn(
            *,
            base_url,
            model_id,
            messages,
            max_output_tokens,
            temperature,
            top_p,
            seed,
            tools,
            verbose,
            api_key,
            provider,
            **kwargs,
        ):
            return (SimpleNamespace(content=content), "stop")

        return mock_fn

    def _make_failing_call_llm(self):
        def mock_fn(
            *,
            base_url,
            model_id,
            messages,
            max_output_tokens,
            temperature,
            top_p,
            seed,
            tools,
            verbose,
            api_key,
            provider,
            **kwargs,
        ):
            raise RuntimeError("LLM unavailable")

        return mock_fn

    def _build_msgs(self):
        """Build a message list with enough turns for dropping."""
        tc1 = _assistant_tc([("tc1", "read_file", "{}")])
        tr1 = _tool("tc1", "r1")
        tc2 = _assistant_tc([("tc2", "read_file", "{}")])
        tr2 = _tool("tc2", "r2")
        tc3 = _assistant_tc([("tc3", "read_file", "{}")])
        tr3 = _tool("tc3", "r3")
        tc4 = _assistant_tc([("tc4", "read_file", "{}")])
        tr4 = _tool("tc4", "r4")
        return [_sys("sys"), _user("q"), tc1, tr1, tc2, tr2, tc3, tr3, tc4, tr4]

    def test_summary_injected_as_assistant_role(self):
        msgs = self._build_msgs()
        result = drop_middle_turns(
            msgs,
            call_llm_fn=self._make_mock_call_llm("Recap of work done."),
            model_id="m",
            base_url="http://x",
            api_key="k",
            top_p=None,
            seed=None,
            provider="lmstudio",
        )
        # Find the recap message
        recaps = [
            m
            for m in result
            if isinstance(m, dict)
            and m.get("role") == "assistant"
            and isinstance(m.get("content"), str)
            and _RECAP_PREFIX in m["content"]
        ]
        assert len(recaps) == 1
        assert recaps[0]["role"] == "assistant"
        assert recaps[0]["content"].startswith(_RECAP_PREFIX)
        assert "Recap of work done." in recaps[0]["content"]

    def test_fallback_to_static_marker_on_failure(self):
        msgs = self._build_msgs()
        result = drop_middle_turns(
            msgs,
            call_llm_fn=self._make_failing_call_llm(),
            model_id="m",
            base_url="http://x",
            api_key="k",
            top_p=None,
            seed=None,
            provider="lmstudio",
        )
        # Should have the static splice marker (role=user)
        markers = [
            m
            for m in result
            if isinstance(m, dict) and "[context compacted" in m.get("content", "")
        ]
        assert len(markers) == 1
        assert markers[0]["role"] == "user"

    def test_fallback_without_llm_params(self):
        """When no call_llm_fn is provided, uses static marker."""
        msgs = self._build_msgs()
        result = drop_middle_turns(msgs)
        markers = [
            m
            for m in result
            if isinstance(m, dict) and "[context compacted" in m.get("content", "")
        ]
        assert len(markers) == 1
        assert markers[0]["role"] == "user"

    def test_no_user_or_system_role_from_summarization(self):
        """The recap must never use role=user or role=system."""
        msgs = self._build_msgs()
        result = drop_middle_turns(
            msgs,
            call_llm_fn=self._make_mock_call_llm("Recap text."),
            model_id="m",
            base_url="http://x",
            api_key="k",
            top_p=None,
            seed=None,
            provider="lmstudio",
        )
        for m in result:
            if not isinstance(m, dict):
                continue
            content = m.get("content", "")
            if isinstance(content, str) and _RECAP_PREFIX in content:
                assert m["role"] == "assistant"


# ---------------------------------------------------------------------------
# CompactionState
# ---------------------------------------------------------------------------


class TestCompactionState:
    def _make_mock_call_llm(self, content="Checkpoint summary."):
        def mock_fn(
            *,
            base_url,
            model_id,
            messages,
            max_output_tokens,
            temperature,
            top_p,
            seed,
            tools,
            verbose,
            api_key,
            provider,
            **kwargs,
        ):
            return (SimpleNamespace(content=content), "stop")

        return mock_fn

    def _make_failing_call_llm(self):
        def mock_fn(
            *,
            base_url,
            model_id,
            messages,
            max_output_tokens,
            temperature,
            top_p,
            seed,
            tools,
            verbose,
            api_key,
            provider,
            **kwargs,
        ):
            raise RuntimeError("LLM unavailable")

        return mock_fn

    def _llm_kwargs(self):
        return dict(
            model_id="test",
            base_url="http://localhost",
            api_key="key",
            top_p=None,
            seed=None,
            provider="lmstudio",
        )

    def _build_messages(self, n_turns):
        """Build a message list with n tool-call turns."""
        msgs = [_sys("system prompt"), _user("initial question")]
        for i in range(n_turns):
            tc = _assistant_tc([(f"tc{i}", "read_file", "{}")])
            tr = _tool(f"tc{i}", f"result {i}")
            msgs.extend([tc, tr])
        return msgs

    def test_checkpoint_fires_at_interval(self):
        state = CompactionState(checkpoint_interval=3)
        msgs = self._build_messages(5)
        mock_llm = self._make_mock_call_llm("checkpoint 1")
        for _ in range(3):
            state.maybe_checkpoint(msgs, mock_llm, **self._llm_kwargs())
        assert len(state.summaries) == 1
        assert state.summaries[0] == "checkpoint 1"

    def test_checkpoint_does_not_fire_early(self):
        state = CompactionState(checkpoint_interval=5)
        msgs = self._build_messages(3)
        mock_llm = self._make_mock_call_llm()
        for _ in range(4):
            state.maybe_checkpoint(msgs, mock_llm, **self._llm_kwargs())
        assert len(state.summaries) == 0

    def test_checkpoint_forwards_provider_kwargs(self):
        """Provider auth extras (geap project/location, bedrock profile) must
        reach the underlying call_llm or credential resolution fails."""
        state = CompactionState(checkpoint_interval=1)
        msgs = self._build_messages(2)
        seen = {}

        def capture_llm(**kwargs):
            seen.update(kwargs)
            return (SimpleNamespace(content="summary"), "stop")

        state.maybe_checkpoint(
            msgs,
            capture_llm,
            **self._llm_kwargs(),
            provider_kwargs={
                "vertex_project": "proj-1",
                "vertex_location": "global",
            },
        )
        assert seen["vertex_project"] == "proj-1"
        assert seen["vertex_location"] == "global"

    def test_counter_resets_on_failure(self):
        """After a failed checkpoint, the counter resets and doesn't retry every turn."""
        state = CompactionState(checkpoint_interval=3)
        msgs = self._build_messages(5)
        call_count = 0

        def counting_fail(
            *,
            base_url,
            model_id,
            messages,
            max_output_tokens,
            temperature,
            top_p,
            seed,
            tools,
            verbose,
            api_key,
            provider,
            **kwargs,
        ):
            nonlocal call_count
            call_count += 1
            raise RuntimeError("fail")

        # Trigger the checkpoint (3 turns)
        for _ in range(3):
            state.maybe_checkpoint(msgs, counting_fail, **self._llm_kwargs())
        assert call_count == 1  # called once at turn 3

        # Next turn should NOT retry
        state.maybe_checkpoint(msgs, counting_fail, **self._llm_kwargs())
        assert call_count == 1  # still 1, not retried

        # Need another full interval before next attempt
        for _ in range(2):
            state.maybe_checkpoint(msgs, counting_fail, **self._llm_kwargs())
        assert call_count == 2  # now attempted again at turn 6

    def test_summaries_never_exceed_max_checkpoints(self):
        """Even with many checkpoints, the list stays bounded."""
        state = CompactionState(checkpoint_interval=1)
        msgs = self._build_messages(5)
        mock_llm = self._make_mock_call_llm("summary")

        # Generate more checkpoints than MAX_CHECKPOINTS
        for i in range(MAX_CHECKPOINTS * 4):
            state.maybe_checkpoint(msgs, mock_llm, **self._llm_kwargs())

        assert len(state.summaries) <= MAX_CHECKPOINTS

    def test_get_full_summary_capped(self):
        state = CompactionState()
        # Manually stuff in large summaries
        state.summaries = ["x" * 5000 for _ in range(10)]
        full = state.get_full_summary()
        cap = MAX_CHECKPOINT_TOKENS * 4
        assert len(full) <= cap + 100  # small margin for truncation marker

    def test_consolidation_failure_drops_oldest(self):
        """When merge fails, oldest summaries are dropped."""
        state = CompactionState(checkpoint_interval=1)
        msgs = self._build_messages(5)
        call_counter = {"n": 0}

        def sometimes_fail(
            *,
            base_url,
            model_id,
            messages,
            max_output_tokens,
            temperature,
            top_p,
            seed,
            tools,
            verbose,
            api_key,
            provider,
            **kwargs,
        ):
            call_counter["n"] += 1
            # First MAX_CHECKPOINTS+1 calls succeed (building up summaries)
            # Then consolidation call fails
            if call_counter["n"] <= MAX_CHECKPOINTS + 1:
                return (SimpleNamespace(content=f"summary {call_counter['n']}"), "stop")
            raise RuntimeError("consolidation failed")

        for _ in range(MAX_CHECKPOINTS + 2):
            state.maybe_checkpoint(msgs, sometimes_fail, **self._llm_kwargs())

        # After consolidation failure, oldest are dropped
        assert len(state.summaries) <= MAX_CHECKPOINTS

    def test_checkpoint_fallback_in_drop_middle_turns(self):
        """When LLM summary fails but checkpoints exist, use checkpoint summary."""
        state = CompactionState()
        state.summaries = ["Earlier: agent read foo.py and found 3 bugs."]

        tc1 = _assistant_tc([("tc1", "read_file", "{}")])
        tr1 = _tool("tc1", "r1")
        tc2 = _assistant_tc([("tc2", "read_file", "{}")])
        tr2 = _tool("tc2", "r2")
        tc3 = _assistant_tc([("tc3", "read_file", "{}")])
        tr3 = _tool("tc3", "r3")
        tc4 = _assistant_tc([("tc4", "read_file", "{}")])
        tr4 = _tool("tc4", "r4")
        msgs = [_sys("sys"), _user("q"), tc1, tr1, tc2, tr2, tc3, tr3, tc4, tr4]

        def failing_llm(
            *,
            base_url,
            model_id,
            messages,
            max_output_tokens,
            temperature,
            top_p,
            seed,
            tools,
            verbose,
            api_key,
            provider,
            **kwargs,
        ):
            raise RuntimeError("LLM down")

        result = drop_middle_turns(
            msgs,
            call_llm_fn=failing_llm,
            model_id="m",
            base_url="http://x",
            api_key="k",
            top_p=None,
            seed=None,
            provider="lmstudio",
            compaction_state=state,
        )
        # Should have used checkpoint summary, not static marker
        recaps = [
            m
            for m in result
            if isinstance(m, dict)
            and m.get("role") == "assistant"
            and "checkpoints" in m.get("content", "")
        ]
        assert len(recaps) == 1
        assert "3 bugs" in recaps[0]["content"]


# ---------------------------------------------------------------------------
# aggressive_drop_turns
# ---------------------------------------------------------------------------


class TestAggressiveDropTurns:
    def _build_msgs(self, n_middle_turns=5):
        """Build messages with system + user + N middle turns + nothing."""
        msgs = [_sys("system prompt"), _user("initial question")]
        for i in range(n_middle_turns):
            tc = _assistant_tc([(f"tc{i}", "read_file", "{}")])
            tr = _tool(f"tc{i}", f"result {i}")
            msgs.extend([tc, tr])
        return msgs

    def test_keeps_system_and_last_2_turns(self):
        msgs = self._build_msgs(6)
        result = aggressive_drop_turns(msgs)
        # Should have: system, recap/marker, last 2 turns (4 messages)
        roles = [
            m.get("role") if isinstance(m, dict) else getattr(m, "role", None)
            for m in result
        ]
        assert roles[0] == "system"
        # Second message should be the recap or splice marker
        assert result[1].get("role") in ("assistant", "user")

    def test_fewer_turns_than_tail_unchanged(self):
        msgs = [_sys("s"), _assistant("done")]
        result = aggressive_drop_turns(msgs)
        assert len(result) == len(msgs)

    def test_with_llm_summary(self):
        msgs = self._build_msgs(5)

        def mock_llm(
            *,
            base_url,
            model_id,
            messages,
            max_output_tokens,
            temperature,
            top_p,
            seed,
            tools,
            verbose,
            api_key,
            provider,
            **kwargs,
        ):
            return (SimpleNamespace(content="Aggressive recap."), "stop")

        result = aggressive_drop_turns(
            msgs,
            call_llm_fn=mock_llm,
            model_id="m",
            base_url="http://x",
            api_key="k",
            top_p=None,
            seed=None,
            provider="lmstudio",
        )
        recaps = [
            m
            for m in result
            if isinstance(m, dict)
            and m.get("role") == "assistant"
            and _RECAP_PREFIX in m.get("content", "")
        ]
        assert len(recaps) == 1
        assert "Aggressive recap." in recaps[0]["content"]

    def test_fallback_to_static_marker(self):
        msgs = self._build_msgs(5)
        result = aggressive_drop_turns(msgs)
        markers = [
            m
            for m in result
            if isinstance(m, dict) and "[context compacted" in m.get("content", "")
        ]
        assert len(markers) == 1

    def test_more_aggressive_than_drop_middle(self):
        """aggressive_drop_turns should produce fewer messages than drop_middle_turns."""
        msgs = self._build_msgs(8)
        drop_result = drop_middle_turns(list(msgs))
        aggressive_result = aggressive_drop_turns(list(msgs))
        assert len(aggressive_result) <= len(drop_result)

    def test_valid_tool_pairing_after_aggressive(self):
        msgs = self._build_msgs(6)
        result = aggressive_drop_turns(msgs)
        _validate_tool_pairing(result)


# ---------------------------------------------------------------------------
# Graduated compaction (integration)
# ---------------------------------------------------------------------------


class TestGraduatedCompaction:
    def test_compaction_levels_produce_valid_messages(self):
        """Each compaction level produces a valid message list."""
        msgs = [_sys("s"), _user("q")]
        for i in range(10):
            tc = _assistant_tc([(f"tc{i}", "read_file", '{"file_path": "f.py"}')])
            tr = _tool(f"tc{i}", "x" * 2000)
            msgs.extend([tc, tr])
        msgs.append(_assistant("done"))

        # Level 1
        r1 = compact_messages(list(msgs))
        _validate_tool_pairing(r1)

        # Level 2
        r2 = drop_middle_turns(list(msgs))
        _validate_tool_pairing(r2)

        # Level 3
        r3 = aggressive_drop_turns(list(msgs))
        _validate_tool_pairing(r3)

    def test_each_level_reduces_size(self):
        """Each graduated level should produce fewer or equal tokens."""
        msgs = [_sys("s"), _user("q")]
        for i in range(10):
            tc = _assistant_tc([(f"tc{i}", "read_file", '{"file_path": "f.py"}')])
            tr = _tool(f"tc{i}", "x" * 2000)
            msgs.extend([tc, tr])
        msgs.append(_assistant("done"))

        t_original = estimate_tokens(msgs)
        t_compact = estimate_tokens(compact_messages(list(msgs)))
        t_drop = estimate_tokens(drop_middle_turns(list(msgs)))
        t_aggressive = estimate_tokens(aggressive_drop_turns(list(msgs)))

        assert t_compact <= t_original
        assert t_drop <= t_compact
        assert t_aggressive <= t_drop

    def test_graduated_levels_all_preserve_system_prompt(self):
        """All compaction levels should preserve the system prompt."""
        msgs = [_sys("You are a helpful assistant."), _user("q")]
        for i in range(8):
            tc = _assistant_tc([(f"tc{i}", "read_file", "{}")])
            tr = _tool(f"tc{i}", "x" * 2000)
            msgs.extend([tc, tr])

        for compact_fn in [compact_messages, drop_middle_turns, aggressive_drop_turns]:
            result = compact_fn(list(msgs))
            first_role = (
                result[0].get("role")
                if isinstance(result[0], dict)
                else getattr(result[0], "role", None)
            )
            assert first_role == "system"


# ---------------------------------------------------------------------------
# _fix_orphaned_tool_calls
# ---------------------------------------------------------------------------


class TestFixOrphanedToolCalls:
    def test_removes_orphaned_tool_calls(self):
        tc = _assistant_tc([("tc1", "read_file", "{}"), ("tc2", "grep", "{}")])
        tr1 = _tool("tc1", "content1")
        # tc2 result is missing
        msgs = [_user("q"), tc, tr1, _assistant("done")]
        assert _fix_orphaned_tool_calls(msgs) is True
        # tc should now only have tc1
        remaining = tc.tool_calls
        assert len(remaining) == 1
        assert remaining[0].id == "tc1"

    def test_removes_all_tool_calls_sets_content(self):
        tc = _assistant_tc([("tc1", "read_file", "{}")])
        # No tool result at all
        msgs = [_user("q"), tc, _assistant("done")]
        assert _fix_orphaned_tool_calls(msgs) is True
        assert tc.tool_calls is None
        assert tc.content == ""

    def test_noop_when_all_results_present(self):
        tc = _assistant_tc([("tc1", "read_file", "{}")])
        tr = _tool("tc1", "ok")
        msgs = [_user("q"), tc, tr, _assistant("done")]
        assert _fix_orphaned_tool_calls(msgs) is False

    def test_dict_messages(self):
        tc = {
            "role": "assistant",
            "content": None,
            "tool_calls": [{"id": "tc1", "function": {"name": "f", "arguments": "{}"}}],
        }
        msgs = [_user("q"), tc, _assistant("done")]
        assert _fix_orphaned_tool_calls(msgs) is True
        assert "tool_calls" not in tc
        assert tc["content"] == ""

    def test_preserves_content_when_tool_calls_removed(self):
        tc = SimpleNamespace(
            role="assistant",
            content="I'll read the file",
            tool_calls=[
                SimpleNamespace(
                    id="tc1", function=SimpleNamespace(name="f", arguments="{}")
                )
            ],
        )
        msgs = [_user("q"), tc, _assistant("done")]
        assert _fix_orphaned_tool_calls(msgs) is True
        assert tc.tool_calls is None
        assert tc.content == "I'll read the file"


# ---------------------------------------------------------------------------
# _emergency_truncate
# ---------------------------------------------------------------------------


class TestEmergencyTruncate:
    """Tests for the last-resort _emergency_truncate compaction."""

    def test_fits_already(self):
        """No-op when messages already fit within context_length."""
        msgs = [_sys("system"), _user("hello")]
        original = [m.copy() for m in msgs]
        _emergency_truncate(msgs, 10_000)
        assert msgs[0]["content"] == original[0]["content"]
        assert msgs[1]["content"] == original[1]["content"]

    def test_compacts_tool_results(self):
        """Stage 1: tool results in tail turns are compacted."""
        big_result = "x" * 5000
        msgs = [
            _sys("sys"),
            _assistant_tc([("tc1", "read_file", '{"file_path": "a.py"}')]),
            _tool("tc1", big_result),
            _user("thanks"),
        ]
        # Use a small context_length so the big tool result must be compacted
        _emergency_truncate(msgs, 2000)
        tool_content = (
            msgs[2]["content"] if isinstance(msgs[2], dict) else msgs[2].content
        )
        assert len(tool_content) < len(big_result)

    def test_truncates_large_messages(self):
        """Stage 2: large non-system messages get progressively truncated."""
        msgs = [
            _sys("short system"),
            _user("q"),
            _assistant("a" * 20_000),
        ]
        _emergency_truncate(msgs, 2000)
        result_content = msgs[2]["content"]
        assert len(result_content) < 20_000
        assert "truncated" in result_content.lower()

    def test_nuclear_keeps_system_and_last_user(self):
        """Stage 3: when truncation isn't enough, keep only system + last user."""
        msgs = [
            _sys("s"),
            _user("first question"),
            _assistant("a" * 10_000),
            _user("b" * 10_000),
            _assistant("c" * 10_000),
            _user("last question"),
        ]
        _emergency_truncate(msgs, 200)
        roles = [m["role"] if isinstance(m, dict) else m.role for m in msgs]
        assert roles[0] == "system"
        assert any(r == "user" for r in roles)
        assert len(msgs) == 2  # system + last user
        # clamp_output_tokens must not raise after emergency truncation
        clamp_output_tokens(msgs, None, 200, 200)

    def test_returns_messages(self):
        """Return value is the same list that was passed in."""
        msgs = [_sys("s"), _user("u")]
        result = _emergency_truncate(msgs, 10_000)
        assert result is msgs

    def test_system_prompt_preserved_when_possible(self):
        """System prompt is not truncated unless absolutely necessary."""
        sys_content = "important system rules"
        msgs = [
            _sys(sys_content),
            _user("q"),
            _assistant("a" * 5000),
        ]
        _emergency_truncate(msgs, 2000)
        assert msgs[0]["content"] == sys_content

    def test_no_system_message(self):
        """Nuclear fallback works when transcript has no leading system message.

        Use enough messages that stage-2 truncation to 200 chars each still
        exceeds the tiny context window, forcing the nuclear path.
        """
        msgs = [_user("q%d" % i) for i in range(80)] + [_user("last question")]
        _emergency_truncate(msgs, 50)
        # Only the last user message should survive
        assert len(msgs) == 1
        assert msgs[0]["role"] == "user"
        clamp_output_tokens(msgs, None, 50, 50)

    def test_tiny_context_clamp_succeeds(self):
        """After emergency truncation with a tiny context, clamp_output_tokens must not raise."""
        msgs = [
            _sys("system " * 500),
            _user("user " * 500),
            _assistant("assistant " * 500),
        ]
        _emergency_truncate(msgs, 100)
        # This is the actual contract: clamp must succeed
        clamp_output_tokens(msgs, None, 100, 100)


# ---------------------------------------------------------------------------
# Prompt budget (safety_ratio / reserve_output)
# ---------------------------------------------------------------------------


def _synthetic(content):
    return {"role": "user", "content": content, "_swival_synthetic": True}


class TestPromptBudget:
    def test_unknown_window_returns_none(self):
        assert _prompt_budget(None, 1024) is None

    def test_reserves_output_room(self):
        # budget = floor(context * 0.90) - reserve(output)
        assert _prompt_budget(10_000, 1000) == 9000 - 1000

    def test_reserve_clamped_to_half_context(self):
        # A huge max_output cannot starve the prompt: reserve caps at ctx // 2.
        budget = _prompt_budget(10_000, 999_999)
        assert budget == 9000 - 5000

    def test_reserve_floor(self):
        # No requested output budget still reserves the minimum.
        assert _prompt_budget(10_000, None) == 9000 - MIN_OUTPUT_TOKENS

    def test_custom_safety_ratio(self):
        assert _prompt_budget(10_000, 1000, safety_ratio=0.5) == 5000 - 1000

    def test_never_negative(self):
        # A tiny window can't produce a negative budget.
        assert _prompt_budget(20, 1000) >= MIN_OUTPUT_TOKENS


# ---------------------------------------------------------------------------
# Synthetic scaffolding GC
# ---------------------------------------------------------------------------


class TestGcScaffolding:
    def test_removes_spent_nudge(self):
        msgs = [
            _sys("system"),
            _user("hello"),
            _synthetic("IMPORTANT: stop repeating that error"),
            _assistant("ok, fixed it"),
        ]
        assert _has_spent_scaffolding(msgs) is True
        assert _gc_scaffolding(msgs) is True
        contents = [m.get("content") for m in msgs]
        assert "IMPORTANT: stop repeating that error" not in contents
        # Real user turn and the assistant response survive.
        assert msgs[0]["role"] == "system"
        assert msgs[1]["content"] == "hello"
        assert msgs[-1]["content"] == "ok, fixed it"

    def test_preserves_active_nudge_without_later_assistant(self):
        # The most recent nudge has not been acted on yet — keep it.
        msgs = [
            _sys("system"),
            _user("hello"),
            _assistant("first answer"),
            _synthetic("Tip: consider using think"),
        ]
        assert _has_spent_scaffolding(msgs) is False
        assert _gc_scaffolding(msgs) is False
        assert msgs[-1]["content"] == "Tip: consider using think"

    def test_preserves_durable_goal_recap(self):
        # Goal recaps carry durable context, not loop scaffolding.
        msgs = [
            _sys("system"),
            _user("hello"),
            _synthetic(GOAL_RECAP_PREFIX + " objective: ship it"),
            _assistant("working on it"),
        ]
        assert _gc_scaffolding(msgs) is False
        assert any(GOAL_RECAP_PREFIX in (m.get("content") or "") for m in msgs)

    def test_non_synthetic_user_untouched(self):
        msgs = [
            _sys("system"),
            _user("IMPORTANT: this is a real user message"),
            _assistant("ack"),
        ]
        assert _gc_scaffolding(msgs) is False

    def test_gc_is_first_rung_in_compact_context(self):
        msgs = [
            _sys("system"),
            _user("hello"),
            _synthetic("Reminder: you have open todos"),
            _assistant("done"),
        ]
        result = compact_context(
            CompactionContext(
                messages=msgs,
                tools=None,
                context_length=1000,
                max_output_tokens=100,
            )
        )
        assert result.strategy == COMPACTION_GC_SCAFFOLDING
        assert result.history_mutated is True


# ---------------------------------------------------------------------------
# compact_to_budget: loop-to-budget, commit once
# ---------------------------------------------------------------------------


class TestCompactToBudget:
    def _big_transcript(self):
        big = "x" * 8000
        return [
            _sys("system prompt"),
            _user("first task"),
            _assistant_tc([("c1", "read_file", "{}")]),
            _tool("c1", big),
            _assistant("read it"),
            _user("second task"),
            _assistant_tc([("c2", "read_file", "{}")]),
            _tool("c2", big),
            _assistant("read it too"),
            _user("third task"),
            _assistant("almost done"),
            _user("fourth task"),
        ]

    def test_noop_when_within_budget(self):
        msgs = [_sys("system"), _user("hello")]
        before = estimate_tokens(msgs, None)
        result = compact_to_budget(
            msgs,
            None,
            budget=10_000,
            context_length=20_000,
            max_output_tokens=1024,
        )
        assert result.strategy == "noop"
        assert result.history_mutated is False
        assert result.tokens_after == before

    def test_unknown_window_is_noop(self):
        msgs = self._big_transcript()
        before_len = len(msgs)
        before = estimate_tokens(msgs, None)
        result = compact_to_budget(
            msgs,
            None,
            budget=None,
            context_length=None,
            max_output_tokens=1024,
        )
        assert result.strategy == "noop"
        assert len(msgs) == before_len
        assert estimate_tokens(msgs, None) == before

    def test_compacts_under_budget_and_commits_in_place(self):
        msgs = self._big_transcript()
        original_id = id(msgs)
        before = estimate_tokens(msgs, None)
        budget = 300
        result = compact_to_budget(
            msgs,
            None,
            budget=budget,
            context_length=4000,
            max_output_tokens=256,
        )
        assert result.tokens_after < before
        assert result.tokens_after <= budget
        # Commit once: the same list object is mutated in place.
        assert id(msgs) == original_id
        assert result.messages is msgs
        assert estimate_tokens(msgs, None) == result.tokens_after

    def test_drops_tools_when_messages_cannot_fit(self):
        # A tiny budget on a minimal transcript forces the tool schemas off.
        msgs = [_sys("system"), _user("hello")]
        tools = _DUMMY_TOOLS
        result = compact_to_budget(
            msgs,
            tools,
            budget=MIN_OUTPUT_TOKENS,
            context_length=4000,
            max_output_tokens=256,
        )
        assert result.dropped_tools is True
        assert result.tools is None

    def test_proactive_flags_skip_emergency_and_tool_drop(self):
        # Proactive mode must not truncate or strip tools — only the safe rungs.
        msgs = [_sys("system"), _user("hello")]
        tools = _DUMMY_TOOLS
        result = compact_to_budget(
            msgs,
            tools,
            budget=MIN_OUTPUT_TOKENS,
            context_length=4000,
            max_output_tokens=256,
            allow_tool_drop=False,
            allow_emergency=False,
        )
        assert result.dropped_tools is False
        assert result.tools is tools

    def test_proactive_does_not_drop_conversation_turns(self):
        # With allow_turn_drop=False the destructive turn-dropping rungs must
        # never fire, even when the budget can't otherwise be met.
        msgs = self._big_transcript()

        def _user_turns(ms):
            return [m for m in ms if isinstance(m, dict) and m.get("role") == "user"]

        before_users = _user_turns(msgs)
        result = compact_to_budget(
            msgs,
            None,
            budget=MIN_OUTPUT_TOKENS,
            context_length=4000,
            max_output_tokens=256,
            allow_tool_drop=False,
            allow_emergency=False,
            allow_turn_drop=False,
        )
        assert COMPACTION_DROP_MIDDLE not in result.strategy
        assert COMPACTION_AGGRESSIVE not in result.strategy
        # Every real user turn still present — no history was dropped.
        assert len(_user_turns(msgs)) == len(before_users)
        # It couldn't reach the tiny budget with only the safe rungs.
        assert result.met_budget is False

    def test_met_budget_flag_reflects_outcome(self):
        # Best-effort: an unreachable budget still commits real reductions but
        # reports met_budget=False; a reachable one reports True.
        msgs = self._big_transcript()
        before = estimate_tokens(msgs, None)
        impossible = compact_to_budget(
            msgs,
            None,
            budget=1,
            context_length=4000,
            max_output_tokens=256,
        )
        assert impossible.met_budget is False
        # Partial progress is still committed (leaner is better than not).
        assert impossible.tokens_after < before

        msgs2 = self._big_transcript()
        ok = compact_to_budget(
            msgs2,
            None,
            budget=400,
            context_length=4000,
            max_output_tokens=256,
        )
        assert ok.met_budget is True
        assert ok.tokens_after <= 400


# ---------------------------------------------------------------------------
# Terminal floor ladder — the absolute backstop for unknown context windows
# ---------------------------------------------------------------------------


def _ok_msg(content="ok"):
    return (
        SimpleNamespace(content=content, tool_calls=None, role="assistant"),
        "stop",
        [],
        0,
        (0, 0),
    )


class TestTerminalFloorLadder:
    """Unit tests for _run_terminal_floor_ladder (pure helper)."""

    @staticmethod
    def _common(**overrides):
        base = dict(
            api_base="http://127.0.0.1:1234",
            model_id="test-model",
            temperature=0.5,
            top_p=None,
            seed=None,
            verbose=False,
            llm_kwargs={},
        )
        base.update(overrides)
        return base

    @staticmethod
    def _big(n=40, chars=4000):
        msgs = [_sys("you are a helpful assistant")]
        for i in range(n):
            msgs.append(_user(f"question {i} " + "x" * chars))
            msgs.append(_assistant(f"answer {i} " + "y" * chars))
        msgs.append(_user("final question " + "z" * chars))
        return msgs

    def test_accepts_first_budget(self):
        msgs = self._big()
        with patch("swival.agent.call_llm", side_effect=lambda *a, **k: _ok_msg()):
            res = _run_terminal_floor_ladder(msgs, **self._common())
        assert isinstance(res, TerminalAttemptResult)
        assert res.budget == _TERMINAL_FLOOR_BUDGETS[0]
        assert res.msg.content == "ok"
        assert res.finish_reason == "stop"
        assert res.tokens_after <= _TERMINAL_FLOOR_BUDGETS[0]

    def test_descends_to_smallest_budget(self):
        msgs = self._big()
        seq = [
            ContextOverflowError("too big"),
            ContextOverflowError("still too big"),
            _ok_msg("recovered"),
        ]

        def picky(*args, **kwargs):
            item = seq.pop(0)
            if isinstance(item, Exception):
                raise item
            return item

        with patch("swival.agent.call_llm", side_effect=picky):
            res = _run_terminal_floor_ladder(msgs, **self._common())
        assert res is not None
        assert res.budget == _TERMINAL_FLOOR_BUDGETS[-1]
        assert res.msg.content == "recovered"

    def test_truncated_tool_call_response_tries_next_budget(self):
        from swival.report import ReportCollector

        msgs = self._big()
        report = ReportCollector()
        calls = 0

        def truncated_then_clean(*args, **kwargs):
            nonlocal calls
            calls += 1
            if calls == 1:
                bad_call = SimpleNamespace(
                    id="tc1",
                    type="function",
                    function=SimpleNamespace(
                        name="read_file",
                        arguments="not-json",
                    ),
                )
                return (
                    SimpleNamespace(
                        content=None,
                        tool_calls=[bad_call],
                        role="assistant",
                    ),
                    "length",
                    [],
                    0,
                    (0, 0),
                )
            return _ok_msg("recovered")

        with patch("swival.agent.call_llm", side_effect=truncated_then_clean):
            res = _run_terminal_floor_ladder(
                msgs,
                budgets=(4096, 1024),
                tools=_DUMMY_TOOLS,
                turn=7,
                report=report,
                **self._common(),
            )

        assert res is not None
        assert res.budget == 1024
        assert res.msg.content == "recovered"
        assert calls == 2
        assert report.recovered_responses == 1

    def test_all_budgets_rejected_returns_none(self):
        msgs = self._big()
        before = copy.deepcopy(msgs)

        def always_overflow(*args, **kwargs):
            raise ContextOverflowError("nope")

        with patch("swival.agent.call_llm", side_effect=always_overflow):
            res = _run_terminal_floor_ladder(msgs, **self._common())
        assert res is None
        assert msgs == before  # pristine on the failure path

    def test_does_not_mutate_input_on_success(self):
        msgs = self._big()
        before = copy.deepcopy(msgs)
        with patch("swival.agent.call_llm", side_effect=lambda *a, **k: _ok_msg()):
            _run_terminal_floor_ladder(msgs, **self._common())
        assert msgs == before

    def test_output_pinned_and_tools_dropped(self):
        msgs = self._big()
        captured = {}

        def capture(*args, **kwargs):
            captured["max_output"] = args[3]
            captured["tools"] = args[7]
            return _ok_msg()

        with patch("swival.agent.call_llm", side_effect=capture):
            _run_terminal_floor_ladder(msgs, **self._common())
        assert captured["max_output"] == MIN_OUTPUT_TOKENS
        assert captured["tools"] is None

    def test_non_overflow_error_propagates(self):
        msgs = self._big()

        def boom(*args, **kwargs):
            raise AgentError("auth failed")

        with patch("swival.agent.call_llm", side_effect=boom):
            with pytest.raises(AgentError, match="auth failed"):
                _run_terminal_floor_ladder(msgs, **self._common())

    def test_collapses_to_system_user_when_forced(self):
        msgs = self._big()
        captured = {}

        def capture(*args, **kwargs):
            captured["prompt"] = args[2]
            return _ok_msg()

        with patch("swival.agent.call_llm", side_effect=capture):
            _run_terminal_floor_ladder(msgs, budgets=(128,), **self._common())
        prompt = captured["prompt"]
        assert len(prompt) <= 2
        assert prompt[-1]["role"] == "user"

    def test_no_orphaned_tool_calls_after_truncation(self):
        msgs = [
            _sys("system " + "s" * 2000),
            _user("original task " + "u" * 2000),
            _assistant_tc([("tc1", "read_file", '{"file_path": "x"}')]),
            _tool("tc1", "file contents " + "c" * 8000),
            _assistant("done reading"),
            _user("now answer " + "q" * 2000),
        ]
        captured = {}

        def capture(*args, **kwargs):
            captured["prompt"] = args[2]
            return _ok_msg()

        with patch("swival.agent.call_llm", side_effect=capture):
            _run_terminal_floor_ladder(msgs, budgets=(256,), **self._common())
        # The collapsed prompt must be a valid request: no tool result left
        # without its originating tool call.
        assert _fix_orphaned_tool_calls(copy.deepcopy(captured["prompt"])) is False


def _noop_compact(messages, tools, *, budget, **kwargs):
    """A compact_to_budget that never shrinks — forces the reactive loop to
    give up so the terminal floor is reached."""
    est = estimate_tokens(messages, tools)
    return SimpleNamespace(
        messages=messages,
        tools=tools,
        strategy="noop",
        description="noop",
        tokens_before=est,
        tokens_after=est,
        history_mutated=False,
        dropped_tools=False,
        met_budget=False,
    )


class TestTerminalFloorLoop:
    """Integration: terminal floor recovery inside run_agent_loop."""

    @staticmethod
    def _loop_kwargs(tmp_path, **overrides):
        return _loop_kwargs(tmp_path, **overrides)

    def test_terminal_floor_recovers_and_records(self, tmp_path):
        from swival.agent import run_agent_loop
        from swival.report import ReportCollector

        def fake_call_llm(*args, **kwargs):
            prompt = args[2]
            if estimate_tokens(prompt, None) > 4090:
                raise ContextOverflowError("context length exceeded")
            return _ok_msg("recovered")

        report = ReportCollector()
        messages = [_sys("system"), _user("question " + "z" * 30000)]
        with (
            patch("swival.agent.compact_to_budget", side_effect=_noop_compact),
            patch("swival.agent.call_llm", side_effect=fake_call_llm),
        ):
            answer, exhausted = run_agent_loop(
                messages,
                _DUMMY_TOOLS,
                **self._loop_kwargs(tmp_path, report=report),
            )

        assert answer == "recovered"
        assert exhausted is False
        # History was committed to the truncated transcript plus the answer.
        assert len(messages) <= 3
        assert messages[-1]["role"] == "assistant"
        llm_events = [e for e in report.events if e["type"] == "llm_call"]
        assert any(e.get("retry_reason") == "terminal_floor" for e in llm_events)

    def test_terminal_floor_total_failure_returns_fallback(self, tmp_path):
        """The load-bearing no-crash regression: when even the minimal
        terminal-floor prompt is rejected, the loop must end the turn with the
        local fallback (exhausted, no raise), not crash."""
        from swival.agent import run_agent_loop
        from swival.report import ReportCollector

        calls = 0

        def always_overflow(*args, **kwargs):
            nonlocal calls
            calls += 1
            raise ContextOverflowError("context length exceeded")

        report = ReportCollector()
        messages = [_sys("system"), _user("question " + "z" * 30000)]
        before = copy.deepcopy(messages)
        with (
            patch("swival.agent.compact_to_budget", side_effect=_noop_compact),
            patch("swival.agent.call_llm", side_effect=always_overflow),
        ):
            answer, exhausted = run_agent_loop(
                messages,
                _DUMMY_TOOLS,
                **self._loop_kwargs(tmp_path, report=report),
            )

        # Degraded, not crashed: the fallback notice is returned as an exhausted
        # (non-success) turn.
        assert answer == _CONTEXT_EXHAUSTED_FALLBACK
        assert exhausted is True
        # Bounded provider calls — never an unbounded retry storm.
        assert 0 < calls <= 20
        # The synthetic fallback is appended; the real transcript is preserved
        # underneath it for /continue.
        assert messages[:-1] == before
        assert messages[-1]["role"] == "assistant"
        assert messages[-1].get("_swival_synthetic") is True
        # The event is legible downstream.
        recovered = [
            e
            for e in report.events
            if e["type"] == "recovered_response"
            and e.get("reason") == _CONTEXT_EXHAUSTED_REASON
        ]
        assert len(recovered) == 1

    def test_terminal_floor_total_failure_writes_continue_file(self, tmp_path):
        from swival.agent import run_agent_loop

        def always_overflow(*args, **kwargs):
            raise ContextOverflowError("context length exceeded")

        messages = [_sys("system"), _user("question " + "z" * 30000)]
        with (
            patch("swival.agent.compact_to_budget", side_effect=_noop_compact),
            patch("swival.agent.call_llm", side_effect=always_overflow),
        ):
            answer, exhausted = run_agent_loop(
                messages,
                _DUMMY_TOOLS,
                **self._loop_kwargs(tmp_path, continue_here=True),
            )
        assert exhausted is True
        assert answer == _CONTEXT_EXHAUSTED_FALLBACK
        # Continue state was preserved on the no-crash path.
        assert (tmp_path / ".swival" / "continue.md").exists()


# ---------------------------------------------------------------------------
# Phase 1: unknown-window overflow detection (_looks_like_context_overflow)
# ---------------------------------------------------------------------------


def _bad_request(message):
    import litellm

    return litellm.BadRequestError(message=message, model="m", llm_provider="openai")


def _api_error(message, status_code):
    import litellm

    return litellm.APIError(
        message=message, status_code=status_code, model="m", llm_provider="openai"
    )


class TestUnknownWindowOverflowDetection:
    """The loose tier engages only when the window is unknown; the strict tier
    always engages; unrelated bad requests stay unclassified in both regimes."""

    STRICT = [
        "maximum context length is 8192 tokens",
        "This model's maximum context length is 4096 tokens",
        "the context window was exceeded",
        "token limit reached",
    ]
    LOOSE = [
        "prompt is too long: 219763 tokens > 200000 maximum",
        "this input is too large",
        "please reduce the length of the messages",
        "string too long",
        "input is too long for this model",
        "request entity too large",
    ]
    UNRELATED = [
        "invalid temperature value",
        "unsupported parameter: foo",
        "you are not authorized",
    ]

    @pytest.mark.parametrize("msg", STRICT)
    def test_strict_classified_in_both_regimes(self, msg):
        assert _looks_like_context_overflow(_bad_request(msg)) is True
        assert (
            _looks_like_context_overflow(_bad_request(msg), unknown_context_window=True)
            is True
        )

    @pytest.mark.parametrize("msg", LOOSE)
    def test_loose_only_when_unknown(self, msg):
        # Known window: loose phrasing is NOT treated as overflow.
        assert _looks_like_context_overflow(_bad_request(msg)) is False
        # Unknown window: it is.
        assert (
            _looks_like_context_overflow(_bad_request(msg), unknown_context_window=True)
            is True
        )

    @pytest.mark.parametrize("msg", UNRELATED)
    def test_unrelated_never_classified(self, msg):
        assert _looks_like_context_overflow(_bad_request(msg)) is False
        assert (
            _looks_like_context_overflow(_bad_request(msg), unknown_context_window=True)
            is False
        )

    def test_413_only_when_unknown(self):
        e = _api_error("Payload Too Large", 413)
        assert _looks_like_context_overflow(e) is False
        assert _looks_like_context_overflow(e, unknown_context_window=True) is True

    def test_typed_exception_always(self):
        import litellm

        e = litellm.ContextWindowExceededError(
            message="whatever", model="m", llm_provider="openai"
        )
        assert _looks_like_context_overflow(e) is True

    def test_call_llm_loose_phrasing_unknown_window(self):
        """call_llm classifies a loose phrasing as overflow only when told the
        window is unknown."""
        with patch("litellm.completion") as mock_comp:
            mock_comp.side_effect = _bad_request("Error: input is too long")
            with pytest.raises(ContextOverflowError):
                call_llm(
                    "http://localhost",
                    "model",
                    [],
                    100,
                    0.1,
                    1.0,
                    None,
                    None,
                    False,
                    unknown_context_window=True,
                )

    def test_call_llm_loose_phrasing_known_window_is_request_shaped(self):
        """With a known window the same phrasing stays a plain AgentError, and
        is marked request-shaped for the (non-applicable) backstop."""
        with patch("litellm.completion") as mock_comp:
            mock_comp.side_effect = _bad_request("Error: input is too long")
            with pytest.raises(AgentError) as ei:
                call_llm(
                    "http://localhost",
                    "model",
                    [],
                    100,
                    0.1,
                    1.0,
                    None,
                    None,
                    False,
                    unknown_context_window=False,
                )
        assert not isinstance(ei.value, ContextOverflowError)
        assert getattr(ei.value, "_request_shaped", False) is True


# ---------------------------------------------------------------------------
# Phase 2: learning the window from the rejection (_parse_context_limit)
# ---------------------------------------------------------------------------


class TestParseContextLimit:
    def test_named_maximum(self):
        assert _parse_context_limit("maximum context length is 8192 tokens") == 8192

    def test_actual_gt_limit_takes_smaller(self):
        assert (
            _parse_context_limit("prompt is too long: 219763 tokens > 200000 maximum")
            == 200000
        )

    def test_context_window_of(self):
        assert _parse_context_limit("the context window of 32768 was exceeded") == 32768

    def test_max_input_tokens(self):
        assert _parse_context_limit("max input tokens: 4096 exceeded") == 4096

    def test_no_number_returns_none(self):
        assert _parse_context_limit("the input is too long") is None
        assert _parse_context_limit("") is None
        assert _parse_context_limit(None) is None

    def test_picks_limit_not_actual_when_both_present(self):
        text = (
            "This model's maximum context length is 8192 tokens. However, your "
            "messages resulted in 10000 tokens."
        )
        # The named-maximum pattern wins and yields the ceiling, not the actual.
        assert _parse_context_limit(text) == 8192


# ---------------------------------------------------------------------------
# Phase 3: adaptive budget (AdaptiveContextBudget)
# ---------------------------------------------------------------------------


class TestAdaptiveContextBudget:
    def test_idle_until_first_reject(self):
        ab = AdaptiveContextBudget()
        assert ab.target() is None
        ab.record_accept(50000)
        assert ab.target() is None  # no ceiling observed yet

    def test_backs_off_below_lone_rejection(self):
        ab = AdaptiveContextBudget()
        ab.record_reject(50000)
        assert ab.target() == int(50000 * REACTIVE_BUDGET_BACKOFF)

    def test_binary_search_between_accept_and_reject(self):
        ab = AdaptiveContextBudget()
        ab.record_accept(20000)
        ab.record_reject(50000)
        assert ab.target() == (20000 + 50000) // 2

    def test_reject_keeps_smallest(self):
        ab = AdaptiveContextBudget()
        ab.record_reject(60000)
        ab.record_reject(40000)
        assert ab.rejected_low == 40000

    def test_accept_keeps_largest(self):
        ab = AdaptiveContextBudget()
        ab.record_accept(10000)
        ab.record_accept(30000)
        ab.record_accept(15000)
        assert ab.accepted_high == 30000


class TestTerminalFloorEligible:
    """The Phase 4 backstop gate: only request-shaped 4xx rejections at an
    unknown window, never auth/tools/vision categories a smaller prompt can't
    fix."""

    def test_request_shaped_unknown_window_eligible(self):
        e = AgentError("LLM call failed: uncategorized 400")
        e._request_shaped = True
        assert _terminal_floor_eligible(e, None) is True

    def test_request_shaped_known_window_not_eligible(self):
        e = AgentError("LLM call failed: uncategorized 400")
        e._request_shaped = True
        assert _terminal_floor_eligible(e, 8192) is False

    def test_unmarked_error_not_eligible(self):
        assert _terminal_floor_eligible(AgentError("auth failed"), None) is False

    def test_tools_not_supported_not_eligible(self):
        from swival.report import ToolsNotSupportedError

        e = ToolsNotSupportedError("no tools")
        e._request_shaped = True  # even if marked, the category is excluded
        assert _terminal_floor_eligible(e, None) is False

    def test_vision_rejection_not_eligible(self):
        e = AgentError("this model does not support image input")
        e._request_shaped = True
        assert _terminal_floor_eligible(e, None) is False


# ---------------------------------------------------------------------------
# Phase 4/5: detection-independent backstop and the no-crash invariant in
# run_agent_loop (the loose tier and the unclassified-error backstop).
# ---------------------------------------------------------------------------


def _loop_kwargs(tmp_path, **overrides):
    from swival.thinking import ThinkingState
    from swival.todo import TodoState

    defaults = dict(
        api_base="http://127.0.0.1:1234",
        model_id="test-model",
        max_turns=2,
        max_output_tokens=1024,
        temperature=0.5,
        top_p=None,
        seed=None,
        context_length=None,
        base_dir=str(tmp_path),
        thinking_state=ThinkingState(verbose=False),
        resolved_commands={},
        skills_catalog={},
        skill_read_roots=[],
        extra_write_roots=[],
        files_mode="some",
        verbose=False,
        llm_kwargs={"provider": "lmstudio", "api_key": None},
        file_tracker=None,
        todo_state=TodoState(verbose=False),
        continue_here=False,
    )
    defaults.update(overrides)
    return defaults


def _completion_ok(content="recovered"):
    msg = SimpleNamespace(content=content, tool_calls=None, role="assistant")
    choice = SimpleNamespace(message=msg, finish_reason="stop")
    return SimpleNamespace(choices=[choice], usage=None)


class TestUnknownWindowOverflowLoop:
    """Integration: a loosely-phrased overflow at an unknown window is detected
    and recovered, and an unclassified request rejection is probed by the
    backstop rather than crashing."""

    def test_loose_phrasing_recovers_in_loop(self, tmp_path):
        from swival.agent import run_agent_loop

        def fake_completion(**kwargs):
            msgs = kwargs["messages"]
            if estimate_tokens(msgs, None) > 4090:
                raise _bad_request("Error code: 400 - input is too long")
            return _completion_ok("recovered")

        messages = [_sys("system"), _user("question " + "z" * 30000)]
        with patch("litellm.completion", side_effect=fake_completion):
            answer, exhausted = run_agent_loop(
                messages, _DUMMY_TOOLS, **_loop_kwargs(tmp_path)
            )
        assert answer == "recovered"
        assert exhausted is False

    def test_backstop_recovers_unclassified_rejection(self, tmp_path):
        """A non-overflow AgentError marked request-shaped at an unknown window
        is probed with a minimal prompt; if the minimal prompt is accepted the
        turn recovers instead of crashing."""
        from swival.agent import run_agent_loop

        def fake_call_llm(*args, **kwargs):
            prompt = args[2]
            if estimate_tokens(prompt, None) > 4090:
                ae = AgentError("LLM call failed: an uncategorized 400")
                ae._request_shaped = True
                raise ae
            return _ok_msg("recovered")

        messages = [_sys("system"), _user("question " + "z" * 30000)]
        with patch("swival.agent.call_llm", side_effect=fake_call_llm):
            answer, exhausted = run_agent_loop(
                messages, _DUMMY_TOOLS, **_loop_kwargs(tmp_path)
            )
        assert answer == "recovered"
        assert exhausted is False

    def test_backstop_surfaces_genuine_request_bug(self, tmp_path):
        """When even a minimal prompt is rejected with the same non-overflow
        error, the backstop surfaces it honestly rather than masking it."""
        from swival.agent import run_agent_loop

        def always_bad(*args, **kwargs):
            ae = AgentError("LLM call failed: genuinely malformed request")
            ae._request_shaped = True
            raise ae

        messages = [_sys("system"), _user("question " + "z" * 30000)]
        with patch("swival.agent.call_llm", side_effect=always_bad):
            with pytest.raises(AgentError, match="malformed request"):
                run_agent_loop(messages, _DUMMY_TOOLS, **_loop_kwargs(tmp_path))

    def test_known_window_unclassified_error_still_raises(self, tmp_path):
        """The backstop is scoped to the unknown-window regime: a request-shaped
        error under a known window surfaces as itself, today's behavior."""
        from swival.agent import run_agent_loop

        def always_bad(*args, **kwargs):
            ae = AgentError("LLM call failed: bad request under known window")
            ae._request_shaped = True
            raise ae

        messages = [_sys("system"), _user("hi")]
        with patch("swival.agent.call_llm", side_effect=always_bad):
            with pytest.raises(AgentError, match="known window"):
                run_agent_loop(
                    messages,
                    _DUMMY_TOOLS,
                    **_loop_kwargs(tmp_path, context_length=8192),
                )


# ---------------------------------------------------------------------------
# Phase 6: a mid-stream overflow still surfaces as ContextOverflowError through
# _completion_with_retry, with the loose tier gated on the unknown-window flag.
# ---------------------------------------------------------------------------


class TestStreamingOverflow:
    @staticmethod
    def _retry(**overrides):
        from swival.agent import _completion_with_retry

        kwargs = dict(max_retries=1, verbose=False, stream=True)
        kwargs.update(overrides)
        return _completion_with_retry({"messages": []}, **kwargs)

    def test_mid_stream_typed_overflow(self):
        import litellm

        def boom(*a, **k):
            raise litellm.ContextWindowExceededError(
                message="ctx", model="m", llm_provider="openai"
            )

        with patch("swival.agent._completion_via_stream", side_effect=boom):
            with pytest.raises(ContextOverflowError):
                self._retry()

    def test_mid_stream_loose_overflow_unknown_window(self):
        def boom(*a, **k):
            raise RuntimeError("the input is too long")

        with patch("swival.agent._completion_via_stream", side_effect=boom):
            with pytest.raises(ContextOverflowError):
                self._retry(unknown_context_window=True)

    def test_mid_stream_loose_not_overflow_known_window(self):
        def boom(*a, **k):
            raise RuntimeError("the input is too long")

        with patch("swival.agent._completion_via_stream", side_effect=boom):
            with pytest.raises(RuntimeError):
                self._retry(unknown_context_window=False)
