"""WS-04 M1 — session-lifecycle reliability tests.

Covers:
- SL-1: the configured session-lock timeout is enforced on lock ACQUISITION in
  ``turn_logic.run_turn`` / ``run_turn_streaming`` and
  ``agent_harness.run_agent_turn_streaming`` (held lock → graceful failure within
  the timeout; happy path unaffected). main.py's advertised
  ``limits.session_lock_timeout_ms`` derives from the backend constant.
- SL-2: an in-flight agent turn is protected from TTL eviction (last_seen is
  refreshed at turn entry so a mid-turn eviction sweep cannot drop the record a
  tool is about to read).
- SL-3: the three alternate loaders (fixture/text/repo) enforce the session cap,
  and /api/load_graph maps the resulting OverflowError to 503 for each input_kind.
- SL-4: save_session logs a warning (instead of silently swallowing) when the
  checkpoint read fails.
- SEC-05: load_repo_session rejects an out-of-base path with a GENERIC message
  (no absolute path leaked) while logging the resolved path server-side.

Every session-mutating test cleans up both ``session_store`` and ``session_locks``
via the autouse fixture below — no session leaks.
"""

from __future__ import annotations

import logging
import time
from collections.abc import AsyncIterator, Iterator
from pathlib import Path
from types import SimpleNamespace
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch

import httpx
import pytest
from loosecanvas.config import ALLOWED_BASE_DIR
from loosecanvas.contracts import SceneState, SourceSnapshot
from loosecanvas.graph_repository import LooseGraphRepository
from loosecanvas.turn_logic import (
    SessionRecord,
    TurnResult,
    UndoHistory,
    _get_lock,
    load_fixture_session,
    load_repo_session,
    load_text_session,
    run_turn,
    run_turn_streaming,
    save_session,
    session_locks,
    session_store,
)

_REPO_ROOT = Path(__file__).resolve().parents[1]
_SMALL_GRAPH = _REPO_ROOT / "fixtures" / "small_graph.json"
_FIXTURE_PATH = str(_SMALL_GRAPH.resolve())
_NO_SELECTION: dict[str, Any] = {"kind": "none", "id": ""}


@pytest.fixture(autouse=True)
def _isolate_sessions() -> Iterator[None]:
    session_store.clear()
    session_locks.clear()
    yield
    session_store.clear()
    session_locks.clear()


def _fresh_record(session_id: str) -> SessionRecord:
    """A minimal, non-stale SessionRecord (last_seen=now) for cap-fill tests."""
    repo = LooseGraphRepository.from_extraction([], [], [], SourceSnapshot())
    now = time.time()
    return SessionRecord(
        session_id=session_id,
        graph=repo,
        scene=SceneState(),
        history=[],
        graph_version=1,
        created_at=now,
        last_seen=now,
        patch_id=0,
        render_scene_id="",
        undo=UndoHistory(),
    )


def _fill_store_to_cap(max_sessions: int) -> None:
    """Fill session_store with ``max_sessions`` fresh (non-stale) sessions."""
    for i in range(max_sessions):
        sid = f"cap-fill-{i}"
        session_store[sid] = _fresh_record(sid)


class _StubGenerator:
    """Stand-in for the ScenePlanGenerator; never reached when the lock blocks."""

    def __init__(self) -> None:
        self.calls = 0

    async def generate(self, context: Any) -> Any:  # pragma: no cover - not reached
        self.calls += 1
        raise AssertionError("generator should not run while the lock is held")


# ── SL-1: session-lock timeout enforced on acquisition ────────────────────────


async def test_run_turn_lock_timeout_returns_failed(monkeypatch: Any) -> None:
    """A held session lock makes the next ``run_turn`` fail gracefully within the timeout."""
    monkeypatch.setattr("loosecanvas.turn_logic._SESSION_LOCK_TIMEOUT_SECONDS", 0.05)
    sid, _ = await load_fixture_session(fixture_path=_SMALL_GRAPH)
    # The per-session lock is created lazily inside the turn path; pre-create it so
    # the test can hold it while the turn tries (and times out) to acquire it.
    lock = _get_lock(sid)
    await lock.acquire()
    try:
        started = time.monotonic()
        result = await run_turn(sid, _NO_SELECTION, "highlight x", _StubGenerator())
        elapsed = time.monotonic() - started
    finally:
        lock.release()

    assert isinstance(result, TurnResult)
    assert result.status == "failed"
    assert any("lock timeout" in w for w in result.warnings)
    # Returned within a small multiple of the (tiny) timeout, not blocked forever.
    assert elapsed < 2.0


async def test_run_turn_streaming_lock_timeout_yields_failed(monkeypatch: Any) -> None:
    """A held lock makes ``run_turn_streaming`` yield a single failed TurnResult."""
    monkeypatch.setattr("loosecanvas.turn_logic._SESSION_LOCK_TIMEOUT_SECONDS", 0.05)
    sid, _ = await load_fixture_session(fixture_path=_SMALL_GRAPH)
    lock = _get_lock(sid)
    await lock.acquire()
    try:
        events = [
            ev
            async for ev in run_turn_streaming(
                sid, _NO_SELECTION, "highlight x", _StubGenerator()
            )
        ]
    finally:
        lock.release()

    results = [ev for ev in events if isinstance(ev, TurnResult)]
    assert len(results) == 1
    assert results[0].status == "failed"
    assert any("lock timeout" in w for w in results[0].warnings)


async def test_lock_released_after_timeout(monkeypatch: Any) -> None:
    """After a timeout the lock acquisition is NOT leaked (lock stays free)."""
    monkeypatch.setattr("loosecanvas.turn_logic._SESSION_LOCK_TIMEOUT_SECONDS", 0.05)
    sid, _ = await load_fixture_session(fixture_path=_SMALL_GRAPH)
    lock = _get_lock(sid)
    await lock.acquire()
    try:
        result = await run_turn(sid, _NO_SELECTION, "highlight x", _StubGenerator())
        assert result.status == "failed"
    finally:
        lock.release()
    # The failed turn must not have acquired the lock for itself.
    assert not lock.locked()


def test_config_lock_timeout_derives_from_constant() -> None:
    """main.py advertises the lock timeout derived from the backend constant."""
    import loosecanvas.main as main_module
    from loosecanvas.turn_logic import _SESSION_LOCK_TIMEOUT_SECONDS

    advertised = main_module._CONFIG_JSON["limits"]["session_lock_timeout_ms"]
    assert advertised == int(_SESSION_LOCK_TIMEOUT_SECONDS * 1000)


# ── SL-2: in-flight turn protected from TTL eviction ──────────────────────────


class _EvictMidTurnAgent:
    """Fake agent whose ``astream`` fires an eviction sweep mid-turn, then reads
    the session_store via a tool — crashing with KeyError if the record was
    evicted (i.e. if last_seen was not refreshed at turn entry)."""

    def __init__(self, ttl_seconds: float) -> None:
        self._ttl_seconds = ttl_seconds
        self.read_session_id: str | None = None

    async def astream(
        self, _inputs: Any, *, stream_mode: Any = None, config: Any = None
    ) -> AsyncIterator[tuple[Any, dict[str, Any]]]:
        from loosecanvas.agent_tools import get_turn_context
        from loosecanvas.turn_logic import _evict_stale_sessions

        # Mid-turn: a TTL sweep runs. With a low TTL, any record whose last_seen
        # predates the turn would be evicted here.
        _evict_stale_sessions()

        # A tool now reads the session_store keyed by the live context's session
        # id — KeyError if the sweep dropped it.
        ctx = get_turn_context()
        self.read_session_id = ctx.session_id
        _record = session_store[ctx.session_id]  # noqa: F841 - the read is the point

        yield (
            SimpleNamespace(content="ok", tool_call_chunks=None),
            {"langgraph_node": "model"},
        )


async def test_inflight_turn_survives_eviction_sweep(monkeypatch: Any) -> None:
    """A mid-turn eviction sweep must NOT drop the in-flight session (SL-2)."""
    from loosecanvas.agent_harness import FinalEvent, run_agent_turn_streaming

    # Low TTL so the aged record below would be evicted UNLESS turn entry refreshes
    # it — but not so low (was 0.0001s = 100us) that the gap between the refresh and
    # the mid-turn sweep itself trips eviction (a slow full-suite run made that a
    # flake: KeyError on the in-flight session). 1.0s gives the refreshed session
    # (last_seen ~= now) ample margin over the sub-ms sweep, while the 10s-aged
    # record below is still far past it — so the refresh remains the only thing that
    # saves the session, preserving the SL-2 contract under test.
    monkeypatch.setattr("loosecanvas.turn_logic._SESSION_TTL_SECONDS", 1.0)
    sid, _ = await load_fixture_session(fixture_path=_SMALL_GRAPH)
    # Age the record so its last_seen predates "now" by more than the TTL.
    session_store[sid].last_seen = time.time() - 10.0

    agent = _EvictMidTurnAgent(ttl_seconds=0.0001)
    events = [
        ev
        async for ev in run_agent_turn_streaming(sid, _NO_SELECTION, "hi", agent=agent)
    ]
    finals = [ev for ev in events if isinstance(ev, FinalEvent)]
    assert len(finals) == 1
    assert agent.read_session_id == sid
    # Survived: the session is still present after the turn.
    assert sid in session_store


# ── SL-3: capacity enforcement across alternate loaders ───────────────────────


async def test_load_fixture_session_enforces_cap(monkeypatch: Any) -> None:
    monkeypatch.setattr("loosecanvas.turn_logic._MAX_SESSIONS", 3)
    _fill_store_to_cap(3)
    with pytest.raises(OverflowError, match="session_cap"):
        await load_fixture_session(fixture_path=_SMALL_GRAPH)


async def test_load_text_session_enforces_cap(monkeypatch: Any) -> None:
    monkeypatch.setattr("loosecanvas.turn_logic._MAX_SESSIONS", 3)
    _fill_store_to_cap(3)
    # The cap guard fires before extraction, so the client is never called.
    fake_client = MagicMock()
    with pytest.raises(OverflowError, match="session_cap"):
        await load_text_session("some prose", "Title", fake_client)


async def test_load_repo_session_enforces_cap(monkeypatch: Any) -> None:
    monkeypatch.setattr("loosecanvas.turn_logic._MAX_SESSIONS", 3)
    _fill_store_to_cap(3)
    repo = LooseGraphRepository.from_extraction([], [], [], SourceSnapshot())
    from loosecanvas.extraction_pipeline import ExtractionResult

    fake_result = ExtractionResult(
        graph_id="cap-test-graph-id",
        scene=SceneState(),
        pathways=[],
        import_report={},
    )
    with patch("loosecanvas.turn_logic.ExtractionOrchestrator") as mock_cls:
        mock_instance = MagicMock()
        mock_instance.run = AsyncMock(return_value=fake_result)
        mock_cls.return_value = mock_instance
        with (
            patch("loosecanvas.turn_logic.get_repository", return_value=repo),
            pytest.raises(OverflowError, match="session_cap"),
        ):
            await load_repo_session(ALLOWED_BASE_DIR)


@pytest.mark.parametrize(
    ("input_kind", "loader_name", "body"),
    [
        (
            "fixture",
            "load_fixture_session",
            {"input_kind": "fixture", "fixture_path": _FIXTURE_PATH},
        ),
        (
            "text",
            "load_text_session",
            {"input_kind": "text", "text": "some prose", "title": "T"},
        ),
        (
            "repo",
            "load_repo_session",
            {"input_kind": "repo", "repo_path": str(ALLOWED_BASE_DIR)},
        ),
    ],
)
async def test_api_load_graph_returns_503_at_cap(
    input_kind: str, loader_name: str, body: dict[str, Any]
) -> None:
    """/api/load_graph maps a loader OverflowError to 503 for every input_kind."""
    import loosecanvas.main as main_module

    async def _raise_cap(*_args: Any, **_kwargs: Any) -> Any:
        raise OverflowError("session_cap")

    with patch.object(main_module, loader_name, new=_raise_cap):
        app = main_module.create_api()
        async with httpx.AsyncClient(
            transport=httpx.ASGITransport(app=app), base_url="http://test"
        ) as client:
            resp = await client.post("/api/load_graph", json=body)

    assert resp.status_code == 503
    assert "session cap" in resp.json()["detail"]


# ── SL-4: save_session logs (not swallows) a checkpoint-read failure ──────────


async def test_save_session_logs_checkpoint_read_failure(
    monkeypatch: Any, caplog: pytest.LogCaptureFixture
) -> None:
    """A failing checkpoint read yields an empty-memory snapshot AND a warning."""
    sid, _ = await load_fixture_session(fixture_path=_SMALL_GRAPH)

    class _BoomApp:
        async def aget_state(self, _config: Any) -> Any:
            raise RuntimeError("checkpoint read boom")

    monkeypatch.setattr("loosecanvas.turn_graph.get_turn_app", lambda: _BoomApp())
    with caplog.at_level(logging.WARNING, logger="loosecanvas.turn_logic"):
        snap = await save_session(sid)

    assert snap.messages == []
    assert snap.recent_events == []
    assert any("failed to read checkpoint" in rec.message for rec in caplog.records)


# ── SEC-05: generic path-rejection error (no absolute path leaked) ────────────


async def test_load_repo_session_rejects_path_generically(
    monkeypatch: Any, caplog: pytest.LogCaptureFixture
) -> None:
    """An out-of-base repo_path → ValueError with NO absolute path; warning logs it."""
    # A path guaranteed outside ALLOWED_BASE_DIR on every host.
    outside = (ALLOWED_BASE_DIR.parent / "definitely-not-allowed-xyz").resolve()
    if outside.is_relative_to(ALLOWED_BASE_DIR):  # pragma: no cover - host-dependent
        pytest.skip("parent of ALLOWED_BASE_DIR is inside it on this host")

    with (
        caplog.at_level(logging.WARNING, logger="loosecanvas.turn_logic"),
        pytest.raises(ValueError) as excinfo,
    ):
        await load_repo_session(outside)

    message = str(excinfo.value)
    assert message == "repo_path not allowed"
    # The leaked resolved path must NOT appear in the client-facing message.
    assert str(outside) not in message
    assert "ALLOWED_BASE_DIR" not in message
    # The server-side warning carries the resolved path for diagnostics.
    assert any(str(outside) in rec.getMessage() for rec in caplog.records)