# -*- coding: utf-8 -*-
# pylint: disable=redefined-outer-name,protected-access,unused-argument
"""Unit tests for :class:`ScrollContextManager`.

Covers write-through dedup, the resume checkpoint (no re-append of a restored
window), the boundary-Msg double-presence fix, tool-result preview persistence,
degraded-durability fail-safe (no eviction when a write fails), and retention.
"""

import json
from pathlib import Path
from types import SimpleNamespace

import pytest
from agentscope.message import (
    Msg,
    TextBlock,
    ToolCallBlock,
    ToolResultBlock,
)

from qwenpaw.agents.context.scroll.history import HistoryStore
from qwenpaw.agents.context.scroll.manager import ScrollContextManager
from qwenpaw.agents.context.scroll.recall_tool import (
    RECALL_PAGE_METADATA_KEY,
    RecallLoopGuard,
)
from qwenpaw.agents.context.types import ContextWindowUnfitError, LogEntry
from qwenpaw.agents.memory.base_memory_manager import BaseMemoryManager
from qwenpaw.agents.tools.utils import truncate_text_output
from qwenpaw.constant import AUTO_MEMORY_SEARCH_BLOCK_IDS_KEY

# -- fixtures ---------------------------------------------------------------


def user(text: str) -> Msg:
    return Msg(
        name="u",
        role="user",
        content=[TextBlock(type="text", text=text)],
    )


def assistant(text: str, headline: str | None = None) -> Msg:
    if headline:
        text = f"{text}\n⟦ {headline} ⟧"
    return Msg(
        name="a",
        role="assistant",
        content=[TextBlock(type="text", text=text)],
    )


def assistant_with_tool(tcid: str, result_text: str = "RESULT") -> Msg:
    """An AS-2.0 accumulated assistant Msg: text + tool_call + tool_result."""
    return Msg(
        name="a",
        role="assistant",
        content=[
            TextBlock(type="text", text="calling a tool"),
            ToolCallBlock(type="tool_call", id=tcid, name="grep", input="{}"),
            ToolResultBlock(
                type="tool_result",
                id=tcid,
                name="grep",
                output=[TextBlock(type="text", text=result_text)],
            ),
        ],
    )


class FakeModel:
    """Constant token count, or a sequence (last value sticks) to model the
    window shrinking as compress() evicts. ``calls`` counts count_tokens
    invocations (compress must not recount an unchanged context)."""

    def __init__(self, tokens, context_size: int = 1000) -> None:
        self._tokens = (
            list(tokens) if isinstance(tokens, (list, tuple)) else [tokens]
        )
        self.context_size = context_size
        self.calls = 0

    async def count_tokens(self, *args, **kwargs) -> int:
        self.calls += 1
        if len(self._tokens) > 1:
            return self._tokens.pop(0)
        return self._tokens[0]


class FakeConfig:
    trigger_ratio = 0.1
    reserve_ratio = 0.5


class FakeState:
    def __init__(self, context: list[Msg]) -> None:
        self.context = context


class FakeAgent:
    """Minimal stand-in exposing the AS-2.0 surface the manager touches."""

    def __init__(
        self,
        context: list[Msg],
        tokens: int | list[int] = 200,
    ) -> None:
        self.state = FakeState(context)
        self.model = FakeModel(tokens)
        self.context_config = FakeConfig()
        self._split_return: tuple | None = None

    async def _prepare_model_input(self) -> dict:
        return {"tools": []}

    async def _split_context_for_compression(self, reserve, tools) -> tuple:
        if self._split_return is not None:
            return self._split_return
        # Default: compress everything but the last msg.
        return (self.state.context[:-1], self.state.context[-1:])


class AutoMemoryMsgBuilder(BaseMemoryManager):
    """Concrete memory manager used only to build synthetic memory messages."""

    async def start(self) -> None:
        pass

    async def close(self) -> bool:
        return True

    def get_memory_prompt(self) -> str:
        return ""

    def list_memory_tools(self) -> list:
        return []


@pytest.fixture
def store(tmp_path: Path) -> HistoryStore:
    h = HistoryStore(tmp_path / "history.db")
    yield h
    h.close()


def make_manager(store: HistoryStore, **kw) -> ScrollContextManager:
    kw.setdefault("session_id", "s1")
    kw.setdefault("agent_id", "ag1")
    return ScrollContextManager(history=store, **kw)


def auto_memory_search_msg(*, query: str, max_results: int, text: str) -> Msg:
    return AutoMemoryMsgBuilder(
        working_dir="",
        agent_id="ag1",
    )._build_auto_memory_search_msg(
        query=query,
        max_results=max_results,
        text=text,
    )


# -- write-through dedup -----------------------------------------------------


def test_persist_new_writes_each_turn_once(store: HistoryStore):
    mgr = make_manager(store)
    ctx = [user("hi"), assistant("there", headline="greeted")]
    agent = FakeAgent(ctx)
    mgr._persist_new(agent)
    mgr._persist_new(agent)  # idempotent: same context again
    assert store.count("s1") == 2


def test_persist_new_records_seq_and_headline_leaf(store: HistoryStore):
    mgr = make_manager(store)
    a = assistant("did it", headline="milestone")
    agent = FakeAgent([user("go"), a])
    mgr._persist_new(agent)
    assert a.id in mgr._leaf_by_id
    assert mgr._leaf_by_id[a.id].headline == "milestone"
    assert a.id in mgr._seq_by_id


def test_tool_result_persisted_under_tool_call_id(store: HistoryStore):
    mgr = make_manager(store)
    msg = assistant_with_tool("call-1", "big output")
    msg.content[2].metadata.update(
        {
            "qwenpaw_truncation": {
                "0": {
                    "file_path": "/tmp/artifact.txt",
                },
            },
        },
    )
    agent = FakeAgent([msg])
    mgr._persist_new(agent)
    rows = store._conn.execute(
        "SELECT content, metadata FROM conversation_history "
        "WHERE kind='tool_result' AND tool_call_id='call-1'",
    ).fetchall()
    assert len(rows) == 1
    assert rows[0]["content"] == "big output"
    assert json.loads(rows[0]["metadata"])["qwenpaw_truncation"]["0"] == {
        "file_path": "/tmp/artifact.txt",
    }


def test_auto_memory_search_message_not_persisted(store: HistoryStore):
    """Auto-search context is live-only and must not pollute history.db."""
    mgr = make_manager(store)
    auto_msg = auto_memory_search_msg(
        query="deploy plan",
        max_results=2,
        text="remembered deployment notes",
    )
    agent = FakeAgent([user("what was the deploy plan?"), auto_msg])

    mgr._persist_new(agent)

    rows = store._conn.execute(
        "SELECT kind, name, content FROM conversation_history ORDER BY seq",
    ).fetchall()
    assert [(r["kind"], r["name"], r["content"]) for r in rows] == [
        ("context_msg", None, "what was the deploy plan?"),
    ]


def test_auto_memory_search_blocks_stripped_from_mixed_message(
    store: HistoryStore,
):
    """If a real Msg also carries auto-search blocks, keep only real blocks."""
    mgr = make_manager(store)
    real_block = TextBlock(type="text", text="real reply")
    synthetic_block = TextBlock(type="text", text="synthetic memory context")
    msg = Msg(
        name="a",
        role="assistant",
        content=[real_block, synthetic_block],
        metadata={
            AUTO_MEMORY_SEARCH_BLOCK_IDS_KEY: [synthetic_block.id],
        },
    )

    mgr._persist_new(FakeAgent([msg]))

    rows = store._conn.execute(
        "SELECT kind, content, metadata FROM conversation_history",
    ).fetchall()
    assert len(rows) == 1
    assert rows[0]["kind"] == "model_turn"
    assert rows[0]["content"] == "real reply"


# -- resume: a restored window is not re-appended ---------------------------


def test_checkpoint_round_trip_prevents_reappend(store: HistoryStore):
    ctx = [
        user("hi"),
        assistant("a1", headline="h1"),
        assistant("a2", headline="h2"),
    ]
    mgr1 = make_manager(store)
    mgr1._persist_new(FakeAgent(ctx))
    assert store.count("s1") == 3
    snap = mgr1.to_dict()

    # Fresh manager (new process / reload) over the SAME restored context.
    mgr2 = make_manager(store)
    mgr2.load_state(snap)
    assert mgr2._persisted_ids == mgr1._persisted_ids
    mgr2._persist_new(FakeAgent(ctx))
    assert store.count("s1") == 3  # nothing re-appended


def test_reappend_blocked_by_db_even_without_checkpoint(store: HistoryStore):
    """Belt-and-suspenders: even a fresh manager with no checkpoint cannot
    duplicate rows, because the ux_dedup unique index drops them."""
    ctx = [user("hi"), assistant("a1", headline="h1")]
    make_manager(store)._persist_new(FakeAgent(ctx))
    make_manager(store)._persist_new(FakeAgent(ctx))  # no load_state
    assert store.count("s1") == 2


def test_load_state_tolerates_garbage(store: HistoryStore):
    mgr = make_manager(store)
    mgr.load_state(None)
    mgr.load_state({})
    assert mgr._persisted_ids == set()


# -- tool-result preview persistence ----------------------------------------


def test_tool_result_preview_is_persisted_once(store: HistoryStore):
    """Tool results are persisted exactly as they appear in live context."""
    mgr = make_manager(store)
    preview = (
        "partial output\n"
        "<<<EXECUTION_TOOL_RESULT_TRUNCATED>>>\n"
        "Full output saved to: /tmp/tool-result.txt."
    )
    agent = FakeAgent([assistant_with_tool("call-1", preview)])
    mgr._persist_new(agent)

    rows = store._conn.execute(
        "SELECT content FROM conversation_history "
        "WHERE kind='tool_result' AND tool_call_id='call-1'",
    ).fetchall()
    assert [row["content"] for row in rows] == [preview]
    mgr._persist_new(agent)
    rows = store._conn.execute(
        "SELECT content FROM conversation_history "
        "WHERE kind='tool_result' AND tool_call_id='call-1'",
    ).fetchall()
    assert [row["content"] for row in rows] == [preview]
    assert "call-1" in mgr._persisted_tcids


# -- compress: eviction + the boundary double-presence fix ------------------


async def test_compress_evicts_middle_into_index(store: HistoryStore):
    # A newer user turn follows the evictable middle: the active turn (last
    # user msg onward) stays live, the finished older turns are evicted.
    ctx = [
        user("task"),
        assistant("step", headline="did-step"),
        user("next question"),
        assistant("recent"),
    ]
    mgr = make_manager(store)
    agent = FakeAgent(ctx, tokens=200)
    agent._split_return = (
        ctx[:2],
        ctx[2:],
    )  # compress [task, step], keep [next, recent]
    await mgr.compress(agent)
    # Context is rebuilt as placeholder + tail.
    assert len(agent.state.context) == 3
    names = [m.name for m in agent.state.context]
    assert names[0] == "memory"  # the index placeholder leads
    assert "did-step" in mgr._index.render()
    memory_text = agent.state.context[0].get_text_content()
    assert "[continuity checkpoint]" in memory_text
    assert "Use recall_history with the listed seq span" in memory_text
    assert mgr.last_compress["evicted"] == 2  # /compact reporting source

    restored = make_manager(store)
    restored.load_state(mgr.to_dict())
    assert restored._continuity_checkpoint == mgr._continuity_checkpoint


async def test_compress_prunes_bookkeeping_to_live_context(
    store: HistoryStore,
):
    old_u = user("old request")
    old_a = assistant("old reply", headline="OLD")
    old_tool = assistant_with_tool("call-old")
    current_u = user("current request")
    current_tool = assistant_with_tool("call-current")
    ctx = [old_u, old_a, old_tool, current_u, current_tool]
    mgr = make_manager(store, summarize_unheadlined=False)
    agent = FakeAgent(ctx, tokens=200)
    agent._split_return = (ctx[:3], ctx[3:])

    await mgr.compress(agent)

    archived_ids = {old_u.id, old_a.id, old_tool.id}
    live_ids = {current_u.id, current_tool.id}
    for mapping in (
        mgr._persisted_ids,
        mgr._seq_by_id,
        mgr._model_turn_seq,
        mgr._model_turn_nblk,
        mgr._leaf_by_id,
    ):
        assert archived_ids.isdisjoint(mapping)
    assert live_ids <= mgr._persisted_ids
    assert live_ids <= mgr._seq_by_id.keys()
    assert "call-old" not in mgr._persisted_tcids
    assert "call-old" not in mgr._seq_by_tcid
    assert "call-current" in mgr._persisted_tcids
    assert "call-current" in mgr._seq_by_tcid
    assert mgr._synthetic_ids == {agent.state.context[0].id}

    previous_placeholder = agent.state.context[0].id
    next_u = user("next request")
    next_a = assistant("next reply")
    agent.state.context.extend([next_u, next_a])
    agent._split_return = (
        agent.state.context[:-2],
        agent.state.context[-2:],
    )

    await mgr.compress(agent)

    for mapping in (
        mgr._persisted_ids,
        mgr._seq_by_id,
        mgr._model_turn_seq,
        mgr._model_turn_nblk,
        mgr._leaf_by_id,
    ):
        assert live_ids.isdisjoint(mapping)
    assert {next_u.id, next_a.id} <= mgr._persisted_ids
    assert "call-current" not in mgr._persisted_tcids
    assert "call-current" not in mgr._seq_by_tcid
    assert len(mgr._synthetic_ids) == 1
    assert previous_placeholder not in mgr._synthetic_ids
    assert mgr._synthetic_ids == {agent.state.context[0].id}


async def test_compress_does_not_index_boundary_msg_still_in_tail(
    store: HistoryStore,
):
    """The boundary Msg is deep-copied into BOTH split halves under the same
    id. It must NOT be folded into the eviction index while its reserve copy
    is still live in the tail."""
    old_task = user("task")
    a = assistant("middle turn", headline="MIDDLE")
    current = user("current request")
    boundary = assistant("boundary turn", headline="BOUNDARY")
    ctx = [old_task, a, current, boundary]
    mgr = make_manager(store)
    agent = FakeAgent(ctx, tokens=200)
    # Mimic AgentScope: boundary id appears in BOTH halves (same id).
    compress_half = boundary
    reserve_half = Msg(
        name="a",
        role="assistant",
        content=[TextBlock(type="text", text="boundary tail blocks")],
    )
    object.__setattr__(
        reserve_half,
        "id",
        boundary.id,
    )  # same id, fewer blocks
    agent._split_return = (
        [old_task, a, current, compress_half],
        [reserve_half],
    )

    await mgr.compress(agent)
    rendered = mgr._index.render()
    assert "MIDDLE" in rendered  # the genuinely evicted turn
    assert "BOUNDARY" not in rendered  # still live → must not be indexed
    # And the boundary id is still present in the live context.
    assert boundary.id in {m.id for m in agent.state.context}
    assert boundary.id in mgr._persisted_ids
    assert boundary.id in mgr._seq_by_id
    assert boundary.id in mgr._model_turn_seq


async def test_compress_restores_complete_non_active_tool_boundary(
    store: HistoryStore,
):
    """A retained non-active boundary Msg must not remain a block fragment.

    AgentScope's splitter can reserve only the tool_result half of a Msg.  The
    orphan sanitizer used to drop that fragment, silently losing the retained
    boundary.  Restore the full live Msg before sanitizing instead.
    """
    old_u = user("older question")
    old_a = assistant("older reply", headline="OLD")
    boundary = assistant_with_tool("call-boundary")
    cur_u = user("current request")
    cur_a = assistant("current reply")
    ctx = [old_u, old_a, boundary, cur_u, cur_a]
    mgr = make_manager(store)
    agent = FakeAgent(ctx, tokens=200)
    reserve_fragment = Msg(
        name="a",
        role="assistant",
        content=[boundary.content[-1]],
    )
    object.__setattr__(reserve_fragment, "id", boundary.id)
    agent._split_return = (
        [old_u, old_a, boundary],
        [reserve_fragment, cur_u, cur_a],
    )

    await mgr.compress(agent)

    retained = next(
        msg for msg in agent.state.context if msg.id == boundary.id
    )
    assert retained is boundary
    assert [block.type for block in retained.content] == [
        "text",
        "tool_call",
        "tool_result",
    ]


async def test_compress_keeps_active_turn_live(store: HistoryStore):
    """The token-based split may push the CURRENT user request (and its
    running assistant chain) into the compress half. The active turn must
    stay live — evicting it makes the model answer an older message
    (#5747)."""
    old_u = user("older question")
    old_a = assistant("older reply", headline="OLD")
    cur_u = user("/heartbeat")
    cur_a = assistant("running tools", headline="RUNNING")
    ctx = [old_u, old_a, cur_u, cur_a]
    mgr = make_manager(store)
    agent = FakeAgent(ctx, tokens=200)
    # A long active turn blows the reserve budget: the split reserves nothing
    # and would evict the current request along with the old turns.
    agent._split_return = (ctx, [])
    await mgr.compress(agent)
    live_ids = [m.id for m in agent.state.context]
    assert cur_u.id in live_ids and cur_a.id in live_ids
    rendered = mgr._index.render()
    assert "OLD" in rendered  # the finished old turn is evicted
    assert "RUNNING" not in rendered  # the active turn is not
    # The active turn sits after the placeholder, mirroring a normal tail.
    names = [m.name for m in agent.state.context]
    assert names.index("memory") < live_ids.index(cur_u.id)


async def test_compress_does_not_evict_user_only_exchange_boundary(
    store: HistoryStore,
):
    """If the split lands between an old user request and its assistant
    reply, pull the reply into the evicted middle. Otherwise scroll archives a
    user-only span, misses the existing assistant headline, and has to call
    the model just to label the index."""
    old_u = user("generate a long fixture")
    old_a = assistant("fixture generated", headline="FIXTURE GENERATED")
    cur_u = user("summarize it")
    cur_a = assistant("summary", headline="SUMMARY")
    ctx = [old_u, old_a, cur_u, cur_a]
    mgr = make_manager(store, summarize_unheadlined=True)
    agent = FakeAgent(ctx, tokens=200)
    agent._split_return = ([old_u], [old_a, cur_u, cur_a])

    async def fail_summarize(*args, **kwargs):
        raise AssertionError("user-only fallback summarization should not run")

    mgr._summarize_span = fail_summarize

    await mgr.compress(agent)

    rendered = mgr._index.render()
    assert "FIXTURE GENERATED" in rendered
    assert old_a.id not in {m.id for m in agent.state.context}
    assert cur_u.id in {m.id for m in agent.state.context}
    assert cur_a.id in {m.id for m in agent.state.context}


def continuation_stub(text: str = "Continue working on the task.") -> Msg:
    """The user-role stub loop gates / stop handlers inject mid-turn."""
    from qwenpaw.constant import (
        LOOP_CONTINUATION_MESSAGE_TAG,
        QWENPAW_MESSAGE_TAG_KEY,
    )

    return Msg(
        name="user",
        role="user",
        content=[TextBlock(type="text", text=text)],
        metadata={QWENPAW_MESSAGE_TAG_KEY: LOOP_CONTINUATION_MESSAGE_TAG},
    )


async def test_active_turn_anchor_skips_continuation_stubs(
    store: HistoryStore,
):
    """A loop-continuation stub is user-role but NOT a new request: the
    active-turn anchor must stay on the real request that started the turn,
    or the real request becomes evictable middle again (#5746, loop-session
    flavor)."""
    old_u = user("older question")
    old_a = assistant("older reply", headline="OLD")
    real_u = user("write the report")
    a1 = assistant("working on it", headline="MID-TASK")
    stub = continuation_stub()
    a2 = assistant("continuing the report")
    ctx = [old_u, old_a, real_u, a1, stub, a2]
    mgr = make_manager(store)
    agent = FakeAgent(ctx, tokens=200)
    agent._split_return = (ctx, [])  # split would evict everything
    await mgr.compress(agent)
    live_ids = [m.id for m in agent.state.context]
    # The whole extended turn — real request, pre-stub reply, the stub
    # itself, and the post-stub reply — stays live.
    for m in (real_u, a1, stub, a2):
        assert m.id in live_ids
    rendered = mgr._index.render()
    assert "OLD" in rendered  # the finished old turn is evicted
    assert "MID-TASK" not in rendered  # the extended active turn is not


async def test_compress_noop_when_active_turn_fits_reserve(
    store: HistoryStore,
):
    """Single-user-msg session (e.g. a cron run): the whole context is the
    active turn and nothing is evictable. While the window still fits the
    reserve, compress leaves it untouched — no compaction, no fold."""
    ctx = [
        user("/heartbeat"),
        assistant("step one", headline="S1"),
        assistant("step two", headline="S2"),
    ]
    mgr = make_manager(store)
    agent = FakeAgent(ctx, tokens=200)  # over trigger, under reserve (500)
    agent._split_return = (ctx, [])
    await mgr.compress(agent)
    assert [m.id for m in agent.state.context] == [m.id for m in ctx]
    assert mgr._index.is_empty


def _multi_tool_turn(n: int = 3, *, padding: int = 0) -> Msg:
    """An accumulated assistant Msg with ``n`` completed call/result pairs."""
    blocks = []
    for i in range(n):
        blocks.append(TextBlock(type="text", text=f"step {i}"))
        blocks.append(
            ToolCallBlock(
                type="tool_call",
                id=f"c{i}",
                name="grep",
                input="{}",
            ),
        )
        blocks.append(
            ToolResultBlock(
                type="tool_result",
                id=f"c{i}",
                name="grep",
                output=[
                    TextBlock(
                        type="text",
                        text=f"RESULT-{i}" + "x" * padding,
                    ),
                ],
            ),
        )
    return Msg(name="a", role="assistant", content=blocks)


async def test_fold_not_triggered_between_reserve_and_trigger(
    store: HistoryStore,
):
    """With REALISTIC ratios (trigger 0.8, reserve 0.1), an active turn that
    exceeds the reserve but leaves most of the window free must NOT be
    folded — the fold is a last resort gated on the compression trigger,
    not on the soft reserve target. (Pre-fix, a 25k active turn in a 200k
    window was stubbed on every compress round of an ordinary long chat.)"""

    class _RealisticConfig:
        trigger_ratio = 0.8  # trigger at 800 of the 1000-token window
        reserve_ratio = 0.1  # reserve target 100

    old_u = user("older question")
    old_a = assistant("older reply", headline="OLD")
    # Deliberately exceed the former fixed 3 KB threshold. Once eviction has
    # relieved the pressure, size alone must not fold these live results.
    turn = _multi_tool_turn(padding=5000)
    ctx = [old_u, old_a, user("/heartbeat"), turn]
    mgr = make_manager(store)
    # 900 at the trigger check; 300 after eviction — over the reserve (100)
    # but far under the trigger (800).
    agent = FakeAgent(ctx, tokens=[900, 300])
    agent.context_config = _RealisticConfig()
    agent._split_return = (ctx[:2], ctx[2:])
    await mgr.compress(agent)

    rendered = mgr._index.render()
    assert "OLD" in rendered  # normal eviction happened
    # ... but every tool result of the active turn stays verbatim.
    for block in turn.content:
        if getattr(block, "type", None) == "tool_result":
            assert block.output[0].text.startswith("RESULT-")


async def test_compress_replaces_old_preview_with_tool_call_pointer(
    store: HistoryStore,
):
    text = "\n".join(f"line {idx}: {'x' * 40}" for idx in range(100))
    preview, metadata = truncate_text_output(
        text,
        start_line=50,
        total_lines=149,
        max_bytes=500,
        file_path="/tmp/full-tool-result.txt",
    )
    turn = assistant_with_tool("call-1", preview)
    turn.content[2].metadata.update(metadata)
    turn.content.extend(
        [
            ToolCallBlock(
                type="tool_call",
                id="call-2",
                name="grep",
                input="{}",
            ),
            ToolResultBlock(
                type="tool_result",
                id="call-2",
                name="grep",
                output=[TextBlock(type="text", text="newest result")],
            ),
        ],
    )
    ctx = [user("current request"), turn]
    mgr = make_manager(store)
    agent = FakeAgent(ctx, tokens=[600, 50])
    agent._split_return = (ctx, [])

    await mgr.compress(agent)

    compacted = turn.content[2].output[0].text
    assert compacted.startswith("[scroll folded]")
    assert 'recall_history(op="recall_tool"' in compacted
    assert "call-1" in compacted
    assert "read_file" not in compacted
    assert "/tmp/full-tool-result.txt" not in compacted
    assert "covers the next 120 bytes" not in compacted
    assert turn.content[-1].output[0].text == "newest result"
    assert mgr.last_compress["folded"] == 1


async def test_pressure_fold_stubs_older_results_keeps_newest(
    store: HistoryStore,
):
    """Nothing evictable and the window still overflows the compression
    trigger: the active turn's completed tool results are stubbed in place
    to recall pointers. The request, tool calls, reasoning, and the NEWEST
    result stay verbatim; the durable rows keep the full outputs; the Msg
    object (and id) is untouched so the runtime keeps extending the same
    message."""
    turn = _multi_tool_turn(padding=500)
    ctx = [user("/heartbeat"), turn]
    mgr = make_manager(store)
    agent = FakeAgent(ctx, tokens=600)  # > trigger (100): sustained pressure
    agent._split_return = (ctx, [])  # split would evict everything
    await mgr.compress(agent)

    # Same live objects — no rebuild happened (nothing was evicted).
    assert agent.state.context == ctx
    assert agent.state.context[-1] is turn

    def out_text(i: int) -> str:
        block = turn.content[3 * i + 2]
        return block.output[0].text

    # folded → tool-call-addressed stub pointing at structured recall
    assert 'recall_history(op="recall_tool"' in out_text(0)
    assert 'recall_history(op="recall_tool"' in out_text(1)
    assert out_text(2) == "RESULT-2" + "x" * 500  # newest stays verbatim
    # The durable rows still hold the FULL outputs (persisted before fold).
    for i in range(3):
        row = store._conn.execute(
            "SELECT content FROM conversation_history "
            f"WHERE kind='tool_result' AND tool_call_id='c{i}'",
        ).fetchone()
        assert row["content"] == f"RESULT-{i}" + "x" * 500

    # /compact reads this to report honestly (fold changes no msg count).
    assert mgr.last_compress["folded"] == 2

    # Idempotent: a second round neither double-folds nor rewrites rows.
    await mgr.compress(agent)
    assert out_text(0).count("[scroll folded]") == 1
    assert out_text(2) == "RESULT-2" + "x" * 500
    assert mgr.last_compress["folded"] == 0  # nothing newly folded


async def test_pressure_fold_does_not_replace_small_results_with_larger_stubs(
    store: HistoryStore,
):
    """There is no fixed lower size threshold, but folding must reclaim
    bytes. Small outputs stay live when their recovery pointers would grow the
    context, even during sustained pressure."""
    turn = _multi_tool_turn()
    ctx = [user("current request"), turn]
    mgr = make_manager(store)
    agent = FakeAgent(ctx, tokens=600)
    agent._split_return = (ctx, [])

    await mgr.compress(agent)

    for index in range(3):
        assert turn.content[3 * index + 2].output[0].text == f"RESULT-{index}"
    assert mgr.last_compress["folded"] == 0
    assert agent.model.calls == 1


async def test_pressure_fold_stops_after_largest_result_relieves_pressure(
    store: HistoryStore,
):
    """Pressure folding is incremental and reclaim-driven, not a fixed-size
    sweep: fold the largest profitable old result, recount, and stop once the
    trigger is met."""

    class _RealisticConfig:
        trigger_ratio = 0.8
        reserve_ratio = 0.1

    turn = _multi_tool_turn(n=4, padding=500)
    # Make the second completed result the best reclaim candidate. The fourth
    # result is newest and must remain verbatim regardless of size.
    turn.content[5].output[0].text = "LARGEST-" + "x" * 5000
    turn.content[11].output[0].text = "NEWEST-" + "x" * 8000
    ctx = [user("current request"), turn]
    mgr = make_manager(store)
    agent = FakeAgent(ctx, tokens=[900, 750])
    agent.context_config = _RealisticConfig()
    agent._split_return = (ctx, [])

    await mgr.compress(agent)

    assert turn.content[2].output[0].text.startswith("RESULT-0")
    assert turn.content[5].output[0].text.startswith("[scroll folded]")
    assert turn.content[8].output[0].text.startswith("RESULT-2")
    assert turn.content[11].output[0].text.startswith("NEWEST-")
    assert mgr.last_compress["folded"] == 1
    assert agent.model.calls == 2


async def test_consumed_recall_page_folds_to_next_cursor(
    store: HistoryStore,
):
    recall_turn = Msg(
        name="a",
        role="assistant",
        content=[
            ToolCallBlock(
                type="tool_call",
                id="recall-1",
                name="recall_history",
                input='{"op":"expand","lo":10,"hi":20}',
            ),
            ToolResultBlock(
                type="tool_result",
                id="recall-1",
                name="recall_history",
                output=[TextBlock(type="text", text="history\n" * 100)],
                metadata={
                    RECALL_PAGE_METADATA_KEY: {
                        "cursor": None,
                        "next_cursor": "0:660",
                        "total_rows": 1,
                        "complete": False,
                    },
                },
            ),
            ToolCallBlock(
                type="tool_call",
                id="newer-1",
                name="grep",
                input="{}",
            ),
            ToolResultBlock(
                type="tool_result",
                id="newer-1",
                name="grep",
                output=[TextBlock(type="text", text="newest result")],
            ),
        ],
    )
    ctx = [user("find the old decision"), recall_turn]
    mgr = make_manager(store)
    agent = FakeAgent(ctx, tokens=[600, 90])
    agent._split_return = (ctx, [])

    await mgr.compress(agent)

    output = recall_turn.content[1].output[0].text
    assert output.startswith("[scroll recall folded]")
    assert '"cursor": "0:660"' in output
    assert recall_turn.content[-1].output[0].text == "newest result"


async def test_single_message_over_hard_limit_fails_closed(
    store: HistoryStore,
):
    mgr = make_manager(store)
    agent = FakeAgent([user("oversized request")], tokens=1200)

    with pytest.raises(ContextWindowUnfitError) as exc:
        await mgr.compress(agent)

    assert exc.value.tokens == 1200
    assert exc.value.hard_limit == 1000


async def test_steady_state_counts_once_and_warns_once(
    store: HistoryStore,
    caplog,
):
    """Over the trigger with nothing evictable, compactable, or foldable:
    each compress pays exactly ONE token count (the trigger check — the
    context never changed, so recounting it is waste), and the
    still-over-trigger warning fires once per overflow episode, not once
    per reasoning step."""
    import logging as _logging

    ctx = [
        user("/heartbeat"),
        assistant("step one"),
        assistant("step two"),
    ]
    mgr = make_manager(store)
    agent = FakeAgent(ctx, tokens=600)  # > trigger (100), nothing to shrink
    agent._split_return = (ctx, [])
    with caplog.at_level(_logging.WARNING):
        await mgr.compress(agent)
        assert agent.model.calls == 1  # trigger check only
        await mgr.compress(agent)
        assert agent.model.calls == 2
    stuck = [
        r for r in caplog.records if "compression trigger" in r.getMessage()
    ]
    assert len(stuck) == 1


async def test_manual_compact_trigger_does_not_warn_below_reserve(
    store: HistoryStore,
    caplog,
):
    """A manual /compact trigger is intentionally near zero and must not be
    reported as a context overflow when the result fits the reserve target."""
    import logging as _logging

    class _ManualConfig:
        trigger_ratio = 1e-6
        reserve_ratio = 0.1

    ctx = [user("old"), assistant("old reply"), user("current")]
    mgr = make_manager(store)
    agent = FakeAgent(ctx, tokens=[200, 80])
    agent._split_return = (ctx[:2], ctx[2:])

    with caplog.at_level(_logging.WARNING):
        await mgr.compress(agent, _ManualConfig())

    assert not any(
        "compression trigger" in record.getMessage()
        for record in caplog.records
    )


async def test_pressure_does_not_compact_index_before_tier_cap(
    store: HistoryStore,
):
    """Context pressure must not roll up index blocks before the tier cap."""
    from qwenpaw.agents.context.scroll.eviction_index import Leaf

    mgr = make_manager(store)
    for i in range(3):  # a multi-block Tier 0 from earlier evictions
        mgr._index.add_eviction(
            [Leaf(seq=i * 10 + 1, headline=f"h{i}")],
            seq_lo=i * 10,
            seq_hi=i * 10 + 9,
        )
    ctx = [user("/heartbeat"), assistant("working", headline="W")]
    mgr._persist_new(FakeAgent(ctx))
    agent = FakeAgent(ctx, tokens=600)  # > reserve: sustained pressure
    agent._split_return = (ctx, [])  # nothing evictable
    before = mgr._index.describe()
    await mgr.compress(agent)
    # All three blocks remain detailed; only the tier cap may roll them up.
    assert mgr._index.describe() == before
    assert (
        len([ln for ln in mgr._index.describe().splitlines() if "[seq" in ln])
        == 3
    )
    # The active turn is still live.
    assert agent.state.context[-1].id == ctx[-1].id


# -- generated headlines for un-headlined evicted spans ---------------------


class _CallableModel(FakeModel):
    """A ``FakeModel`` that is also callable as a chat model.

    ``reply`` is the text an index call returns (``<n>: headline`` section
    lines by convention); set it to an ``Exception`` instance to have the call
    raise (to exercise the fallback). ``call_count`` records how many times it
    was invoked as a chat model — so a test can assert the index path was (or
    was not) taken. ``last_body`` captures the user message (the numbered
    sections the harness assembled) of the last call."""

    def __init__(self, tokens, reply="1: a legacy 1.x decision", **kw):
        super().__init__(tokens, **kw)
        self._reply = reply
        self.call_count = 0
        self.last_body = ""

    async def __call__(self, messages, *args, **kwargs):
        self.call_count += 1
        self.last_body = messages[-1].get_text_content()
        if isinstance(self._reply, Exception):
            raise self._reply
        return SimpleNamespace(text=self._reply)


def _agent_with_callable_model(
    ctx,
    reply="1: a legacy 1.x decision",
    tokens=200,
):
    agent = FakeAgent(ctx, tokens=tokens)
    agent.model = _CallableModel(tokens, reply=reply)
    return agent


def _index_headline_lines(mgr) -> list[str]:
    """The ``·`` headline lines of the eviction-index map (text after ⟦)."""
    return [ln for ln in mgr._index.describe().splitlines() if "·" in ln]


async def test_unheadlined_span_gets_generated_summary(store: HistoryStore):
    """An evicted span with no headline is labelled by the model's synthesized
    headline instead of the bare ``(no milestone)`` marker."""
    ctx = [
        user("what was the old plan"),
        assistant("we shipped v1 without milestones"),  # NO headline
        user("next question"),
        assistant("recent"),
    ]
    mgr = make_manager(store, summarize_unheadlined=True)
    agent = _agent_with_callable_model(
        ctx,
        reply="1: shipped v1 sans milestones",
    )
    agent._split_return = (ctx[:2], ctx[2:])
    await mgr.compress(agent)
    index = mgr._index.describe()
    assert "shipped v1 sans milestones" in index
    assert "(no milestone)" not in index
    assert agent.model.call_count == 1


async def test_unheadlined_span_tiled_into_multiple_headlines(
    store: HistoryStore,
):
    """A longer un-headlined span is tiled into several harness-addressed
    headlines, each its own ``·`` line — structurally like real milestones.
    The model only writes ``<n>: headline``; the seqs are the harness's."""
    ctx = [
        user("q1 about billing"),
        assistant("answered billing"),  # NO headline
        user("q2 about shipping"),
        assistant("answered shipping"),  # NO headline
        user("next question"),
        assistant("recent"),
    ]
    mgr = make_manager(store, summarize_unheadlined=True)
    agent = _agent_with_callable_model(
        ctx,
        reply=(
            "1: billing question resolved\n" "2: shipping question resolved"
        ),
    )
    agent._split_return = (ctx[:4], ctx[4:])
    await mgr.compress(agent)
    index = mgr._index.describe()
    assert "billing question resolved" in index
    assert "shipping question resolved" in index
    # Two distinct headline lines, not one coarse summary.
    assert len(_index_headline_lines(mgr)) == 2
    # The harness split the span into numbered sections for the model.
    assert "[1]" in agent.model.last_body
    assert "[2]" in agent.model.last_body


async def test_skipped_section_keeps_extractive_fallback(store: HistoryStore):
    """A section the model omits is still labelled — the harness fills it with
    an extractive fallback drawn from that section's own text, never
    ``(no milestone)`` for a section that had content."""
    ctx = [
        user("distinctive-billing-question"),
        assistant("answered billing"),  # NO headline
        user("distinctive-shipping-question"),
        assistant("answered shipping"),  # NO headline
        user("next question"),
        assistant("recent"),
    ]
    mgr = make_manager(store, summarize_unheadlined=True)
    # Model labels section 1 only; section 2 must fall back to its own content.
    agent = _agent_with_callable_model(ctx, reply="1: billing resolved")
    agent._split_return = (ctx[:4], ctx[4:])
    await mgr.compress(agent)
    index = mgr._index.describe()
    assert "billing resolved" in index  # model's headline for section 1
    assert "distinctive-shipping-question" in index  # fallback for section 2
    assert len(_index_headline_lines(mgr)) == 2


async def test_bare_reply_lines_map_positionally(store: HistoryStore):
    """A reply without ``<n>:`` prefixes still lines up: bare headline lines
    are assigned to sections in order."""
    ctx = [
        user("old thing"),
        assistant("did old thing"),  # NO headline
        user("next question"),
        assistant("recent"),
    ]
    mgr = make_manager(store, summarize_unheadlined=True)
    agent = _agent_with_callable_model(
        ctx,
        reply="This stretch was about the old thing",
    )
    agent._split_return = (ctx[:2], ctx[2:])
    await mgr.compress(agent)
    index = mgr._index.describe()
    assert "This stretch was about the old thing" in index
    assert "(no milestone)" not in index
    assert len(_index_headline_lines(mgr)) == 1


async def test_headlined_span_never_calls_summary_model(store: HistoryStore):
    """A span that already has a headline uses it as the leaf — no index
    call is made (leaves present)."""
    ctx = [
        user("task"),
        assistant("step", headline="did-step"),
        user("next question"),
        assistant("recent"),
    ]
    mgr = make_manager(store, summarize_unheadlined=True)
    agent = _agent_with_callable_model(ctx)
    agent._split_return = (ctx[:2], ctx[2:])
    await mgr.compress(agent)
    assert "did-step" in mgr._index.describe()
    assert agent.model.call_count == 0


async def test_unheadlined_span_falls_back_when_summary_fails(
    store: HistoryStore,
):
    """A model/timeout error must never abort eviction — the span keeps the
    ``(no milestone)`` marker and the evicted count is unaffected."""
    ctx = [
        user("old thing"),
        assistant("did old thing"),  # NO headline
        user("next question"),
        assistant("recent"),
    ]
    mgr = make_manager(store, summarize_unheadlined=True)
    agent = _agent_with_callable_model(ctx, reply=RuntimeError("model down"))
    agent._split_return = (ctx[:2], ctx[2:])
    await mgr.compress(agent)
    assert "(no milestone)" in mgr._index.describe()
    assert mgr.last_compress["evicted"] == 2


async def test_summary_disabled_keeps_no_milestone(store: HistoryStore):
    """With the flag off the span stays ``(no milestone)`` and the model is
    never called — the default behaviour is preserved."""
    ctx = [
        user("old thing"),
        assistant("did old thing"),  # NO headline
        user("next question"),
        assistant("recent"),
    ]
    mgr = make_manager(store, summarize_unheadlined=False)
    agent = _agent_with_callable_model(ctx)
    agent._split_return = (ctx[:2], ctx[2:])
    await mgr.compress(agent)
    assert "(no milestone)" in mgr._index.describe()
    assert agent.model.call_count == 0


def test_seq_by_tcid_round_trips_through_checkpoint(store: HistoryStore):
    mgr = make_manager(store)
    mgr._persist_new(FakeAgent([assistant_with_tool("call-7", "out")]))
    assert "call-7" in mgr._seq_by_tcid
    mgr2 = make_manager(store)
    mgr2.load_state(mgr.to_dict())
    assert mgr2._seq_by_tcid == mgr._seq_by_tcid


# -- degraded durability: no eviction on write failure ----------------------


async def test_compress_does_not_evict_when_persist_fails(
    store: HistoryStore,
    monkeypatch,
):
    import sqlite3

    ctx = [user("task"), assistant("step", headline="s"), assistant("more")]
    mgr = make_manager(store)
    agent = FakeAgent(ctx, tokens=200)

    def boom(*a, **k):
        raise sqlite3.OperationalError("disk full")

    monkeypatch.setattr(store, "append", boom)
    await mgr.compress(agent)
    # Persist failed → degraded, and the context was left untouched (no
    # placeholder injected, no rows pointing at nonexistent durable data).
    assert store.degraded is True
    assert [m.id for m in agent.state.context] == [m.id for m in ctx]


def test_on_save_swallows_write_failure(store: HistoryStore, monkeypatch):
    import sqlite3

    mgr = make_manager(store)
    agent = FakeAgent([user("hi")])

    def boom(*a, **k):
        raise sqlite3.OperationalError("io error")

    monkeypatch.setattr(store, "append", boom)
    mgr.on_save(agent, None)  # must not raise
    assert store.degraded is True


def test_on_save_after_close_is_quiet_noop(store: HistoryStore):
    """Teardown race: an on_save after close is skipped quietly, not reported
    as degraded durability."""
    mgr = make_manager(store)
    agent = FakeAgent([user("hi"), assistant("there", headline="h")])
    store.close()
    assert store.closed is True

    mgr.on_save(agent, None)  # must not raise "closed database"
    # Skipped, not failed: durability stays healthy and nothing was persisted.
    assert store.degraded is False
    assert store.write_failures == 0
    assert mgr._persisted_ids == set()


def test_on_save_resets_recall_guard_for_next_real_user_turn(
    store: HistoryStore,
):
    guard = RecallLoopGuard()
    first_user = user("first request")
    agent = FakeAgent([first_user])
    mgr = make_manager(store, recall_loop_guard=guard)

    mgr.on_save(agent, None)
    payload = {"lo": 1, "hi": 3}
    generation, notice = guard.claim("expand", payload)
    assert generation is not None
    assert notice is None
    guard.finish("expand", payload, generation, block=True)
    assert guard.is_blocked("expand", payload) is True

    second_user = user("second request")
    agent.state.context.append(second_user)
    mgr.on_save(agent, None)

    assert guard.turn_id == second_user.id
    assert guard.is_blocked("expand", payload) is False


# -- optional dialog offload (offload_dialog opt-in) ------------------------


class _RecordingOffloader:
    def __init__(self) -> None:
        self.calls: list = []

    async def offload_context(self, session_id, msgs):
        self.calls.append((session_id, [m.id for m in msgs]))
        return "dialog/2026-06-19.jsonl"


def _compactable(store, **kw):
    ctx = [
        user("task"),
        assistant("step", headline="did-step"),
        user("next question"),
        assistant("recent"),
    ]
    mgr = make_manager(store, **kw)
    agent = FakeAgent(ctx, tokens=200)
    agent._split_return = (
        ctx[:2],
        ctx[2:],
    )  # evict [step]; keep task + [next, recent]
    return mgr, agent, ctx


async def test_compress_offloads_evicted_middle_when_configured(store):
    off = _RecordingOffloader()
    mgr, agent, ctx = _compactable(store, offloader=off)
    await mgr.compress(agent)
    assert len(off.calls) == 1
    session_id, ids = off.calls[0]
    assert session_id == "s1"
    assert ids == [ctx[0].id, ctx[1].id]  # exactly the evicted middle


async def test_compress_does_not_offload_without_offloader(store):
    mgr, agent, _ = _compactable(store)  # no offloader wired
    await mgr.compress(agent)  # must work + write nothing to dialog
    assert "memory" in [m.name for m in agent.state.context]


async def test_offload_failure_does_not_abort_eviction(store):
    class _Boom:
        async def offload_context(self, session_id, msgs):
            raise OSError("disk full")

    mgr, agent, _ = _compactable(store, offloader=_Boom())
    await mgr.compress(agent)  # best-effort archive: swallow + keep evicting
    assert "did-step" in mgr._index.render()
    assert "memory" in [m.name for m in agent.state.context]


# -- retention ---------------------------------------------------------------


def test_purge_old_zero_keeps_everything(store: HistoryStore):
    mgr = make_manager(store)
    store.append(
        session_id="s1",
        dedup_key="m1",
        entry=LogEntry(
            kind="model_turn",
            content="x",
            created_at="2000-01-01T00:00:00+00:00",
        ),
    )
    assert mgr.purge_old(0) == 0
    assert store.count("s1") == 1


def test_purge_old_drops_rows_past_window(store: HistoryStore):
    mgr = make_manager(store)
    store.append(
        session_id="s1",
        dedup_key="m1",
        entry=LogEntry(
            kind="model_turn",
            content="ancient",
            created_at="2000-01-01T00:00:00+00:00",
        ),
    )
    assert mgr.purge_old(1) == 1
    assert store.count("s1") == 0


def test_serialize_persists_runtime_tag():
    """The qwenpaw_tag survives into the durable row's metadata, so the
    recall layer's SQL floor can tell continuation stubs from requests."""
    from qwenpaw.agents.context.scroll.serialize import msg_to_entries
    from qwenpaw.constant import (
        LOOP_CONTINUATION_MESSAGE_TAG,
        QWENPAW_MESSAGE_TAG_KEY,
    )

    (entry,) = msg_to_entries(continuation_stub())
    assert entry.metadata == {
        QWENPAW_MESSAGE_TAG_KEY: LOOP_CONTINUATION_MESSAGE_TAG,
    }
    (plain,) = msg_to_entries(user("hello"))
    assert not plain.metadata


def test_serialize_captures_tool_input():
    """A tool call's arguments land in the ``tool_input`` column (it used to be
    dropped — only ``blocks`` carried them — so ``recall_tool`` returned None).
    """
    from qwenpaw.agents.context.scroll.serialize import msg_to_entries

    msg = Msg(
        name="a",
        role="assistant",
        content=[
            TextBlock(type="text", text="reading a file"),
            ToolCallBlock(
                type="tool_call",
                id="call-1",
                name="read_file",
                input='{"file_path": "PROFILE.md"}',
            ),
        ],
    )
    entries = msg_to_entries(msg)
    turn = next(e for e in entries if e.kind == "model_turn")
    assert turn.name == "read_file"
    assert turn.tool_call_id == "call-1"
    assert turn.tool_input == '{"file_path": "PROFILE.md"}'


def test_tool_input_round_trips_to_db(store: HistoryStore):
    """End-to-end: the persisted row's ``tool_input`` column is populated."""
    from qwenpaw.agents.context.scroll.serialize import msg_to_entries

    msg = Msg(
        name="a",
        role="assistant",
        content=[
            ToolCallBlock(
                type="tool_call",
                id="call-9",
                name="grep",
                input='{"pattern": "x"}',
            ),
        ],
    )
    (turn,) = msg_to_entries(msg)
    store.append(session_id="s1", dedup_key="m1", entry=turn)
    row = store._conn.execute(
        "SELECT tool_input FROM conversation_history "
        "WHERE tool_call_id='call-9'",
    ).fetchone()
    assert row["tool_input"] == '{"pattern": "x"}'
