"""Tests for pre-flight compaction (Feature D).

Covers:
- Pre-flight triggers when transcript exceeds the configured context-window ratio
- Pre-flight does NOT trigger when under threshold
- cron: and subagent: sessions are skipped
- Missing sessions don't error
"""

from __future__ import annotations

import asyncio
from collections.abc import AsyncIterator
from types import SimpleNamespace
from typing import Any
from unittest.mock import ANY, AsyncMock, MagicMock, patch

import pytest

from opensquilla.engine import runtime as runtime_module
from opensquilla.engine.runtime import TurnRunner
from opensquilla.gateway.config import GatewayConfig
from opensquilla.provider import DoneEvent as ProviderDone
from opensquilla.provider import Message, ModelInfo
from opensquilla.provider import TextDeltaEvent as ProviderText
from opensquilla.provider.model_catalog import ModelCatalog
from opensquilla.session.compaction import CompactionConfig
from opensquilla.session.manager import SessionManager
from opensquilla.session.models import TranscriptEntry
from opensquilla.session.storage import SessionStorage
from opensquilla.tools.types import CallerKind, ToolContext

# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _make_entry(content: str, role: str = "user") -> TranscriptEntry:
    return TranscriptEntry(
        session_id="test-session-id",
        session_key="test:key",
        role=role,
        content=content,
    )


def _flush_enabled_config(**overrides: Any) -> SimpleNamespace:
    memory = {
        "flush_enabled": True,
        "flush_pre_compaction": True,
        "flush_timeout_seconds": 0.25,
        "flush_background_timeout_seconds": 120.0,
        "flush_compaction_requires_safe_receipt": False,
    }
    memory.update(overrides)
    return SimpleNamespace(memory=SimpleNamespace(**memory))


def _make_assistant_tool_entry(content: str, tool_calls: list[dict[str, Any]]) -> TranscriptEntry:
    return TranscriptEntry(
        session_id="test-session-id",
        session_key="test:key",
        role="assistant",
        content=content,
        tool_calls=tool_calls,
        token_count=1,
    )


def _make_assistant_reasoning_entry(content: str, reasoning_content: str) -> TranscriptEntry:
    return TranscriptEntry(
        session_id="test-session-id",
        session_key="test:key",
        role="assistant",
        content=content,
        reasoning_content=reasoning_content,
        token_count=1,
    )


def _checkpoint_receipt() -> SimpleNamespace:
    return SimpleNamespace(
        scope="checkpoint",
        status="checkpoint_saved",
        source_path="memory/.checkpoints/s/turn.jsonl",
        content_hash="h1",
    )


def _flush_receipt(**overrides):
    payload = {
        "mode": "llm",
        "error": None,
        "indexed_chunk_count": 1,
        "integrity_status": "ok",
        "output_coverage_status": "ok",
        "invalid_candidate_count": 0,
        "candidate_missing_ids": [],
        "obligation_status": "ok",
        "obligation_missing_ids": [],
    }
    payload.update(overrides)
    return SimpleNamespace(**payload)


class _FakeCompactionProvider:
    provider_name = "openai"

    def __init__(self, model: str = "provider/model") -> None:
        self._api_key = "preflight-provider-key"
        self._model = model
        self._base_url = "https://openrouter.ai/api/v1"

    @property
    def model(self) -> str:
        return self._model

    def chat(self, messages: list[Message], tools=None, config=None) -> AsyncIterator:
        return self._stream()

    async def _stream(self) -> AsyncIterator:
        yield ProviderText(text="ok")
        yield ProviderDone(stop_reason="end_turn", input_tokens=1, output_tokens=1)

    async def list_models(self) -> list[ModelInfo]:
        return []


class _FakeSelectorClone:
    current_config = SimpleNamespace(model="provider/model")

    def __init__(self, provider: _FakeCompactionProvider) -> None:
        self.provider = provider
        self.override_calls: list[str] = []

    def override_model(self, model: str) -> None:
        self.override_calls.append(model)
        self.provider._model = model
        self.current_config = SimpleNamespace(model=model)

    def resolve(self) -> _FakeCompactionProvider:
        return self.provider


class _FakeProviderSelector:
    current_config = SimpleNamespace(model="provider/model")

    def __init__(self, provider: _FakeCompactionProvider | None = None) -> None:
        self.provider = provider or _FakeCompactionProvider()
        self.clone_instance = _FakeSelectorClone(self.provider)
        self.override_calls: list[str] = []

    def clone(self) -> _FakeSelectorClone:
        return self.clone_instance

    def override_model(self, model: str) -> None:
        self.override_calls.append(model)

    def resolve(self) -> _FakeCompactionProvider:
        return self.provider


class _ResultCompactionSessionManager:
    def __init__(self, transcript: list[TranscriptEntry]) -> None:
        self._transcript = transcript
        self.compact_with_result_calls: list[tuple[str, int, object | None]] = []
        self.compact_with_result_kwargs: list[dict[str, object | None]] = []

    async def get_transcript(self, session_key: str) -> list[TranscriptEntry]:
        return list(self._transcript)

    async def compact_with_result(
        self,
        session_key: str,
        context_window_tokens: int,
        config: object | None = None,
        **kwargs,
    ) -> SimpleNamespace:
        self.compact_with_result_calls.append((session_key, context_window_tokens, config))
        self.compact_with_result_kwargs.append(dict(kwargs))
        return SimpleNamespace(
            summary="summary text",
            kept_entries=[{"role": "assistant", "content": "tail"}],
            removed_count=4,
            chunks_processed=3,
            summary_source="llm",
            tokens_before=1000,
            tokens_after=200,
            remaining_budget_tokens=800,
        )


class _FailingResultCompactionSessionManager(_ResultCompactionSessionManager):
    async def compact_with_result(
        self,
        session_key: str,
        context_window_tokens: int,
        config: object | None = None,
        **kwargs,
    ) -> SimpleNamespace:
        self.compact_with_result_calls.append((session_key, context_window_tokens, config))
        self.compact_with_result_kwargs.append(dict(kwargs))
        raise RuntimeError("preimage write failed")


class _StaleResultCompactionSessionManager(_ResultCompactionSessionManager):
    async def compact_with_result(
        self,
        session_key: str,
        context_window_tokens: int,
        config: object | None = None,
        **kwargs,
    ) -> SimpleNamespace:
        self.compact_with_result_calls.append((session_key, context_window_tokens, config))
        self.compact_with_result_kwargs.append(dict(kwargs))
        return SimpleNamespace(
            summary="",
            kept_entries=list(self._transcript),
            removed_count=0,
            chunks_processed=0,
            summary_source="skipped",
            skip_reason="stale_preimage",
            tokens_before=2400,
            tokens_after=2400,
            remaining_budget_tokens=0,
        )


@pytest.fixture
async def session_mgr(tmp_path):
    storage = SessionStorage(":memory:")
    await storage.connect()
    mgr = SessionManager(storage, checkpoint_workspace_dir=tmp_path)
    yield mgr
    await storage.close()


# ---------------------------------------------------------------------------
# Tests: _maybe_preflight_compact
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_preflight_no_session_manager_is_noop() -> None:
    """When session_manager is None, pre-flight silently returns."""
    runner = TurnRunner(provider_selector=MagicMock(), session_manager=None)
    # Should not raise
    await runner._maybe_preflight_compact("some:session", 200_000)


@pytest.mark.asyncio
async def test_preflight_skips_cron_sessions() -> None:
    """cron: prefixed sessions are skipped regardless of token count."""
    mock_sm = MagicMock()
    mock_sm.compact = AsyncMock()
    mock_sm.get_transcript = AsyncMock()

    runner = TurnRunner(provider_selector=MagicMock(), session_manager=mock_sm)
    await runner._maybe_preflight_compact("cron:daily-job", 200_000)

    mock_sm.get_transcript.assert_not_called()
    mock_sm.compact.assert_not_called()


@pytest.mark.asyncio
async def test_preflight_skips_subagent_sessions() -> None:
    """subagent: prefixed sessions are skipped regardless of token count."""
    mock_sm = MagicMock()
    mock_sm.compact = AsyncMock()
    mock_sm.get_transcript = AsyncMock()

    runner = TurnRunner(provider_selector=MagicMock(), session_manager=mock_sm)
    await runner._maybe_preflight_compact("subagent:worker-1", 200_000)

    mock_sm.get_transcript.assert_not_called()
    mock_sm.compact.assert_not_called()


@pytest.mark.asyncio
async def test_preflight_missing_session_does_not_error() -> None:
    """KeyError from get_transcript (session doesn't exist) is swallowed."""
    mock_sm = MagicMock()
    mock_sm.compact = AsyncMock()
    mock_sm.get_transcript = AsyncMock(side_effect=KeyError("Session not found"))

    runner = TurnRunner(provider_selector=MagicMock(), session_manager=mock_sm)
    # Should not raise
    await runner._maybe_preflight_compact("missing:session", 200_000)

    mock_sm.compact.assert_not_called()


@pytest.mark.asyncio
async def test_preflight_empty_transcript_is_noop() -> None:
    """Empty transcript skips compaction."""
    mock_sm = MagicMock()
    mock_sm.compact = AsyncMock()
    mock_sm.get_transcript = AsyncMock(return_value=[])

    runner = TurnRunner(provider_selector=MagicMock(), session_manager=mock_sm)
    await runner._maybe_preflight_compact("user:session", 200_000)

    mock_sm.compact.assert_not_called()


@pytest.mark.asyncio
async def test_preflight_under_threshold_does_not_compact() -> None:
    """Transcript under the configured context-window ratio → no compaction."""
    # 100 tokens worth of content (estimate_tokens("x" * 400) ≈ 100 with len//4 fallback)
    entries = [_make_entry("x" * 400)]  # ~100 tokens

    mock_sm = MagicMock()
    mock_sm.compact = AsyncMock()
    mock_sm.get_transcript = AsyncMock(return_value=entries)

    runner = TurnRunner(provider_selector=MagicMock(), session_manager=mock_sm)
    # Threshold = 200_000 * 0.85 = 170_000 — 100 tokens is well under
    await runner._maybe_preflight_compact("user:session", 200_000)

    mock_sm.compact.assert_not_called()


@pytest.mark.asyncio
async def test_preflight_uses_configured_compaction_ratio() -> None:
    """Operators can tune the preflight threshold without code changes."""
    context_window = 1000
    entries = [_make_entry("a")]

    mock_sm = MagicMock()
    mock_sm.compact = AsyncMock(return_value="summary text")
    mock_sm.get_transcript = AsyncMock(return_value=entries)

    runner = TurnRunner(
        provider_selector=MagicMock(),
        session_manager=mock_sm,
        config=SimpleNamespace(preflight_compact_ratio=0.5),
    )

    with patch("opensquilla.session.tokenizer.estimate_tokens", return_value=600):
        await runner._maybe_preflight_compact("user:session", context_window)

    mock_sm.compact.assert_called_once_with("user:session", context_window)


@pytest.mark.asyncio
async def test_preflight_above_threshold_triggers_compact() -> None:
    """Transcript exceeding the default 85% context-window ratio triggers compaction."""
    # Use patch to control estimate_tokens so threshold math is deterministic
    context_window = 1000

    # Create entries whose total estimated tokens exceed threshold
    entries = [_make_entry("a" * 4000)]  # len//4 = 1000 tokens > 850

    mock_sm = MagicMock()
    mock_sm.compact = AsyncMock(return_value="summary text")
    mock_sm.get_transcript = AsyncMock(return_value=entries)

    runner = TurnRunner(provider_selector=MagicMock(), session_manager=mock_sm)

    with patch("opensquilla.session.tokenizer.estimate_tokens", return_value=1000):
        await runner._maybe_preflight_compact("user:session", context_window)

    mock_sm.compact.assert_called_once_with("user:session", context_window)


@pytest.mark.asyncio
async def test_preflight_checkpoint_runs_before_compact() -> None:
    context_window = 1000
    entries = [_make_entry("early durable fact " + ("a" * 4000))]
    calls: list[str] = []
    mock_sm = MagicMock()
    mock_sm.get_transcript = AsyncMock(return_value=entries)
    mock_sm.compact = AsyncMock(
        side_effect=lambda *args, **kwargs: calls.append("compact") or "summary"
    )
    mock_sm.record_memory_checkpoint = AsyncMock(
        side_effect=lambda *args, **kwargs: calls.append("checkpoint")
        or _checkpoint_receipt()
    )

    runner = TurnRunner(provider_selector=MagicMock(), session_manager=mock_sm)

    with patch("opensquilla.session.tokenizer.estimate_tokens", return_value=1000):
        await runner._maybe_preflight_compact("agent:ops:long-session", context_window)

    assert calls[0] == "checkpoint"
    assert "compact" in calls


@pytest.mark.asyncio
async def test_preflight_compacts_when_distill_fails_after_checkpoint() -> None:
    context_window = 1000
    entries = [_make_entry("early durable fact " + ("a" * 4000))]
    calls: list[str] = []
    mock_sm = MagicMock()
    mock_sm.get_transcript = AsyncMock(return_value=entries)
    mock_sm.record_memory_checkpoint = AsyncMock(
        side_effect=lambda *args, **kwargs: calls.append("checkpoint")
        or _checkpoint_receipt()
    )
    mock_sm.compact = AsyncMock(
        side_effect=lambda *args, **kwargs: calls.append("compact") or "summary text"
    )

    async def _flush_fails(*args: Any, **kwargs: Any) -> SimpleNamespace:
        calls.append("flush")
        raise RuntimeError("bad json")

    flush_service = MagicMock()
    flush_service.execute = AsyncMock(side_effect=_flush_fails)
    runner = TurnRunner(
        provider_selector=MagicMock(),
        session_manager=mock_sm,
        session_flush_service=flush_service,
        config=SimpleNamespace(
            memory=SimpleNamespace(
                flush_enabled=True,
                flush_pre_compaction=True,
                flush_timeout_seconds=0.25,
                flush_background_timeout_seconds=42.0,
            )
        ),
    )

    with patch("opensquilla.session.tokenizer.estimate_tokens", return_value=1000):
        await runner._maybe_preflight_compact("agent:ops:long-session", context_window)

    assert calls[:2] == ["checkpoint", "compact"]
    await asyncio.sleep(0)
    assert "flush" in calls
    mock_sm.compact.assert_awaited_once_with("agent:ops:long-session", context_window)


@pytest.mark.asyncio
async def test_preflight_checkpoint_failure_prevents_destructive_compaction() -> None:
    context_window = 1000
    entries = [_make_entry("early durable fact " + ("a" * 4000))]
    mock_sm = MagicMock()
    mock_sm.get_transcript = AsyncMock(return_value=entries)
    mock_sm.record_memory_checkpoint = AsyncMock(
        side_effect=RuntimeError("checkpoint write failed")
    )
    mock_sm.compact = AsyncMock(return_value="summary text")
    flush_service = MagicMock()
    flush_service.execute = AsyncMock(return_value=_flush_receipt())
    runner = TurnRunner(
        provider_selector=MagicMock(),
        session_manager=mock_sm,
        session_flush_service=flush_service,
        config=SimpleNamespace(
            memory=SimpleNamespace(
                flush_enabled=True,
                flush_pre_compaction=True,
                flush_timeout_seconds=0.25,
                flush_background_timeout_seconds=42.0,
            )
        ),
    )

    with (
        patch("opensquilla.session.tokenizer.estimate_tokens", return_value=1000),
        pytest.raises(RuntimeError, match="checkpoint write failed"),
    ):
        await runner._maybe_preflight_compact("agent:ops:long-session", context_window)

    flush_service.execute.assert_not_called()
    mock_sm.compact.assert_not_called()


@pytest.mark.asyncio
async def test_preflight_completed_event_reports_compaction_metadata(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    context_window = 1000
    entries = [_make_entry("a" * 4000)]
    sm = _ResultCompactionSessionManager(entries)
    events: list[tuple[str, dict[str, Any]]] = []
    monkeypatch.setattr(
        runtime_module,
        "notify_compaction",
        lambda session_key, **payload: events.append((session_key, payload)),
    )
    runner = TurnRunner(provider_selector=MagicMock(), session_manager=sm)

    with patch("opensquilla.session.tokenizer.estimate_tokens", return_value=1000):
        await runner._maybe_preflight_compact("user:session", context_window)

    assert sm.compact_with_result_calls == [("user:session", context_window, None)]
    assert [(key, payload["status"]) for key, payload in events] == [
        ("user:session", "started"),
        ("user:session", "observed"),
        ("user:session", "observed"),
        ("user:session", "completed"),
    ]
    compaction_ids = {payload.get("compaction_id") for _, payload in events}
    assert len(compaction_ids) == 1
    assert None not in compaction_ids
    assert events[0][1]["event"] == "compaction.triggered"
    assert events[1][1]["event"] == "compaction.chunk_summarized"
    assert events[2][1]["event"] == "compaction.summary_verified"
    completed = events[-1][1]
    assert completed["applied"] is True
    assert completed["durability"] == "durable"
    assert completed["user_visible"] is True
    assert completed["event"] == "compaction.persisted"
    assert completed["event_chain"] == [
        "compaction.triggered",
        "compaction.chunk_summarized",
        "compaction.summary_verified",
        "compaction.persisted",
    ]
    assert completed["coverage_status"] == "unknown"
    assert completed["chunk_count"] == 3
    assert completed["summary_source"] == "llm"
    assert completed["removed_count"] == 4
    assert completed["kept_count"] == 1
    assert completed["tokens_after"] == 200
    assert completed["remaining_budget_tokens"] == 800


@pytest.mark.asyncio
async def test_preflight_counts_tool_call_arguments_when_deciding_to_compact() -> None:
    context_window = 1000
    entries = [
        _make_assistant_tool_entry(
            "small visible answer",
            [
                {
                    "type": "tool_use",
                    "tool_use_id": "write-stale",
                    "name": "write_file",
                    "input": {
                        "path": "index.html",
                        "content": "x" * 5000,
                    },
                }
            ],
        )
    ]
    mock_sm = MagicMock()
    mock_sm.compact = AsyncMock(return_value="summary text")
    mock_sm.get_transcript = AsyncMock(return_value=entries)
    runner = TurnRunner(provider_selector=MagicMock(), session_manager=mock_sm)

    with patch(
        "opensquilla.session.tokenizer.estimate_tokens",
        side_effect=lambda text: max(1, len(str(text)) // 4),
    ):
        await runner._maybe_preflight_compact("user:session", context_window)

    mock_sm.compact.assert_called_once_with("user:session", context_window)


@pytest.mark.asyncio
async def test_preflight_counts_reasoning_content_when_deciding_to_compact() -> None:
    context_window = 1000
    entries = [
        _make_assistant_reasoning_entry(
            "small visible answer",
            "reasoning " + ("r" * 5000),
        )
    ]
    mock_sm = MagicMock()
    mock_sm.compact = AsyncMock(return_value="summary text")
    mock_sm.get_transcript = AsyncMock(return_value=entries)
    runner = TurnRunner(provider_selector=MagicMock(), session_manager=mock_sm)

    with patch(
        "opensquilla.session.tokenizer.estimate_tokens",
        side_effect=lambda text: max(1, len(str(text)) // 4),
    ):
        await runner._maybe_preflight_compact("user:session", context_window)

    mock_sm.compact.assert_called_once_with("user:session", context_window)


@pytest.mark.asyncio
async def test_preflight_starts_full_transcript_flush_without_blocking_compact() -> None:
    """Preflight starts full-coverage memory flush in the background."""

    context_window = 1000
    entries = [_make_entry("early durable fact " + ("a" * 4000))]
    calls: list[str] = []

    mock_sm = MagicMock()
    mock_sm.get_transcript = AsyncMock(return_value=entries)

    async def compact(session_key, context_window_tokens):
        calls.append("compact")
        return "summary text"

    mock_sm.compact = AsyncMock(side_effect=compact)

    flush_service = MagicMock()

    async def flush_execute(*args, **kwargs):
        calls.append("flush")
        return _flush_receipt()

    flush_service.execute = AsyncMock(side_effect=flush_execute)

    runner = TurnRunner(
        provider_selector=MagicMock(),
        session_manager=mock_sm,
        session_flush_service=flush_service,
        config=_flush_enabled_config(),
    )

    with patch("opensquilla.session.tokenizer.estimate_tokens", return_value=1000):
        await runner._maybe_preflight_compact("agent:ops:long-session", context_window)

    assert calls == ["compact"]
    await asyncio.sleep(0)
    assert "flush" in calls
    flush_service.execute.assert_awaited_once_with(
        entries,
        "agent:ops:long-session",
        agent_id="ops",
        message_window=0,
        segment_mode="auto",
        timeout=120.0,
        raw_capture_policy="required",
        turn_id=ANY,
        checkpoint_exists=False,
    )
    mock_sm.compact.assert_awaited_once_with("agent:ops:long-session", context_window)


@pytest.mark.parametrize(
    "receipt",
    [
        _flush_receipt(mode="raw", raw_reason="no_provider"),
        _flush_receipt(integrity_status="missing_chunks"),
        _flush_receipt(output_coverage_status="coverage_warning"),
        _flush_receipt(invalid_candidate_count=1),
        _flush_receipt(candidate_missing_ids=["candidate-1"]),
        _flush_receipt(obligation_missing_ids=["obligation-1"]),
    ],
)
@pytest.mark.asyncio
async def test_preflight_degraded_flush_receipts_do_not_block_compaction(
    receipt: SimpleNamespace,
) -> None:
    context_window = 1000
    entries = [_make_entry("early durable fact " + ("a" * 4000))]
    mock_sm = MagicMock()
    mock_sm.get_transcript = AsyncMock(return_value=entries)
    mock_sm.compact = AsyncMock(return_value="summary text")

    flush_service = MagicMock()
    flush_service.execute = AsyncMock(return_value=receipt)
    runner = TurnRunner(
        provider_selector=MagicMock(),
        session_manager=mock_sm,
        session_flush_service=flush_service,
        config=_flush_enabled_config(),
    )

    with patch("opensquilla.session.tokenizer.estimate_tokens", return_value=1000):
        await runner._maybe_preflight_compact("agent:ops:long-session", context_window)

    await asyncio.sleep(0)
    flush_service.execute.assert_awaited_once()
    mock_sm.compact.assert_awaited_once_with("agent:ops:long-session", context_window)


@pytest.mark.asyncio
async def test_preflight_strict_flush_receipt_skips_destructive_compaction() -> None:
    context_window = 1000
    entries = [_make_entry("early durable fact " + ("a" * 4000))]
    mock_sm = MagicMock()
    mock_sm.get_transcript = AsyncMock(return_value=entries)
    mock_sm.compact = AsyncMock(return_value="summary text")

    flush_service = MagicMock()
    flush_service.execute = AsyncMock(
        return_value=_flush_receipt(integrity_status="missing_chunks")
    )
    runner = TurnRunner(
        provider_selector=MagicMock(),
        session_manager=mock_sm,
        session_flush_service=flush_service,
        config=SimpleNamespace(
            memory=SimpleNamespace(
                flush_enabled=True,
                flush_pre_compaction=True,
                flush_timeout_seconds=0.25,
                flush_background_timeout_seconds=42.0,
                flush_compaction_requires_safe_receipt=True,
            )
        ),
    )

    with patch("opensquilla.session.tokenizer.estimate_tokens", return_value=1000):
        await runner._maybe_preflight_compact("agent:ops:long-session", context_window)

    flush_service.execute.assert_awaited_once()
    mock_sm.compact.assert_not_called()


@pytest.mark.asyncio
async def test_preflight_protect_flush_receipt_marks_degraded_forensic() -> None:
    context_window = 1000
    entries = [_make_entry("early durable fact " + ("a" * 4000))]
    sm = _ResultCompactionSessionManager(entries)
    flush_service = MagicMock()
    flush_service.execute = AsyncMock(
        return_value=_flush_receipt(integrity_status="missing_chunks")
    )
    runner = TurnRunner(
        provider_selector=MagicMock(),
        session_manager=sm,
        session_flush_service=flush_service,
        config=SimpleNamespace(
            memory=SimpleNamespace(
                flush_enabled=True,
                flush_pre_compaction=True,
                flush_timeout_seconds=0.25,
                flush_background_timeout_seconds=42.0,
                flush_compaction_safety_mode="protect",
            )
        ),
    )

    with patch("opensquilla.session.tokenizer.estimate_tokens", return_value=1000):
        await runner._maybe_preflight_compact("agent:ops:long-session", context_window)

    await asyncio.sleep(0)
    flush_service.execute.assert_awaited_once()
    assert sm.compact_with_result_calls == [("agent:ops:long-session", context_window, None)]
    assert sm.compact_with_result_kwargs[0]["flush_receipt_status"] == "degraded_forensic"


@pytest.mark.asyncio
async def test_preflight_compact_failure_uses_emergency_ephemeral_history_trim() -> None:
    session_key = "agent:ops:preflight-emergency"
    context_window = 1000
    entries = [
        TranscriptEntry(
            session_id="test-session-id",
            session_key=session_key,
            role="user" if index % 2 == 0 else "assistant",
            content=f"historic message {index} " + ("x" * 500),
            token_count=300,
        )
        for index in range(8)
    ]
    sm = _FailingResultCompactionSessionManager(entries)
    flush_service = MagicMock()
    flush_service.execute = AsyncMock(return_value=_flush_receipt(mode="raw"))
    runner = TurnRunner(
        provider_selector=MagicMock(),
        session_manager=sm,
        session_flush_service=flush_service,
        config=SimpleNamespace(
            memory=SimpleNamespace(
                flush_enabled=True,
                flush_pre_compaction=True,
                flush_timeout_seconds=0.25,
                flush_background_timeout_seconds=42.0,
                flush_compaction_safety_mode="protect",
            )
        ),
    )

    await runner._maybe_preflight_compact(session_key, context_window)

    class _HistoryCapture:
        provider = SimpleNamespace(provider_name="test")

        def __init__(self) -> None:
            self.history: list[Any] = []

        def set_history(self, history: list[Any]) -> None:
            self.history = history

    agent = _HistoryCapture()
    summary_context = await runner._load_history(agent, session_key, trim_last_user=False)

    assert sm.compact_with_result_calls == [(session_key, context_window, None)]
    assert len(await sm.get_transcript(session_key)) == len(entries)
    assert 0 < len(agent.history) < len(entries)
    assert summary_context is not None
    assert "emergency request-scoped compaction" in summary_context.lower()


@pytest.mark.asyncio
async def test_preflight_compact_failure_reports_emergency_without_failed_event(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    session_key = "agent:ops:preflight-emergency-event"
    context_window = 1000
    entries = [
        TranscriptEntry(
            session_id="test-session-id",
            session_key=session_key,
            role="user" if index % 2 == 0 else "assistant",
            content=f"historic message {index} " + ("x" * 500),
            token_count=300,
        )
        for index in range(8)
    ]
    sm = _FailingResultCompactionSessionManager(entries)
    events: list[tuple[str, dict[str, Any]]] = []
    monkeypatch.setattr(
        runtime_module,
        "notify_compaction",
        lambda session_key, **payload: events.append((session_key, payload)),
    )
    runner = TurnRunner(provider_selector=MagicMock(), session_manager=sm)
    runner._compaction_failures[session_key] = runtime_module._CompactionFailureState(count=1)

    await runner._maybe_preflight_compact(session_key, context_window)

    statuses = [payload["status"] for _, payload in events]
    assert statuses == ["started", "emergency_ephemeral"]
    emergency = events[-1][1]
    assert emergency["applied"] is True
    assert emergency["durability"] == "request_scoped"
    assert emergency["user_visible"] is True
    assert emergency["reason"] == "compact_failed"
    assert emergency["flush_receipt_status"] == "emergency_ephemeral"
    assert runner._compaction_failures[session_key].count == 2


@pytest.mark.asyncio
async def test_preflight_open_circuit_still_uses_request_scoped_emergency_trim(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    session_key = "agent:ops:preflight-open-circuit"
    context_window = 1000
    entries = [
        TranscriptEntry(
            session_id="test-session-id",
            session_key=session_key,
            role="user" if index % 2 == 0 else "assistant",
            content=f"historic message {index} " + ("x" * 500),
            token_count=300,
        )
        for index in range(8)
    ]
    mock_sm = MagicMock()
    mock_sm.compact = AsyncMock()
    mock_sm.get_transcript = AsyncMock(return_value=entries)
    events: list[tuple[str, dict[str, Any]]] = []
    monkeypatch.setattr(
        runtime_module,
        "notify_compaction",
        lambda session_key, **payload: events.append((session_key, payload)),
    )
    runner = TurnRunner(provider_selector=MagicMock(), session_manager=mock_sm)
    runner._compaction_failures[session_key] = runtime_module._CompactionFailureState(
        count=3,
        opened_at=runtime_module.time.monotonic(),
    )

    await runner._maybe_preflight_compact(session_key, context_window)

    mock_sm.compact.assert_not_awaited()
    assert [payload["status"] for _, payload in events] == ["emergency_ephemeral"]
    emergency = events[-1][1]
    assert emergency["reason"] == "durable_compaction_circuit_open"
    assert emergency["applied"] is True
    assert emergency["durability"] == "request_scoped"
    assert runner._compaction_failures[session_key].count == 3


@pytest.mark.asyncio
async def test_preflight_empty_summary_uses_emergency_ephemeral_history_trim() -> None:
    session_key = "agent:ops:preflight-empty-summary"
    context_window = 1000
    entries = [
        TranscriptEntry(
            session_id="test-session-id",
            session_key=session_key,
            role="user" if index % 2 == 0 else "assistant",
            content=f"historic message {index} " + ("x" * 500),
            token_count=300,
        )
        for index in range(8)
    ]

    class _EmptySummarySessionManager(_ResultCompactionSessionManager):
        async def compact_with_result(
            self,
            session_key: str,
            context_window_tokens: int,
            config: object | None = None,
            **kwargs,
        ) -> SimpleNamespace:
            self.compact_with_result_calls.append((session_key, context_window_tokens, config))
            self.compact_with_result_kwargs.append(dict(kwargs))
            return SimpleNamespace(
                summary="",
                kept_entries=entries,
                removed_count=0,
                chunks_processed=0,
                summary_source="skipped",
                tokens_before=2400,
                tokens_after=2400,
                remaining_budget_tokens=0,
            )

    sm = _EmptySummarySessionManager(entries)
    runner = TurnRunner(provider_selector=MagicMock(), session_manager=sm)

    await runner._maybe_preflight_compact(session_key, context_window)

    class _HistoryCapture:
        provider = SimpleNamespace(provider_name="test")

        def __init__(self) -> None:
            self.history: list[Any] = []

        def set_history(self, history: list[Any]) -> None:
            self.history = history

    agent = _HistoryCapture()
    summary_context = await runner._load_history(agent, session_key, trim_last_user=False)

    assert sm.compact_with_result_calls == [(session_key, context_window, None)]
    assert len(await sm.get_transcript(session_key)) == len(entries)
    assert 0 < len(agent.history) < len(entries)
    assert summary_context is not None
    assert "emergency request-scoped compaction" in summary_context.lower()


@pytest.mark.asyncio
async def test_preflight_stale_preimage_skip_does_not_use_emergency_trim(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    session_key = "agent:ops:preflight-stale-summary"
    context_window = 1000
    entries = [
        TranscriptEntry(
            session_id="test-session-id",
            session_key=session_key,
            role="user" if index % 2 == 0 else "assistant",
            content=f"historic message {index} " + ("x" * 500),
            token_count=300,
        )
        for index in range(8)
    ]
    sm = _StaleResultCompactionSessionManager(entries)
    events: list[tuple[str, dict[str, Any]]] = []
    monkeypatch.setattr(
        runtime_module,
        "notify_compaction",
        lambda session_key, **payload: events.append((session_key, payload)),
    )
    runner = TurnRunner(provider_selector=MagicMock(), session_manager=sm)
    runner._compaction_failures[session_key] = runtime_module._CompactionFailureState(count=1)

    await runner._maybe_preflight_compact(session_key, context_window)

    assert sm.compact_with_result_calls == [(session_key, context_window, None)]
    assert runner.has_compacted_this_turn(session_key) is False
    assert runner._compaction_failures[session_key].count == 1
    skipped = [payload for _, payload in events if payload.get("status") == "skipped"]
    assert skipped[-1]["reason"] == "stale_preimage"
    assert skipped[-1]["applied"] is False
    assert skipped[-1]["durability"] == "none"
    assert skipped[-1]["user_visible"] is False

    class _HistoryCapture:
        provider = SimpleNamespace(provider_name="test")

        def __init__(self) -> None:
            self.history: list[Any] = []

        def set_history(self, history: list[Any]) -> None:
            self.history = history

    agent = _HistoryCapture()
    summary_context = await runner._load_history(agent, session_key, trim_last_user=False)
    assert summary_context is None
    assert len(agent.history) == len(entries)


@pytest.mark.asyncio
async def test_preflight_backfilled_flush_receipt_allows_compact() -> None:
    context_window = 1000
    entries = [_make_entry("early durable fact " + ("a" * 4000))]
    mock_sm = MagicMock()
    mock_sm.get_transcript = AsyncMock(return_value=entries)
    mock_sm.compact = AsyncMock(return_value="summary text")

    flush_service = MagicMock()
    flush_service.execute = AsyncMock(return_value=_flush_receipt(obligation_status="backfilled"))
    runner = TurnRunner(
        provider_selector=MagicMock(),
        session_manager=mock_sm,
        session_flush_service=flush_service,
        config=_flush_enabled_config(),
    )

    with patch("opensquilla.session.tokenizer.estimate_tokens", return_value=1000):
        await runner._maybe_preflight_compact("agent:ops:long-session", context_window)

    await asyncio.sleep(0)
    flush_service.execute.assert_awaited_once()
    mock_sm.compact.assert_awaited_once_with("agent:ops:long-session", context_window)


@pytest.mark.asyncio
async def test_preflight_uses_background_timeout_for_flush_service() -> None:
    context_window = 1000
    entries = [_make_entry("early durable fact " + ("a" * 4000))]
    mock_sm = MagicMock()
    mock_sm.get_transcript = AsyncMock(return_value=entries)
    mock_sm.compact = AsyncMock(return_value="summary text")

    flush_service = MagicMock()
    flush_service.execute = AsyncMock(return_value=_flush_receipt())
    runner = TurnRunner(
        provider_selector=MagicMock(),
        session_manager=mock_sm,
        session_flush_service=flush_service,
        config=SimpleNamespace(
            memory=SimpleNamespace(
                flush_enabled=True,
                flush_pre_compaction=True,
                flush_timeout_seconds=0.25,
                flush_background_timeout_seconds=42.0,
            )
        ),
    )

    with patch("opensquilla.session.tokenizer.estimate_tokens", return_value=1000):
        await runner._maybe_preflight_compact("agent:ops:long-session", context_window)

    await asyncio.sleep(0)
    assert flush_service.execute.await_args.kwargs["timeout"] == 42.0
    mock_sm.compact.assert_awaited_once_with("agent:ops:long-session", context_window)


@pytest.mark.asyncio
async def test_preflight_flush_grace_timeout_does_not_block_compaction() -> None:
    context_window = 1000
    entries = [_make_entry("early durable fact " + ("a" * 4000))]
    mock_sm = MagicMock()
    mock_sm.get_transcript = AsyncMock(return_value=entries)
    mock_sm.compact = AsyncMock(return_value="summary text")

    flush_service = MagicMock()

    async def slow_flush(*_args, **_kwargs):
        import asyncio

        await asyncio.sleep(0.05)
        return _flush_receipt()

    flush_service.execute = AsyncMock(side_effect=slow_flush)
    runner = TurnRunner(
        provider_selector=MagicMock(),
        session_manager=mock_sm,
        session_flush_service=flush_service,
        config=SimpleNamespace(
            memory=SimpleNamespace(
                flush_enabled=True,
                flush_pre_compaction=True,
                flush_timeout_seconds=0.001,
                flush_background_timeout_seconds=42.0,
            )
        ),
    )

    with patch("opensquilla.session.tokenizer.estimate_tokens", return_value=1000):
        await runner._maybe_preflight_compact("agent:ops:long-session", context_window)

    await asyncio.sleep(0)
    assert flush_service.execute.await_args.kwargs["timeout"] == 42.0
    mock_sm.compact.assert_awaited_once_with("agent:ops:long-session", context_window)
    await asyncio.sleep(0.06)


@pytest.mark.asyncio
async def test_preflight_memory_flush_disabled_compacts_without_flush() -> None:
    context_window = 1000
    entries = [_make_entry("early durable fact " + ("a" * 4000))]
    mock_sm = MagicMock()
    mock_sm.get_transcript = AsyncMock(return_value=entries)
    mock_sm.compact = AsyncMock(return_value="summary text")

    flush_service = MagicMock()
    flush_service.execute = AsyncMock(return_value=_flush_receipt())
    runner = TurnRunner(
        provider_selector=MagicMock(),
        session_manager=mock_sm,
        session_flush_service=flush_service,
        config=SimpleNamespace(
            memory=SimpleNamespace(flush_enabled=False, flush_timeout_seconds=0.25)
        ),
    )

    with patch("opensquilla.session.tokenizer.estimate_tokens", return_value=1000):
        await runner._maybe_preflight_compact("agent:ops:long-session", context_window)

    flush_service.execute.assert_not_called()
    mock_sm.compact.assert_awaited_once_with("agent:ops:long-session", context_window)


@pytest.mark.parametrize("value", ["0", "false", "no", "off"])
@pytest.mark.asyncio
async def test_preflight_env_flush_disabled_compacts_without_flush(
    monkeypatch: pytest.MonkeyPatch,
    value: str,
) -> None:
    monkeypatch.setenv("OPENSQUILLA_SESSION_FLUSH", value)
    context_window = 1000
    entries = [_make_entry("early durable fact " + ("a" * 4000))]
    mock_sm = MagicMock()
    mock_sm.get_transcript = AsyncMock(return_value=entries)
    mock_sm.compact = AsyncMock(return_value="summary text")

    flush_service = MagicMock()
    flush_service.execute = AsyncMock(return_value=_flush_receipt())
    runner = TurnRunner(
        provider_selector=MagicMock(),
        session_manager=mock_sm,
        session_flush_service=flush_service,
        config=SimpleNamespace(
            memory=SimpleNamespace(flush_enabled=True, flush_timeout_seconds=0.25)
        ),
    )

    with patch("opensquilla.session.tokenizer.estimate_tokens", return_value=1000):
        await runner._maybe_preflight_compact("agent:ops:long-session", context_window)

    flush_service.execute.assert_not_called()
    mock_sm.compact.assert_awaited_once_with("agent:ops:long-session", context_window)


@pytest.mark.asyncio
async def test_preflight_flush_service_unavailable_does_not_block_compaction() -> None:
    context_window = 1000
    entries = [_make_entry("early durable fact " + ("a" * 4000))]
    mock_sm = MagicMock()
    mock_sm.get_transcript = AsyncMock(return_value=entries)
    mock_sm.compact = AsyncMock(return_value="summary text")

    runner = TurnRunner(
        provider_selector=MagicMock(),
        session_manager=mock_sm,
        session_flush_service=None,
        config=SimpleNamespace(
            memory=SimpleNamespace(flush_enabled=True, flush_timeout_seconds=0.25)
        ),
    )

    with patch("opensquilla.session.tokenizer.estimate_tokens", return_value=1000):
        await runner._maybe_preflight_compact("agent:ops:long-session", context_window)

    mock_sm.compact.assert_awaited_once_with("agent:ops:long-session", context_window)


@pytest.mark.asyncio
async def test_preflight_passes_provider_backed_compaction_config_after_flush() -> None:
    context_window = 1000
    entries = [_make_entry("early durable fact " + ("a" * 4000))]
    captured_configs: list[CompactionConfig | None] = []

    mock_sm = MagicMock()
    mock_sm.get_transcript = AsyncMock(return_value=entries)

    async def compact(session_key, context_window_tokens, config=None):
        captured_configs.append(config)
        return "summary text"

    mock_sm.compact = AsyncMock(side_effect=compact)

    flush_service = MagicMock()
    flush_service.execute = AsyncMock(return_value=_flush_receipt())

    runner = TurnRunner(
        provider_selector=MagicMock(),
        session_manager=mock_sm,
        session_flush_service=flush_service,
        config=SimpleNamespace(
            compaction=SimpleNamespace(enabled=True, model=None, timeout_seconds=17.5)
        ),
    )
    provider = _FakeCompactionProvider(model="provider/model")

    with patch("opensquilla.session.tokenizer.estimate_tokens", return_value=1000):
        await runner._maybe_preflight_compact(
            "agent:ops:long-session",
            context_window,
            compaction_provider=provider,
            compaction_model="routed/model",
        )

    assert len(captured_configs) == 1
    config = captured_configs[0]
    assert isinstance(config, CompactionConfig)
    assert config.api_key == "preflight-provider-key"
    assert config.model == "routed/model"
    assert config.base_url == "https://openrouter.ai/api/v1"
    assert config.timeout_seconds == 17.5


@pytest.mark.asyncio
async def test_preflight_passes_profile_config_without_provider_or_model() -> None:
    context_window = 1000
    entries = [_make_entry("early durable fact " + ("a" * 4000))]
    sm = _ResultCompactionSessionManager(entries)
    runner = TurnRunner(
        provider_selector=MagicMock(),
        session_manager=sm,
        config=SimpleNamespace(
            compaction=SimpleNamespace(
                enabled=False,
                compaction_profile="coding",
                protected_recent_messages=7,
                timeout_seconds=17.5,
            )
        ),
    )

    with patch("opensquilla.session.tokenizer.estimate_tokens", return_value=1000):
        await runner._maybe_preflight_compact("agent:ops:long-session", context_window)

    assert len(sm.compact_with_result_calls) == 1
    config = sm.compact_with_result_calls[0][2]
    assert isinstance(config, CompactionConfig)
    assert config.model is None
    assert config.api_key == ""
    assert config.compaction_profile == "coding"
    assert config.protected_recent_messages == 7
    assert config.timeout_seconds == 17.5


@pytest.mark.asyncio
async def test_preflight_keeps_legacy_compact_manager_compatible() -> None:
    context_window = 1000
    entries = [_make_entry("early durable fact " + ("a" * 4000))]
    calls: list[tuple[str, int]] = []

    mock_sm = MagicMock()
    mock_sm.get_transcript = AsyncMock(return_value=entries)

    async def compact(session_key, context_window_tokens):
        calls.append((session_key, context_window_tokens))
        return "summary text"

    mock_sm.compact = AsyncMock(side_effect=compact)

    runner = TurnRunner(
        provider_selector=MagicMock(),
        session_manager=mock_sm,
        config=SimpleNamespace(
            compaction=SimpleNamespace(enabled=True, model=None, timeout_seconds=17.5)
        ),
    )

    with patch("opensquilla.session.tokenizer.estimate_tokens", return_value=1000):
        await runner._maybe_preflight_compact(
            "agent:ops:long-session",
            context_window,
            compaction_provider=_FakeCompactionProvider(model="provider/model"),
            compaction_model="routed/model",
        )

    assert calls == [("agent:ops:long-session", context_window)]


@pytest.mark.asyncio
async def test_run_falls_back_to_generic_preflight_after_t3_flush_failed() -> None:
    selector = _FakeProviderSelector()
    runner = TurnRunner(provider_selector=selector, config=GatewayConfig())
    seen: dict[str, object] = {}

    async def fake_t3(session_key, turn, context_window_tokens, **kwargs):
        seen["t3_session_key"] = session_key
        return "flush_failed"

    async def spy_preflight(session_key, context_window_tokens, **kwargs):
        seen["preflight_session_key"] = session_key
        seen.update(kwargs)

    runner._maybe_compact_on_t3_upgrade = fake_t3  # type: ignore[method-assign]
    runner._maybe_preflight_compact = spy_preflight  # type: ignore[method-assign]
    tool_ctx = ToolContext(is_owner=True, caller_kind=CallerKind.CLI)

    async for _ in runner.run(
        "hello",
        "agent:main:abc123",
        tool_context=tool_ctx,
        model="routed/model",
    ):
        pass

    assert seen["t3_session_key"] == "agent:main:abc123"
    assert seen["preflight_session_key"] == "agent:main:abc123"
    assert seen["compaction_model"] == "routed/model"


@pytest.mark.asyncio
async def test_run_forwards_routed_provider_and_model_to_preflight() -> None:
    selector = _FakeProviderSelector()
    runner = TurnRunner(
        provider_selector=selector,
        config=GatewayConfig(),
        model_catalog=ModelCatalog(),
    )
    seen: dict[str, object] = {}

    async def spy_preflight(session_key, context_window_tokens, **kwargs):
        seen["session_key"] = session_key
        seen["context_window_tokens"] = context_window_tokens
        seen.update(kwargs)

    runner._maybe_preflight_compact = spy_preflight  # type: ignore[method-assign]
    tool_ctx = ToolContext(is_owner=True, caller_kind=CallerKind.CLI)

    async for _ in runner.run(
        "hello",
        "agent:main:abc123",
        tool_context=tool_ctx,
        model="z-ai/glm-5.1",
    ):
        pass

    assert seen["session_key"] == "agent:main:abc123"
    # Conservative-min static fallback for glm-5.1 (formerly 202_752 under the
    # z-ai/ spelling; merged with the bare glm-5.1 entry's 200_000 window).
    assert seen["context_window_tokens"] == 200_000
    assert seen["compaction_model"] == "z-ai/glm-5.1"
    assert getattr(seen["compaction_provider"], "model") == "z-ai/glm-5.1"
    assert selector.override_calls == []
    assert selector.clone_instance.override_calls[-1] == "z-ai/glm-5.1"


@pytest.mark.asyncio
async def test_preflight_compact_called_with_correct_args() -> None:
    """compact() is called with (session_key, context_window_tokens)."""
    entries = [_make_entry("content")]

    mock_sm = MagicMock()
    mock_sm.compact = AsyncMock(return_value="")
    mock_sm.get_transcript = AsyncMock(return_value=entries)

    runner = TurnRunner(provider_selector=MagicMock(), session_manager=mock_sm)

    with patch("opensquilla.session.tokenizer.estimate_tokens", return_value=90_000):
        await runner._maybe_preflight_compact("user:long-session", 100_000)

    mock_sm.compact.assert_called_once_with("user:long-session", 100_000)


@pytest.mark.asyncio
async def test_preflight_exactly_at_threshold_does_not_compact() -> None:
    """Transcript at exactly the threshold (not exceeding) → no compaction."""
    context_window = 1000
    threshold = int(context_window * 0.85)  # default threshold: 850

    entries = [_make_entry("a")]

    mock_sm = MagicMock()
    mock_sm.compact = AsyncMock()
    mock_sm.get_transcript = AsyncMock(return_value=entries)

    runner = TurnRunner(provider_selector=MagicMock(), session_manager=mock_sm)

    with patch("opensquilla.session.tokenizer.estimate_tokens", return_value=threshold):
        await runner._maybe_preflight_compact("user:session", context_window)

    mock_sm.compact.assert_not_called()


@pytest.mark.asyncio
async def test_preflight_integration_with_real_session_manager(session_mgr, tmp_path) -> None:
    """Integration: pre-flight with real SessionManager compacts when over threshold."""
    mgr = session_mgr
    key = "user:preflight-test"
    await mgr.create(key)

    # Seed transcript entries
    await mgr.append_message(key, role="user", content="message one")
    await mgr.append_message(key, role="assistant", content="reply one")
    await mgr.append_message(key, role="user", content="message two")

    runner = TurnRunner(provider_selector=MagicMock(), session_manager=mgr)

    # Patch compact_with_result() to verify the metadata-preserving path is used.
    original_compact_with_result = mgr.compact_with_result
    compact_calls: list[dict[str, object]] = []

    async def _spy_compact_with_result(
        session_key,
        context_window_tokens,
        config=None,
        *,
        compaction_id=None,
        trigger_reason=None,
        flush_receipt_status=None,
    ):
        compact_calls.append(
            {
                "session_key": session_key,
                "context_window_tokens": context_window_tokens,
                "compaction_id": compaction_id,
                "trigger_reason": trigger_reason,
                "flush_receipt_status": flush_receipt_status,
            }
        )
        return await original_compact_with_result(
            session_key,
            context_window_tokens,
            config,
            compaction_id=compaction_id,
            trigger_reason=trigger_reason,
            flush_receipt_status=flush_receipt_status,
        )

    mgr.compact_with_result = _spy_compact_with_result  # type: ignore[method-assign]

    # Force all tokens to exceed the default threshold (100 * 0.85 = 85)
    with patch("opensquilla.session.tokenizer.estimate_tokens", return_value=1000):
        await runner._maybe_preflight_compact(key, 100)

    # compact_with_result() was invoked with the correct args.
    assert len(compact_calls) == 1
    assert compact_calls[0]["session_key"] == key
    assert compact_calls[0]["context_window_tokens"] == 100
    assert compact_calls[0]["compaction_id"]
    assert compact_calls[0]["trigger_reason"] == "preflight"
    assert compact_calls[0]["flush_receipt_status"] == "not_required"
    receipts = await mgr.storage.list_memory_durable_receipts(
        session_key=key,
        status="checkpoint_saved",
        limit=1,
    )
    assert receipts
    assert receipts[0].scope == "checkpoint"
    assert receipts[0].status == "checkpoint_saved"
    assert receipts[0].source_path
    assert (tmp_path / receipts[0].source_path).exists()


@pytest.mark.asyncio
async def test_preflight_compaction_circuit_breaker_retries_after_cooldown() -> None:
    context_window = 1000
    entries = [_make_entry("a" * 4000)]

    mock_sm = MagicMock()
    mock_sm.compact = AsyncMock(side_effect=RuntimeError("compact failed"))
    mock_sm.get_transcript = AsyncMock(return_value=entries)

    runner = TurnRunner(provider_selector=MagicMock(), session_manager=mock_sm)

    with patch("opensquilla.session.tokenizer.estimate_tokens", return_value=1000):
        for _ in range(4):
            await runner._maybe_preflight_compact("user:session", context_window)
            runner.clear_compaction_turn_state("user:session")

    assert mock_sm.compact.await_count == 3

    runner._compaction_failures["user:session"].opened_at = 0.0
    with (
        patch("opensquilla.session.tokenizer.estimate_tokens", return_value=1000),
        patch("opensquilla.engine.runtime.time.monotonic", return_value=999.0),
    ):
        await runner._maybe_preflight_compact("user:session", context_window)

    assert mock_sm.compact.await_count == 4