turnstone/tests/test_session_ui_base.py

"""Tests for ``SessionUIBase`` — the shared UI scaffolding.

Covers listener fan-out, approval blocking gates, intent-judge
verdict bookkeeping, and the approval-cycle reset invariant that
prevents a late verdict from inheriting the previous round's
``user_decision``.

These are unit tests exercising the base class directly via a thin
concrete subclass — subclass-specific behaviour (WebUI's per-UI
metrics broadcast, ConsoleCoordinatorUI's collector fan-out) lives
in its own test files.
"""

from __future__ import annotations

import queue
import threading
from typing import Any
from unittest.mock import MagicMock, patch

import pytest

from tests.conftest import resolve_when_pending
from turnstone.core.session_ui_base import SessionUIBase


class _ConcreteUI(SessionUIBase):
    """Minimal concrete subclass — no kind-specific overrides.

    Exists only so we can instantiate the base (it's designed to be
    subclassed). Inherits the full base behaviour verbatim.
    """


def _make_ui(ws_id: str = "ws-1", user_id: str = "u1") -> _ConcreteUI:
    return _ConcreteUI(ws_id=ws_id, user_id=user_id)


# ---------------------------------------------------------------------------
# Listener fan-out
# ---------------------------------------------------------------------------


def test_register_listener_returns_fresh_queue() -> None:
    ui = _make_ui()
    lq = ui._register_listener()
    assert isinstance(lq, queue.Queue)
    assert lq in ui._listeners


def test_enqueue_fans_out_to_all_listeners() -> None:
    ui = _make_ui()
    lq1 = ui._register_listener()
    lq2 = ui._register_listener()
    ui._enqueue({"type": "hello"})
    # ``_enqueue`` stamps ``_event_id`` on every event so the ring
    # buffer can key replay against ``Last-Event-ID``; non-token
    # events (``hello`` isn't ``content`` / ``reasoning``) skip
    # ``_seq``.  Both listeners observe the SAME dict reference
    # (covered by ``test_listeners_share_dict_reference_warning``).
    assert lq1.get_nowait() == {"type": "hello", "ws_id": "ws-1", "_event_id": 1}
    assert lq2.get_nowait() == {"type": "hello", "ws_id": "ws-1", "_event_id": 1}


def test_enqueue_preserves_existing_ws_id() -> None:
    """When payload already carries ws_id, don't overwrite it — this
    supports the coord fan-out path where child events carry their own
    ws_id and parent forwarding mutates in place."""
    ui = _make_ui()
    lq = ui._register_listener()
    ui._enqueue({"type": "child_event", "ws_id": "child-9"})
    assert lq.get_nowait()["ws_id"] == "child-9"


def test_unregister_listener_removes_from_fanout() -> None:
    ui = _make_ui()
    lq = ui._register_listener()
    ui._unregister_listener(lq)
    ui._enqueue({"type": "hello"})
    assert lq.empty()


def test_enqueue_tolerates_full_listener_queue() -> None:
    """A slow SSE consumer shouldn't break the session's fan-out."""
    ui = _make_ui()
    lq = ui._register_listener(maxsize=1)
    lq.put_nowait({"type": "filler"})
    ui._enqueue({"type": "hello"})  # must not raise


# ---------------------------------------------------------------------------
# Approval gates
# ---------------------------------------------------------------------------


def test_resolve_approval_sets_result_and_unblocks_event() -> None:
    ui = _make_ui()
    ui._approval_event.clear()
    ui.resolve_approval(True, "looks good")
    assert ui._approval_result == (True, "looks good")
    assert ui._approval_event.is_set()


def test_resolve_approval_broadcasts_approval_resolved() -> None:
    ui = _make_ui()
    lq = ui._register_listener()
    ui.resolve_approval(False, "nope")
    event = lq.get_nowait()
    assert event["type"] == "approval_resolved"
    assert event["approved"] is False
    assert event["feedback"] == "nope"


# ---------------------------------------------------------------------------
# Intent-verdict bookkeeping
# ---------------------------------------------------------------------------


def _mock_storage(storage: Any = None) -> Any:
    storage = storage or MagicMock()
    return storage


def _patch_get_storage(storage: Any):  # type: ignore[no-untyped-def]
    """Patch ``turnstone.core.storage._registry.get_storage`` to return
    the supplied stub so the fire-and-forget persistence paths in
    SessionUIBase are observable under test."""
    return patch("turnstone.core.storage._registry.get_storage", return_value=storage)


def test_on_intent_verdict_caches_for_sse_replay() -> None:
    ui = _make_ui()
    with _patch_get_storage(MagicMock()):
        ui.on_intent_verdict({"verdict_id": "v1", "call_id": "c1", "risk_level": "low"})
    assert ui._llm_verdicts["c1"]["verdict_id"] == "v1"


def test_on_intent_verdict_persists_verdict_row() -> None:
    storage = MagicMock()
    ui = _make_ui()
    verdict = {
        "verdict_id": "v1",
        "call_id": "c1",
        "func_name": "bash",
        "risk_level": "medium",
        "confidence": 0.7,
        "recommendation": "review",
        "evidence": ["line-1"],
    }
    with _patch_get_storage(storage):
        ui.on_intent_verdict(verdict)
    storage.upsert_intent_verdict.assert_called_once()
    kwargs = storage.upsert_intent_verdict.call_args.kwargs
    assert kwargs["verdict_id"] == "v1"
    assert kwargs["ws_id"] == "ws-1"
    assert kwargs["call_id"] == "c1"


def test_on_intent_verdict_queues_pending_when_decision_unset() -> None:
    ui = _make_ui()
    with _patch_get_storage(MagicMock()):
        ui.on_intent_verdict({"verdict_id": "v1", "call_id": "c1"})
    assert ui._pending_verdicts == [{"verdict_id": "v1", "call_id": "c1"}]


def test_on_intent_verdict_stamps_immediately_when_decision_already_set() -> None:
    """Late-arriving verdict (after approval resolved) gets
    user_decision stamped immediately instead of queued."""
    storage = MagicMock()
    ui = _make_ui()
    ui._last_verdict_decision = "approved"
    with _patch_get_storage(storage):
        ui.on_intent_verdict({"verdict_id": "v-late", "call_id": "c-late"})
    # Not queued — decision was already set.
    assert ui._pending_verdicts == []
    storage.update_intent_verdict.assert_called_once_with("v-late", user_decision="approved")


def test_on_superseded_intent_verdict_persists_without_live_surfaces() -> None:
    """The persist-only audit hook for verdicts that landed after a newer
    turn replaced their judge generation: the row reaches storage with
    user_decision="superseded", but NONE of the live surfaces move — no
    SSE event, no ``_llm_verdicts`` cache entry (Smart Approvals must
    never see a stale call_id), no ``_pending_verdicts`` park (the next
    ``resolve_approval`` must not stamp it with the wrong decision)."""
    storage = MagicMock()
    ui = _make_ui()
    lq = ui._register_listener()
    verdict = {
        "verdict_id": "v-late",
        "call_id": "c-late",
        "func_name": "bash",
        "risk_level": "low",
        "tier": "llm",
    }
    with _patch_get_storage(storage):
        ui.on_superseded_intent_verdict(verdict)
    storage.upsert_intent_verdict.assert_called_once()
    kwargs = storage.upsert_intent_verdict.call_args.kwargs
    assert kwargs["verdict_id"] == "v-late"
    assert kwargs["user_decision"] == "superseded"
    assert lq.empty()  # no SSE delivery
    assert "c-late" not in ui._llm_verdicts  # no replay-cache write
    assert ui._pending_verdicts == []  # no decision-stamp park
    assert "user_decision" not in verdict  # caller's dict not mutated


def test_llm_verdict_cache_evicts_oldest_at_cap() -> None:
    """FIFO eviction at ``_LLM_VERDICT_CACHE_MAX`` prevents unbounded
    growth on a long-running session."""
    ui = _make_ui()
    cap = SessionUIBase._LLM_VERDICT_CACHE_MAX
    with _patch_get_storage(MagicMock()):
        for i in range(cap + 5):
            ui.on_intent_verdict({"verdict_id": f"v{i}", "call_id": f"c{i}"})
    assert len(ui._llm_verdicts) == cap
    # Oldest five should have been evicted.
    assert "c0" not in ui._llm_verdicts
    assert "c4" not in ui._llm_verdicts
    assert f"c{cap + 4}" in ui._llm_verdicts


# ---------------------------------------------------------------------------
# Approval cycle reset — the bug-1 regression
# ---------------------------------------------------------------------------


def test_reset_approval_cycle_clears_decision_and_cache() -> None:
    ui = _make_ui()
    ui._last_verdict_decision = "approved"
    ui._llm_verdicts["c-stale"] = {"verdict_id": "stale"}
    ui._reset_approval_cycle()
    assert ui._last_verdict_decision == ""
    assert ui._llm_verdicts == {}


def test_late_verdict_in_new_round_not_stamped_with_prior_decision() -> None:
    """Regression test for the ultrareview bug-1 finding.

    Round 1: approve → _last_verdict_decision = "approved".
    Round 2 begins: caller calls _reset_approval_cycle().
    A verdict fires mid-round 2: must NOT inherit "approved" from
    round 1. Must land in _pending_verdicts waiting for this round's
    resolution.
    """
    storage = MagicMock()
    ui = _make_ui()
    # Simulate round 1 completion.
    with _patch_get_storage(storage):
        ui.on_intent_verdict({"verdict_id": "v1", "call_id": "c1"})
        ui.resolve_approval(True, None)
    assert ui._last_verdict_decision == "approved"
    # Round 2 begins — subclass approve_tools calls this at entry.
    ui._reset_approval_cycle()
    # Late judge fires during round 2 BEFORE the user decides.
    with _patch_get_storage(storage):
        ui.on_intent_verdict({"verdict_id": "v2", "call_id": "c2"})
    # The new verdict must be pending (awaiting this round's decision),
    # NOT already stamped with round 1's "approved".
    assert ui._pending_verdicts == [{"verdict_id": "v2", "call_id": "c2"}]
    # update_intent_verdict was only called ONCE: for v1 when round 1
    # resolved. v2 should NOT have been stamped.
    for call in storage.update_intent_verdict.call_args_list:
        assert call.args[0] != "v2", "late verdict was stamped with prior round's decision"


def test_both_subclasses_call_reset_from_approve_tools() -> None:
    """Regression for bug-1: the real subclass ``approve_tools``
    methods must invoke ``_reset_approval_cycle`` at entry. Without
    this, coord sessions that already resolved a prior approval stamp
    the next round's late verdicts with the stale decision.
    """
    import turnstone.server
    from turnstone.console.coordinator_ui import ConsoleCoordinatorUI

    webui = turnstone.server.WebUI

    for cls in (webui, ConsoleCoordinatorUI):
        ui = cls(ws_id="ws-x", user_id="u1")
        # Stage state as if a prior approval round already finished.
        ui._last_verdict_decision = "approved"
        ui._llm_verdicts["stale"] = {"verdict_id": "stale"}
        # Entering approve_tools for a new round — the reset must fire.
        # Pass items with needs_approval=False so approve_tools returns
        # without blocking on user input.
        with _patch_get_storage(MagicMock()):
            ui.approve_tools([{"func_name": "ls", "needs_approval": False}])
        assert ui._last_verdict_decision == "", (
            f"{cls.__name__}.approve_tools did not call _reset_approval_cycle "
            "— next round's verdicts would inherit the prior decision"
        )
        assert ui._llm_verdicts == {}, (
            f"{cls.__name__}.approve_tools did not clear the LLM verdict cache"
        )


def test_on_intent_verdict_decision_check_and_queue_are_atomic() -> None:
    """Regression for the on_intent_verdict ↔ resolve_approval race.

    Prior implementation acquired ``_ws_lock`` twice: once to read
    ``_last_verdict_decision``, once to append to
    ``_pending_verdicts``. Between those two acquisitions
    ``resolve_approval`` could swap-and-clear the pending list and
    set the decision — our verdict then got appended to the fresh
    list and stamped with the NEXT round's decision.

    Fix: decision check + append happen under a single lock
    acquisition. This test counts lock acquisitions during one
    ``on_intent_verdict`` and fails if the release-then-reacquire
    pattern returns.
    """
    ui = _make_ui()
    acquire_count = 0
    original_lock = ui._ws_lock

    class _CountingLock:
        def __init__(self, inner: threading.Lock) -> None:
            self._inner = inner

        def __enter__(self) -> None:
            nonlocal acquire_count
            acquire_count += 1
            self._inner.acquire()

        def __exit__(self, *a: Any) -> None:
            self._inner.release()

        def acquire(self, *a: Any, **kw: Any) -> bool:
            return self._inner.acquire(*a, **kw)

        def release(self) -> None:
            self._inner.release()

    ui._ws_lock = _CountingLock(original_lock)  # type: ignore[assignment]
    with _patch_get_storage(MagicMock()):
        ui.on_intent_verdict({"verdict_id": "v1", "call_id": "c1"})
    # Two acquisitions: one for the cache write (call_id is truthy),
    # one for decision-check + pending-append. Before the fix there
    # were three, with a window resolve_approval could slip into.
    assert acquire_count == 2, (
        f"on_intent_verdict acquired _ws_lock {acquire_count} times; "
        "decision-check + pending-append must happen under ONE acquisition "
        "to avoid a race with resolve_approval"
    )


def test_resolve_approval_stamps_all_pending_verdicts() -> None:
    """Normal path: multiple verdicts queued during the round, all get
    stamped with the user's decision on resolve."""
    storage = MagicMock()
    ui = _make_ui()
    with _patch_get_storage(storage):
        ui.on_intent_verdict({"verdict_id": "v1", "call_id": "c1"})
        ui.on_intent_verdict({"verdict_id": "v2", "call_id": "c2"})
    assert len(ui._pending_verdicts) == 2
    with _patch_get_storage(storage):
        ui.resolve_approval(False, "too risky")
    # Both verdicts get stamped.
    stamped_ids = {c.args[0] for c in storage.update_intent_verdict.call_args_list}
    assert stamped_ids == {"v1", "v2"}
    # Pending list cleared after resolve.
    assert ui._pending_verdicts == []
    assert ui._last_verdict_decision == "denied"


# ---------------------------------------------------------------------------
# user_decision value space — pending / approved / denied / timeout
# / auto-approve reasons (policy / blanket / skill / always / auto_approve_tools).
# Guards the "user_decision is never empty for new rows" invariant.
# ---------------------------------------------------------------------------


def test_resolve_approval_timeout_kwarg_writes_timeout_value() -> None:
    """``resolve_approval(False, ..., timeout=True)`` writes
    ``user_decision="timeout"`` so the audit trail can distinguish a
    passive timeout expiry from an active user denial — the feedback
    string used to carry this distinction but the column alone could not."""
    storage = MagicMock()
    ui = _make_ui()
    with _patch_get_storage(storage):
        ui.on_intent_verdict({"verdict_id": "v1", "call_id": "c1"})
    with _patch_get_storage(storage):
        ui.resolve_approval(False, "expired", timeout=True)
    storage.update_intent_verdict.assert_any_call("v1", user_decision="timeout")
    assert ui._last_verdict_decision == "timeout"


def test_resolve_approval_timeout_with_approved_raises() -> None:
    """``timeout=True`` is mutually exclusive with ``approved=True`` —
    the combination would land a row whose audit column says
    ``"timeout"`` while the SSE event reports ``approved=True``. Fail
    loud so the inconsistency can't ship silently."""
    import pytest

    ui = _make_ui()
    with pytest.raises(ValueError, match="timeout"):
        ui.resolve_approval(True, timeout=True)


def test_record_auto_approves_populates_reason_lookup() -> None:
    """``_record_auto_approves`` must seed
    ``_auto_approve_reasons[call_id]`` with the per-item reason so a
    late-arriving LLM judge verdict can recover the auto-approve
    reason via ``on_intent_verdict``."""
    storage = MagicMock()
    ui = _make_ui()
    items = [
        {
            "call_id": "c-policy",
            "func_name": "bash",
            "auto_approved": True,
            "auto_approve_reason": "policy",
        },
        {
            "call_id": "c-blanket",
            "func_name": "list_workstreams",
            "auto_approved": True,
            "auto_approve_reason": "blanket",
        },
    ]
    with _patch_get_storage(storage):
        ui._record_auto_approves(items)
    assert "c-policy" in ui._auto_approve_reasons
    assert "c-blanket" in ui._auto_approve_reasons
    assert ui._auto_approve_reasons["c-policy"][0] == "policy"
    assert ui._auto_approve_reasons["c-blanket"][0] == "blanket"


def test_on_intent_verdict_consumes_auto_approve_reason() -> None:
    """A late LLM verdict for a previously auto-approved call_id picks
    up the reason from ``_auto_approve_reasons``, stamps it on the
    verdict before persist, and pops the entry so re-use isn't
    possible. Closes the misdiagnosis bug where auto-approved tools
    landed verdict rows with ``user_decision=""``."""
    storage = MagicMock()
    ui = _make_ui()
    ui._auto_approve_reasons["c-x"] = ("auto_approve_tools", 0.0)
    with _patch_get_storage(storage):
        ui.on_intent_verdict({"verdict_id": "v-x", "call_id": "c-x"})
    storage.upsert_intent_verdict.assert_called_once()
    kwargs = storage.upsert_intent_verdict.call_args.kwargs
    assert kwargs["user_decision"] == "auto_approve_tools"
    # Consumed on read so the same call_id can't double-stamp later.
    assert "c-x" not in ui._auto_approve_reasons
    # Auto-stamped verdicts must NOT join _pending_verdicts — the
    # row's final decision is already set; appending would let a
    # later resolve_approval overwrite the auto-reason with the
    # manual decision (real audit-trail clobber bug).
    assert ui._pending_verdicts == []


def test_on_intent_verdict_auto_reason_survives_resolve_cycle() -> None:
    """Mixed-batch case: one tool was auto-approved (policy), another
    needs manual approval. The LLM judge fires for the auto-approved
    sibling DURING the manual-approval wait. The verdict must land
    with ``user_decision="policy"`` and stay that way even after
    ``resolve_approval`` fires for the pending sibling — the prior
    bug was that the auto-stamped row got overwritten with
    ``"approved"``/``"denied"`` by the resolve path."""
    storage = MagicMock()
    ui = _make_ui()
    ui._auto_approve_reasons["c-auto"] = ("policy", 0.0)
    with _patch_get_storage(storage):
        # LLM verdict fires for the auto-approved sibling.
        ui.on_intent_verdict({"verdict_id": "v-auto", "call_id": "c-auto"})
        # Now the pending sibling gets a verdict + manual resolve.
        ui.on_intent_verdict({"verdict_id": "v-pending", "call_id": "c-pending"})
        ui.resolve_approval(True, "looks good")
    # Only the pending verdict should be UPDATEd to "approved" — the
    # auto-stamped one stays "policy" via its INSERT.
    update_calls = {
        c.args[0]: c.kwargs.get("user_decision")
        for c in storage.update_intent_verdict.call_args_list
    }
    assert update_calls == {"v-pending": "approved"}
    # The auto verdict's INSERT carried the policy reason.
    insert_calls = {
        c.kwargs["verdict_id"]: c.kwargs["user_decision"]
        for c in storage.upsert_intent_verdict.call_args_list
    }
    assert insert_calls["v-auto"] == "policy"
    assert insert_calls["v-pending"] == "pending"


def test_persist_auto_approved_heuristic_verdicts_stamps_reason() -> None:
    """The auto-approve early-return branches in ``approve_tools`` used
    to drop heuristic verdicts on the floor — auditors couldn't tell
    whether the judge ran or the call was simply silently auto-approved.
    ``_persist_auto_approved_heuristic_verdicts`` closes that gap and
    stamps each verdict with the item's reason."""
    storage = MagicMock()
    ui = _make_ui()
    items = [
        {
            "call_id": "c-1",
            "auto_approved": True,
            "auto_approve_reason": "blanket",
            "_heuristic_verdict": {
                "verdict_id": "v-1",
                "call_id": "c-1",
                "risk_level": "low",
                "recommendation": "review",
            },
        },
        # No _heuristic_verdict — skipped (judge didn't run for this item).
        {"call_id": "c-2", "auto_approved": True, "auto_approve_reason": "blanket"},
        # Not auto_approved — skipped (this helper only handles auto-approved).
        {
            "call_id": "c-3",
            "_heuristic_verdict": {"verdict_id": "v-3", "call_id": "c-3"},
        },
    ]
    with _patch_get_storage(storage):
        ui._persist_auto_approved_heuristic_verdicts(items)
    storage.create_intent_verdicts_bulk.assert_called_once()
    rows = storage.create_intent_verdicts_bulk.call_args.args[0]
    assert len(rows) == 1
    assert rows[0]["verdict_id"] == "v-1"
    assert rows[0]["user_decision"] == "blanket"


def test_auto_approve_reasons_ttl_prune_drops_stale_entries() -> None:
    """Lazy TTL eviction at write time: entries older than
    ``_AUTO_APPROVE_REASON_TTL`` are pruned on the next
    ``_record_auto_approves`` call. Without this, a session with the
    LLM judge disabled would accumulate entries that never get
    consumed."""
    import time as time_module

    storage = MagicMock()
    ui = _make_ui()
    # Seed two stale entries (well past the TTL).
    stale_ts = time_module.time() - ui._AUTO_APPROVE_REASON_TTL - 30.0
    ui._auto_approve_reasons["c-stale-1"] = ("policy", stale_ts)
    ui._auto_approve_reasons["c-stale-2"] = ("blanket", stale_ts)
    items = [
        {
            "call_id": "c-fresh",
            "auto_approved": True,
            "auto_approve_reason": "skill",
            "func_name": "bash",
        }
    ]
    with _patch_get_storage(storage):
        ui._record_auto_approves(items)
    # Stale entries pruned; only the fresh one remains.
    assert "c-stale-1" not in ui._auto_approve_reasons
    assert "c-stale-2" not in ui._auto_approve_reasons
    assert "c-fresh" in ui._auto_approve_reasons


# ---------------------------------------------------------------------------
# Output guard persistence
# ---------------------------------------------------------------------------


def test_on_output_warning_enqueues_only() -> None:
    # Persistence was decoupled from on_output_warning when the LLM
    # judge stage landed — the session now calls record_output_assessment
    # directly per tier.  on_output_warning is UI-dispatch only.
    storage = MagicMock()
    ui = _make_ui()
    lq = ui._register_listener()
    assessment = {
        "func_name": "bash",
        "flags": ["secret_leak"],
        "risk_level": "high",
        "output_length": 200,
    }
    with _patch_get_storage(storage):
        ui.on_output_warning("call-1", assessment)
    event = lq.get_nowait()
    assert event["type"] == "output_warning"
    assert event["call_id"] == "call-1"
    assert event["risk_level"] == "high"
    storage.record_output_assessment.assert_not_called()


def test_record_output_assessment_persists_with_tier() -> None:
    storage = MagicMock()
    ui = _make_ui()
    assessment = {
        "func_name": "web_fetch",
        "flags": ["camouflaged_injection"],
        "risk_level": "medium",
        "output_length": 4096,
    }
    with _patch_get_storage(storage):
        ui.record_output_assessment(
            "call-2",
            assessment,
            tier="llm",
            reasoning="LLM saw a camouflaged directive",
            judge_model="gpt-5-mini",
            latency_ms=142,
        )
    storage.record_output_assessment.assert_called_once()
    kwargs = storage.record_output_assessment.call_args.kwargs
    assert kwargs["tier"] == "llm"
    assert kwargs["reasoning"] == "LLM saw a camouflaged directive"
    assert kwargs["judge_model"] == "gpt-5-mini"
    assert kwargs["latency_ms"] == 142
    assert kwargs["risk_level"] == "medium"


def test_record_output_assessment_defaults_to_heuristic_tier() -> None:
    storage = MagicMock()
    ui = _make_ui()
    assessment = {
        "func_name": "bash",
        "flags": [],
        "risk_level": "none",
        "output_length": 0,
    }
    with _patch_get_storage(storage):
        ui.record_output_assessment("call-3", assessment)
    kwargs = storage.record_output_assessment.call_args.kwargs
    assert kwargs["tier"] == "heuristic"
    assert kwargs["reasoning"] == ""
    assert kwargs["judge_model"] == ""
    assert kwargs["latency_ms"] == 0


# ---------------------------------------------------------------------------
# Concurrency smoke
# ---------------------------------------------------------------------------


# ---------------------------------------------------------------------------
# serialize_pending_approval_detail — dashboard projection
# ---------------------------------------------------------------------------


def test_serialize_pending_approval_detail_returns_none_when_unset() -> None:
    ui = _make_ui()
    assert ui.serialize_pending_approval_detail() is None


def test_serialize_pending_approval_detail_returns_none_when_items_empty() -> None:
    ui = _make_ui()
    ui._pending_approval = {"type": "approve_request", "items": [], "judge_pending": False}
    assert ui.serialize_pending_approval_detail() is None


def test_serialize_pending_approval_detail_merges_judge_verdict() -> None:
    ui = _make_ui()
    ui._pending_approval = {
        "type": "approve_request",
        "items": [
            {
                "call_id": "c-1",
                "header": "bash",
                "preview": "$ ls",
                "func_name": "bash",
                "approval_label": "bash",
                "needs_approval": True,
                "error": None,
                "verdict": {"recommendation": "review", "tier": "heuristic"},
            }
        ],
        "judge_pending": True,
    }
    ui._llm_verdicts["c-1"] = {
        "verdict_id": "v-1",
        "call_id": "c-1",
        "risk_level": "high",
        "recommendation": "deny",
        "tier": "llm",
    }
    detail = ui.serialize_pending_approval_detail()
    assert detail is not None
    assert detail["call_id"] == "c-1"
    assert detail["judge_pending"] is True
    assert len(detail["items"]) == 1
    item = detail["items"][0]
    assert item["call_id"] == "c-1"
    assert item["header"] == "bash"
    assert item["preview"] == "$ ls"
    assert item["heuristic_verdict"] == {"recommendation": "review", "tier": "heuristic"}
    assert item["judge_verdict"]["recommendation"] == "deny"
    assert item["judge_verdict"]["risk_level"] == "high"


def test_serialize_pending_approval_detail_judge_verdict_none_when_missing() -> None:
    """No cached verdict for the call_id → judge_verdict is None,
    not absent or some sentinel."""
    ui = _make_ui()
    ui._pending_approval = {
        "type": "approve_request",
        "items": [{"call_id": "c-1", "func_name": "ls", "needs_approval": True}],
        "judge_pending": True,
    }
    detail = ui.serialize_pending_approval_detail()
    assert detail is not None
    assert detail["items"][0]["judge_verdict"] is None
    assert detail["items"][0]["heuristic_verdict"] is None


def test_serialize_pending_approval_detail_multi_item() -> None:
    ui = _make_ui()
    ui._pending_approval = {
        "type": "approve_request",
        "items": [
            {"call_id": "c-1", "func_name": "bash", "needs_approval": True},
            {"call_id": "c-2", "func_name": "mcp__sf__query", "needs_approval": True},
        ],
        "judge_pending": False,
    }
    ui._llm_verdicts["c-2"] = {"recommendation": "deny", "risk_level": "crit"}
    detail = ui.serialize_pending_approval_detail()
    assert detail is not None
    assert detail["call_id"] == "c-1"  # primary = first item
    assert len(detail["items"]) == 2
    assert detail["items"][0]["judge_verdict"] is None
    assert detail["items"][1]["judge_verdict"]["recommendation"] == "deny"


def test_serialize_pending_approval_detail_tool_policy_denied_passthrough() -> None:
    """A tool-policy-denied item carries error + needs_approval=False
    after WebUI.approve_tools mutates the items list. The serializer
    must round-trip both fields so the JS can detect the
    POLICY-BLOCKED matrix row and render the banner instead of
    approve/deny buttons."""
    ui = _make_ui()
    ui._pending_approval = {
        "type": "approve_request",
        "items": [
            {
                "call_id": "c-1",
                "func_name": "rm_rf",
                "approval_label": "rm_rf",
                "needs_approval": False,
                "error": "Blocked by tool policy (pattern match for 'rm_rf')",
            }
        ],
        "judge_pending": False,
    }
    detail = ui.serialize_pending_approval_detail()
    assert detail is not None
    item = detail["items"][0]
    # Both fields are the JS detection keys for the POLICY-BLOCKED
    # branch in renderApprovalBlock — drift here silently regresses
    # to a buttoned approve UI on a server-blocked call.
    assert item["needs_approval"] is False
    assert item["error"] == "Blocked by tool policy (pattern match for 'rm_rf')"


def test_serialize_pending_approval_detail_judge_unavailable_path() -> None:
    """No judge_verdict + no heuristic_verdict + judge_pending=False
    is the (judge unavailable) matrix row — the JS detects it via
    !verdict && !judgePending && !policyBlocked. Verify the
    serialized payload preserves the absence of all three signals."""
    ui = _make_ui()
    ui._pending_approval = {
        "type": "approve_request",
        "items": [
            {
                "call_id": "c-1",
                "func_name": "bash",
                "approval_label": "bash",
                "needs_approval": True,
            }
        ],
        "judge_pending": False,
    }
    detail = ui.serialize_pending_approval_detail()
    assert detail is not None
    assert detail["judge_pending"] is False
    item = detail["items"][0]
    assert item["judge_verdict"] is None
    assert item["heuristic_verdict"] is None
    assert item["needs_approval"] is True
    assert item["error"] is None


def test_serialize_pending_approval_detail_returned_dict_is_decoupled() -> None:
    """Mutating the returned dict must not corrupt the cached
    verdict, which other consumers may still read."""
    ui = _make_ui()
    ui._pending_approval = {
        "type": "approve_request",
        "items": [{"call_id": "c-1", "func_name": "bash", "needs_approval": True}],
        "judge_pending": False,
    }
    ui._llm_verdicts["c-1"] = {"recommendation": "approve"}
    detail = ui.serialize_pending_approval_detail()
    assert detail is not None
    detail["items"][0]["judge_verdict"]["recommendation"] = "MUTATED"
    assert ui._llm_verdicts["c-1"]["recommendation"] == "approve"


# ---------------------------------------------------------------------------
# Auto-approve visibility — _serialize_approval_items + _record_auto_approves
# + serialize_recent_auto_approvals
# ---------------------------------------------------------------------------


def test_serialize_approval_items_forwards_auto_approve_fields() -> None:
    """When the upstream pipeline tags an item with ``auto_approved`` +
    ``auto_approve_reason``, the serialized payload must carry both
    so the dashboard pill / per-ws SSE consumer can show *which*
    path bypassed the operator gate."""
    ui = _make_ui()
    items = [
        {
            "call_id": "c1",
            "func_name": "bash",
            "approval_label": "bash",
            "needs_approval": False,
            "auto_approved": True,
            "auto_approve_reason": "skill",
        },
        {
            "call_id": "c2",
            "func_name": "read_file",
            "needs_approval": False,
            # No auto_approved tag — read-only tool that never needed approval.
        },
    ]
    out = ui._serialize_approval_items(items)
    assert out[0]["auto_approved"] is True
    assert out[0]["auto_approve_reason"] == "skill"
    # Items not flagged as auto-approved must NOT carry the fields —
    # otherwise the dashboard would show pills for read-only tools too.
    assert "auto_approved" not in out[1]
    assert "auto_approve_reason" not in out[1]


def test_serialize_approval_items_forwards_denial_msg_as_error() -> None:
    """Denied items surface their ``denial_msg`` as ``error`` so the
    /dashboard / SSE consumer renders the policy-block reason
    without exposing the raw item shape."""
    ui = _make_ui()
    items = [
        {
            "call_id": "c1",
            "func_name": "bash",
            "denied": True,
            "denial_msg": "Blocked by tool policy (pattern match for 'bash')",
        }
    ]
    out = ui._serialize_approval_items(items)
    assert out[0]["error"] == "Blocked by tool policy (pattern match for 'bash')"


def test_record_auto_approves_appends_only_tagged_items() -> None:
    """Items without ``auto_approved=True`` are skipped — the ring
    buffer is meant to surface bypassed-the-gate calls, not a
    record of every tool invocation."""
    storage = MagicMock()
    ui = _make_ui()
    items = [
        {
            "call_id": "c1",
            "func_name": "bash",
            "approval_label": "bash",
            "auto_approved": True,
            "auto_approve_reason": "skill",
        },
        {
            "call_id": "c2",
            "func_name": "read_file",
            # No auto_approved tag — read-only tool, gets skipped.
        },
    ]
    with _patch_get_storage(storage):
        ui._record_auto_approves(items)
    snapshot = ui.serialize_recent_auto_approvals()
    assert len(snapshot) == 1
    assert snapshot[0]["func_name"] == "bash"
    assert snapshot[0]["auto_approve_reason"] == "skill"
    # Audit row recorded — one row per call (not per item) so
    # tool-heavy turns don't blow up the audit table.
    storage.record_audit_event.assert_called_once()
    call_kwargs = storage.record_audit_event.call_args.kwargs
    assert call_kwargs["action"] == "tool.auto_approved"


def test_record_auto_approves_caps_buffer_at_max() -> None:
    """Bounded ring buffer — a long-running skill workstream can't
    fill the /dashboard payload with stale rows.  The cap is the
    class-level constant, exercised here to lock the contract."""
    ui = _make_ui()
    cap = ui._RECENT_AUTO_APPROVALS_MAX
    # Push (cap + 5) items; only the most recent ``cap`` survive.
    for i in range(cap + 5):
        with _patch_get_storage(MagicMock()):
            ui._record_auto_approves(
                [
                    {
                        "call_id": f"c{i}",
                        "func_name": f"tool_{i}",
                        "auto_approved": True,
                        "auto_approve_reason": "blanket",
                    }
                ]
            )
    snapshot = ui.serialize_recent_auto_approvals()
    assert len(snapshot) == cap
    # Tail preserved — oldest entries roll off the head.
    assert snapshot[-1]["func_name"] == f"tool_{cap + 5 - 1}"
    assert snapshot[0]["func_name"] == f"tool_{5}"


def test_record_auto_approves_noop_when_no_tagged_items() -> None:
    """No tagged items → no buffer write, no audit — matters for
    the every-tool-call-was-read-only case where ``items`` is
    non-empty but nothing was an auto-approve."""
    storage = MagicMock()
    ui = _make_ui()
    with _patch_get_storage(storage):
        ui._record_auto_approves(
            [{"call_id": "c1", "func_name": "read_file"}]  # no auto_approved tag
        )
    assert ui.serialize_recent_auto_approvals() == []
    storage.record_audit_event.assert_not_called()


def test_record_auto_approves_swallows_audit_failure() -> None:
    """An audit-write exception must not break the tool-execution
    path — visibility is best-effort, the SSE event + ring buffer
    already shipped to operators by the time this fires."""
    storage = MagicMock()
    storage.record_audit_event.side_effect = RuntimeError("audit table down")
    ui = _make_ui()
    items = [
        {
            "call_id": "c1",
            "func_name": "bash",
            "auto_approved": True,
            "auto_approve_reason": "policy",
        }
    ]
    # Must not raise — the docstring explicitly promises best-effort.
    with _patch_get_storage(storage):
        ui._record_auto_approves(items)
    # Buffer write still happened (it's first, before the audit).
    assert len(ui.serialize_recent_auto_approvals()) == 1


def test_replay_recent_auto_approvals_from_audit_seeds_buffer() -> None:
    """Audit-replay seeds the ring buffer on UI construction so the
    dashboard pill survives UI rebuilds (saved-workstream rehydrate /
    coord→node click-through / process restart all create a fresh UI
    whose buffer would otherwise be empty even though the audit row
    is still on disk)."""
    storage = MagicMock()
    storage.list_audit_events.return_value = [
        # DESC order — newest first.
        {
            "timestamp": "2026-04-27T18:00:00",
            "detail": (
                '{"tools": [{"call_id": "c2", "func_name": "edit_file",'
                ' "approval_label": "edit_file", "reason": "policy"}],'
                ' "count": 1}'
            ),
        },
        {
            "timestamp": "2026-04-27T17:00:00",
            "detail": (
                '{"tools": [{"call_id": "c1", "func_name": "bash",'
                ' "approval_label": "bash", "reason": "skill"}],'
                ' "count": 1}'
            ),
        },
    ]
    with _patch_get_storage(storage):
        ui = _make_ui(ws_id="ws-replay")
    # Buffer holds the replayed entries in chronological order
    # (oldest first), matching what live appends produce.
    snapshot = ui.serialize_recent_auto_approvals()
    assert len(snapshot) == 2
    assert snapshot[0]["func_name"] == "bash"
    assert snapshot[0]["auto_approve_reason"] == "skill"
    assert snapshot[1]["func_name"] == "edit_file"
    assert snapshot[1]["auto_approve_reason"] == "policy"
    # And the audit query was scoped to this ws + tool.auto_approved.
    storage.list_audit_events.assert_called_once()
    call_kwargs = storage.list_audit_events.call_args.kwargs
    assert call_kwargs["action"] == "tool.auto_approved"
    assert call_kwargs["resource_id"] == "ws-replay"


def test_replay_swallows_audit_storage_failure() -> None:
    """A storage outage at construction time must not break UI
    instantiation — the buffer simply stays empty until the next
    live auto-approve populates it."""
    storage = MagicMock()
    storage.list_audit_events.side_effect = RuntimeError("audit table down")
    with _patch_get_storage(storage):
        ui = _make_ui(ws_id="ws-replay")
    assert ui.serialize_recent_auto_approvals() == []


def test_replay_skips_when_ws_id_missing() -> None:
    """No ws_id → no audit query.  Test fixtures sometimes
    construct a UI with the default empty ws_id; the replay must
    not fire a wildcard query that returns rows from other ws's."""
    storage = MagicMock()
    with _patch_get_storage(storage):
        ui = _make_ui(ws_id="")
    storage.list_audit_events.assert_not_called()
    assert ui.serialize_recent_auto_approvals() == []


def test_replay_tolerates_malformed_audit_detail() -> None:
    """Unparseable / wrong-shape audit detail rows are skipped, not
    propagated.  A historic audit row with a different schema (e.g.
    pre-fix migration leftover) must not crash UI construction."""
    storage = MagicMock()
    storage.list_audit_events.return_value = [
        {"timestamp": "2026-04-27T18:00:00", "detail": "not-json"},
        {"timestamp": "2026-04-27T17:30:00", "detail": '{"tools": "wrong-shape"}'},
        {
            "timestamp": "2026-04-27T17:00:00",
            "detail": '{"tools": [{"func_name": "bash", "reason": "skill"}], "count": 1}',
        },
    ]
    with _patch_get_storage(storage):
        ui = _make_ui(ws_id="ws-replay")
    # Only the well-shaped row contributes.
    snapshot = ui.serialize_recent_auto_approvals()
    assert len(snapshot) == 1
    assert snapshot[0]["func_name"] == "bash"


def test_parse_audit_timestamp_treats_naive_strings_as_utc() -> None:
    """Audit rows are stored as naive UTC strings (e.g.
    ``2026-04-27T18:00:00`` with no timezone marker); a server in
    a non-UTC timezone would mis-stamp pill entries by hours
    without explicit UTC.replace at parse time."""
    from datetime import UTC, datetime

    from turnstone.core.session_ui_base import SessionUIBase

    expected = datetime(2026, 4, 27, 18, 0, 0, tzinfo=UTC).timestamp()
    assert SessionUIBase._parse_audit_timestamp("2026-04-27T18:00:00") == expected
    # Explicit-offset strings parse correctly too — the UTC stamp
    # only applies when tzinfo is None.
    assert SessionUIBase._parse_audit_timestamp("2026-04-27T18:00:00+00:00") == expected


def test_replay_caps_at_buffer_max() -> None:
    """Replay output is bounded by the same cap as live appends.
    A long-lived workstream with hundreds of audit rows must not
    blow past the 10-entry limit during replay."""
    storage = MagicMock()
    # Generate many fake rows.
    storage.list_audit_events.return_value = [
        {
            "timestamp": f"2026-04-27T{i:02d}:00:00",
            "detail": (
                f'{{"tools": [{{"func_name": "tool_{i}", "reason": "skill"}}], "count": 1}}'
            ),
        }
        for i in range(20)
    ]
    with _patch_get_storage(storage):
        ui = _make_ui(ws_id="ws-replay")
    snapshot = ui.serialize_recent_auto_approvals()
    # Cap holds even when audit-replay fans in past it.
    assert len(snapshot) == ui._RECENT_AUTO_APPROVALS_MAX


def test_serialize_recent_auto_approvals_returns_a_copy() -> None:
    """Mutating the returned list must not corrupt the buffer —
    HTTP handler should not be able to drain or reorder it."""
    ui = _make_ui()
    with _patch_get_storage(MagicMock()):
        ui._record_auto_approves(
            [
                {
                    "call_id": "c1",
                    "func_name": "bash",
                    "auto_approved": True,
                    "auto_approve_reason": "skill",
                }
            ]
        )
    snapshot = ui.serialize_recent_auto_approvals()
    snapshot.clear()
    snapshot.append({"poisoned": True})
    # Buffer state survives the caller's mutation.
    fresh = ui.serialize_recent_auto_approvals()
    assert len(fresh) == 1
    assert fresh[0]["func_name"] == "bash"


# ---------------------------------------------------------------------------


def test_concurrent_enqueue_and_listener_registration() -> None:
    """Fan-out under concurrent enqueue + register/unregister shouldn't
    drop events or crash on the lock. Sanity-level stress."""
    ui = _make_ui()

    def _producer() -> None:
        for i in range(100):
            ui._enqueue({"type": "tick", "n": i})

    def _subscriber() -> None:
        for _ in range(20):
            lq = ui._register_listener()
            ui._unregister_listener(lq)

    producer = threading.Thread(target=_producer)
    subscribers = [threading.Thread(target=_subscriber) for _ in range(4)]
    producer.start()
    for s in subscribers:
        s.start()
    producer.join()
    for s in subscribers:
        s.join()
    # Test's job is to surface any RuntimeError / lock inversion
    # during concurrent enqueue + register/unregister. If we got
    # here every thread completed cleanly — assert explicitly so the
    # intent survives optimization-mode assertion stripping.
    assert not producer.is_alive()
    assert all(not s.is_alive() for s in subscribers)


# ---------------------------------------------------------------------------
# Per-turn inflight buffers — SSE refresh-resume snapshot path
# ---------------------------------------------------------------------------


def test_on_content_token_writes_to_both_buffers() -> None:
    """``on_content_token`` writes to the multi-turn buffer (IDLE
    piggyback) AND the per-turn inflight buffer (SSE snapshot)."""
    ui = _make_ui()
    ui.on_content_token("hello")
    assert ui._ws_turn_content == ["hello"]
    assert ui._ws_inflight_content == ["hello"]
    assert ui._event_id == 1


def test_on_reasoning_token_writes_to_inflight_buffer_only() -> None:
    """Reasoning has no multi-turn IDLE piggyback — only the inflight
    buffer + the seq counter."""
    ui = _make_ui()
    ui.on_reasoning_token("thinking...")
    assert ui._ws_inflight_reasoning == ["thinking..."]
    assert ui._event_id == 1
    # Multi-turn buffer is content-only and untouched by reasoning.
    assert ui._ws_turn_content == []


def test_inflight_seq_advances_on_every_emit_even_at_cap() -> None:
    """Cap-hit content tokens MUST advance ``_event_id``,
    even though the buffer rejected the append. If seq stalled at
    high-water-pre-cap, a subscriber registering AFTER the cap is
    hit would capture ``snap_seq == stalled_seq`` and every
    subsequent live token (also tagged with the stalled seq) would
    be filter-dropped by the events handler — silently losing the
    rest of the stream. The cap is a buffer-size limit, not a
    "stop streaming" signal."""
    from turnstone.core.session_ui_base import _MAX_TURN_CONTENT_CHARS

    ui = _make_ui()
    chunk = "x" * 1024
    while ui._ws_inflight_content_size < _MAX_TURN_CONTENT_CHARS:
        ui.on_content_token(chunk)
    seq_at_cap = ui._event_id

    # Cap-hit token: seq MUST advance (no buffer append, but the
    # event still gets a fresh seq for the dedup filter).
    ui.on_content_token(chunk)
    assert ui._event_id == seq_at_cap + 1
    # Buffer remains bounded — the cap-hit token is NOT in inflight.
    assert ui._ws_inflight_content_size <= _MAX_TURN_CONTENT_CHARS + len(chunk)


def test_subscriber_after_cap_hit_receives_subsequent_tokens() -> None:
    """Regression for Copilot's cap+seq finding: a subscriber that
    connects AFTER the inflight buffer is at cap must still receive
    live tokens past the cap. Past-cap tokens are absent from
    ``snap.content`` (the snapshot text was truncated at cap) but
    the live stream past them must NOT be filter-dropped."""
    from turnstone.core.session_ui_base import _MAX_TURN_CONTENT_CHARS

    ui = _make_ui()
    chunk = "x" * 1024
    while ui._ws_inflight_content_size < _MAX_TURN_CONTENT_CHARS:
        ui.on_content_token(chunk)
    # Stream a few tokens PAST the cap before subscribing.
    for _ in range(3):
        ui.on_content_token(chunk)

    lq, snap = ui.register_listener_with_in_progress_snapshot()
    snap_seq = snap["seq"]

    # Live token past cap.
    ui.on_content_token(chunk)
    ev = lq.get_nowait()
    assert ev["type"] == "content"
    # The critical invariant: seq advances per-emit, so the new
    # event's _seq is strictly greater than the snap_seq the
    # subscriber captured. Without this, the events handler's
    # ``seq <= snap_seq`` filter would drop every token past the
    # cap (silent token loss for refresh-past-cap).
    assert ev["_seq"] > snap_seq, (
        f"Token past cap has _seq={ev['_seq']} which is <= "
        f"snap_seq={snap_seq} — would be silently dropped after a "
        f"refresh past the cap."
    )


def test_subscriber_after_reasoning_cap_hit_receives_subsequent_tokens() -> None:
    """Same invariant as content cap: reasoning subscribers past
    cap must keep receiving live reasoning tokens."""
    from turnstone.core.session_ui_base import _MAX_TURN_CONTENT_CHARS

    ui = _make_ui()
    chunk = "x" * 1024
    while ui._ws_inflight_reasoning_size < _MAX_TURN_CONTENT_CHARS:
        ui.on_reasoning_token(chunk)
    for _ in range(3):
        ui.on_reasoning_token(chunk)

    lq, snap = ui.register_listener_with_in_progress_snapshot()
    snap_seq = snap["seq"]

    ui.on_reasoning_token(chunk)
    ev = lq.get_nowait()
    assert ev["type"] == "reasoning"
    assert ev["_seq"] > snap_seq


def test_on_turn_committed_resets_inflight_after_commit() -> None:
    """``on_turn_committed`` fires immediately after each
    ``messages.append(assistant_msg)`` in the send loop. Without it,
    the inflight buffer keeps the just-committed turn's content
    during the post-commit tool-execution window — and a refresh in
    that window would show the assistant turn TWICE (history list
    + in_progress_snapshot)."""
    ui = _make_ui()
    ui.on_content_token("Just-finished turn ")
    ui.on_reasoning_token("Reasoning for the turn ")
    # Sanity: buffer is populated pre-commit.
    assert ui._ws_inflight_content == ["Just-finished turn "]
    assert ui._ws_inflight_reasoning == ["Reasoning for the turn "]

    ui.on_turn_committed()

    # Inflight content + reasoning reset; seq stays monotonic.
    assert ui._ws_inflight_content == []
    assert ui._ws_inflight_reasoning == []
    # Multi-turn buffer is NOT reset by commit (it drains at idle).
    assert ui._ws_turn_content == ["Just-finished turn "]


def test_inflight_snapshot_empty_during_post_commit_tool_window() -> None:
    """Models the user-reported bug: refresh during a tool-execution
    window between commit and the next stream. Pre-fix: snapshot has
    the just-committed turn's text → double-renders against history.
    Post-fix: snapshot is empty → no double-render. Seq stays
    monotonic (carries the high-water mark across turn boundaries)."""
    ui = _make_ui()
    ui.on_content_token("Calling tool with these args: ")
    seq_pre_commit = ui._event_id
    ui.on_turn_committed()  # session.py fires this after messages.append
    # We're now in the tool-execution window. A reconnecting client
    # would call register_listener_with_in_progress_snapshot.
    _, snap = ui.register_listener_with_in_progress_snapshot()
    assert snap["content"] == ""
    assert snap["reasoning"] == ""
    # Seq did NOT reset — must remain monotonic across turns.
    assert snap["seq"] == seq_pre_commit


def test_on_turn_start_resets_inflight_content_and_reasoning() -> None:
    """``on_turn_start`` clears the per-turn content + reasoning
    buffers but does NOT touch the multi-turn ``_ws_turn_content``
    (which the dashboard's IDLE-piggyback payload depends on) and
    does NOT reset the seq counter (must remain monotonic across
    turn boundaries — see ``test_inflight_seq_monotonic_across_turn_boundaries``)."""
    ui = _make_ui()
    ui.on_content_token("turn-1 ")
    ui.on_reasoning_token("reasoning-1 ")
    multi_pre = list(ui._ws_turn_content)
    multi_pre_size = ui._ws_turn_content_size

    ui.on_turn_start()

    assert ui._ws_inflight_content == []
    assert ui._ws_inflight_content_size == 0
    assert ui._ws_inflight_reasoning == []
    assert ui._ws_inflight_reasoning_size == 0
    # Multi-turn untouched.
    assert ui._ws_turn_content == multi_pre
    assert ui._ws_turn_content_size == multi_pre_size


def test_register_listener_with_in_progress_snapshot_empty() -> None:
    ui = _make_ui()
    lq, snap = ui.register_listener_with_in_progress_snapshot()
    assert isinstance(lq, queue.Queue)
    assert lq in ui._listeners
    assert snap == {"content": "", "reasoning": "", "seq": 0}


def test_register_listener_with_in_progress_snapshot_populated() -> None:
    ui = _make_ui()
    ui.on_content_token("Hello, ")
    ui.on_content_token("world!")
    ui.on_reasoning_token("planning a greeting")
    lq, snap = ui.register_listener_with_in_progress_snapshot()
    assert snap["content"] == "Hello, world!"
    assert snap["reasoning"] == "planning a greeting"
    # seq counts every successful append across BOTH buffers.
    assert snap["seq"] == 3
    # Listener is registered — later live tokens land in lq.
    ui.on_content_token(" Goodbye.")
    ev = lq.get_nowait()
    assert ev["type"] == "content"
    assert ev["text"] == " Goodbye."
    assert ev["_seq"] == 4


def test_register_listener_with_in_progress_snapshot_only_inflight_not_multi_turn() -> None:
    """The snapshot reflects the in-progress turn only — anything
    cleared by ``on_turn_start`` (a prior committed turn within the
    same send) must NOT appear in the snapshot, even though the
    multi-turn buffer still has it."""
    ui = _make_ui()
    ui.on_content_token("PRIOR_TURN ")
    ui.on_turn_start()  # commit boundary — inflight reset
    ui.on_content_token("CURRENT")
    _, snap = ui.register_listener_with_in_progress_snapshot()
    assert snap["content"] == "CURRENT"
    # Multi-turn buffer still has both turns (drives the IDLE piggyback).
    assert "".join(ui._ws_turn_content) == "PRIOR_TURN CURRENT"


def test_seq_filter_dedup_round_trip() -> None:
    """End-to-end dedup invariant: every token appears exactly once
    when reconstructing from snapshot + listener queue under live
    writes that race the registration. Models the events handler."""
    ui = _make_ui()
    for ch in "abcde":
        ui.on_content_token(ch)
    lq, snap = ui.register_listener_with_in_progress_snapshot()
    for ch in "fgh":
        ui.on_content_token(ch)

    reconstructed = snap["content"]
    while True:
        try:
            ev = lq.get_nowait()
        except queue.Empty:
            break
        if ev.get("_seq", 0) <= snap["seq"]:
            continue
        reconstructed += ev["text"]
    assert reconstructed == "abcdefgh"


def test_seq_filter_drops_overlap_when_register_lands_after_writer() -> None:
    """Race: writer appends + emits while a second register snapshots
    after the writer. The live event has _seq <= snap.seq → must be
    dropped to avoid double-render."""
    ui = _make_ui()
    # Register a first listener so the writer's enqueue lands somewhere.
    lq1, _ = ui.register_listener_with_in_progress_snapshot()
    ui.on_content_token("X")
    # Second register snapshots AFTER the write — snap has "X" AND
    # the writer's enqueue is in lq1.
    _, snap2 = ui.register_listener_with_in_progress_snapshot()
    assert snap2["content"] == "X"
    # Drain lq1 with the filter against snap2.seq — duplicate dropped.
    duped: list[str] = []
    while True:
        try:
            ev = lq1.get_nowait()
        except queue.Empty:
            break
        if ev.get("_seq", 0) <= snap2["seq"]:
            continue
        duped.append(ev["text"])
    assert duped == []


def test_inflight_seq_monotonic_across_turn_boundaries() -> None:
    """Regression: a subscriber registered mid-turn-N must still
    receive turn N+1's tokens. The seq counter is monotonic across
    turn boundaries — resetting it at on_turn_committed/on_turn_start
    would silently drop turn N+1's first M tokens (M = the snap_seq
    captured mid-turn-N) via the events handler's `seq <= snap_seq`
    filter."""
    ui = _make_ui()
    # Turn N: stream tokens, register a listener mid-turn.
    ui.on_content_token("turn-N tok1 ")
    ui.on_content_token("turn-N tok2 ")
    lq, snap = ui.register_listener_with_in_progress_snapshot()
    snap_seq = snap["seq"]
    assert snap_seq == 2
    # Turn N completes, turn N+1 begins.
    ui.on_turn_committed()
    ui.on_turn_start()
    # Turn N+1's first content token. With the q-1 fix, seq is
    # monotonic (3), not reset to 1. The events handler's
    # `seq <= snap_seq` filter must NOT swallow it.
    ui.on_content_token("turn-N+1 tok1 ")
    ev = lq.get_nowait()
    assert ev["type"] == "content"
    assert ev["text"] == "turn-N+1 tok1 "
    assert ev["_seq"] > snap_seq, (
        f"Token from turn N+1 has _seq={ev['_seq']} which is <= "
        f"snap_seq={snap_seq} — the events handler's dedup filter "
        f"would silently drop it on a long-lived SSE subscription."
    )


def test_snapshot_and_consume_drains_inflight_at_idle() -> None:
    """Regression for the cancel/error path: ``on_turn_committed`` is
    NOT called from cancel handlers, but every exit path eventually
    fires ``_emit_state("idle")`` (cancel) or ``_emit_state("error")``
    (exception). The IDLE/ERROR branches of
    ``snapshot_and_consume_state_payload`` must drain the inflight
    buffers so a refresh post-cancel doesn't double-render the
    cancelled fragment against history's marker'd version."""
    ui = _make_ui()
    ui.on_content_token("partial cancelled text ")
    ui.on_reasoning_token("partial reasoning ")
    assert ui._ws_inflight_content_size > 0
    assert ui._ws_inflight_reasoning_size > 0

    ui.snapshot_and_consume_state_payload("idle")

    assert ui._ws_inflight_content == []
    assert ui._ws_inflight_content_size == 0
    assert ui._ws_inflight_reasoning == []
    assert ui._ws_inflight_reasoning_size == 0


def test_snapshot_and_consume_drains_inflight_at_error() -> None:
    """Regression for the exception path: ERROR-branch must drain
    inflight too (parallel to the IDLE branch)."""
    ui = _make_ui()
    ui.on_content_token("partial errored text ")
    ui.on_reasoning_token("partial errored reasoning ")

    ui.snapshot_and_consume_state_payload("error")

    assert ui._ws_inflight_content == []
    assert ui._ws_inflight_reasoning == []


def test_snapshot_and_consume_does_not_reset_seq_at_idle_or_error() -> None:
    """The IDLE/ERROR drain clears content + reasoning but must NOT
    reset the seq counter — long-lived subscribers' snap_seq must
    stay valid across turn boundaries (see the q-1 invariant test)."""
    ui = _make_ui()
    ui.on_content_token("a")
    ui.on_content_token("b")
    assert ui._event_id == 2

    ui.snapshot_and_consume_state_payload("idle")
    assert ui._event_id == 2

    ui.snapshot_and_consume_state_payload("error")
    assert ui._event_id == 2


def test_listeners_share_dict_reference_warning() -> None:
    """Pinning the shape that necessitated the events-handler shallow
    copy: ``_enqueue`` puts ONE dict reference into every listener
    queue. If multiple SSE coroutines mutate (e.g. ``del event[\"_seq\"]``)
    without copying first, they corrupt each other's view. The fix
    in make_events_handler is ``event = dict(event)`` immediately
    after ``client_queue.get`` — verify the underlying invariant
    here so a future refactor of ``_enqueue`` can't silently break
    the assumption the events handler relies on."""
    ui = _make_ui()
    lq1, _ = ui.register_listener_with_in_progress_snapshot()
    lq2, _ = ui.register_listener_with_in_progress_snapshot()
    ui.on_content_token("X")
    ev1 = lq1.get_nowait()
    ev2 = lq2.get_nowait()
    # Same reference today — consumers MUST shallow-copy before any
    # mutation. If a future _enqueue change makes this no longer
    # true, the events handler's defensive copy becomes redundant
    # but harmless; if this assertion suddenly fails the underlying
    # invariant has shifted and the handler comment should be updated.
    assert ev1 is ev2


def test_concurrent_writer_and_register_with_snapshot_no_loss_no_dup() -> None:
    """Stress: many tokens streaming + a register_with_snapshot landing
    at a random point. End state: snapshot ∪ filtered_live == every
    token written, exactly once."""
    ui = _make_ui()
    n_tokens = 500
    snap_box: dict[str, Any] = {}
    lq_box: dict[str, queue.Queue[Any]] = {}

    def _writer() -> None:
        for i in range(n_tokens):
            ui.on_content_token(f"{i},")

    def _registrar() -> None:
        # Tiny sleep so the writer is mid-flight.
        threading.Event().wait(0.001)
        lq, snap = ui.register_listener_with_in_progress_snapshot()
        snap_box["snap"] = snap
        lq_box["lq"] = lq

    w = threading.Thread(target=_writer)
    r = threading.Thread(target=_registrar)
    w.start()
    r.start()
    w.join()
    r.join()

    snap = snap_box["snap"]
    lq = lq_box["lq"]
    reconstructed = snap["content"]
    while True:
        try:
            ev = lq.get_nowait()
        except queue.Empty:
            break
        if ev.get("_seq", 0) <= snap["seq"]:
            continue
        reconstructed += ev["text"]
    expected = "".join(f"{i}," for i in range(n_tokens))
    assert reconstructed == expected, (
        f"reconstruction mismatch: len(rec)={len(reconstructed)}, len(exp)={len(expected)}"
    )


# ---------------------------------------------------------------------------
# Smart Approvals (judge.smart_approvals)
# ---------------------------------------------------------------------------


class _SeedingUI(_ConcreteUI):
    """Re-delivers seeded LLM verdicts right after the approval-cycle
    reset clears the cache — simulates the async judge daemon delivering
    them via ``on_intent_verdict`` during the Smart Approvals wait, which
    is the only point at which they can land and survive the reset."""

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        super().__init__(*args, **kwargs)
        self.seed_verdicts: list[dict[str, Any]] = []

    def _reset_approval_cycle(self) -> None:
        super()._reset_approval_cycle()
        for verdict in self.seed_verdicts:
            self.on_intent_verdict(dict(verdict))


def _patch_policies(verdicts: dict[str, str]):  # type: ignore[no-untyped-def]
    """Neutralise the admin tool-policy stage so approve_tools tests
    isolate the Smart Approvals gate."""
    return patch(
        "turnstone.core.policy.evaluate_tool_policies_batch",
        return_value=verdicts,
    )


def _drain(lq: queue.Queue[Any]) -> list[dict[str, Any]]:
    """Drain all currently-queued events off a listener queue."""
    out: list[dict[str, Any]] = []
    while True:
        try:
            out.append(lq.get_nowait())
        except queue.Empty:
            return out


def _smart_ui() -> _ConcreteUI:
    ui = _make_ui()
    ui.smart_approvals_enabled = True
    ui.smart_approval_threshold = 0.95
    ui.smart_approval_wait_seconds = 1.0
    return ui


def _pending_item(call_id: str, func_name: str = "bash") -> dict[str, Any]:
    """A still-pending tool call carrying a heuristic verdict, matching
    what ``ChatSession._evaluate_intent`` attaches before the gate."""
    return {
        "call_id": call_id,
        "func_name": func_name,
        "approval_label": func_name,
        "header": f"Tool: {func_name}",
        "preview": "",
        "needs_approval": True,
        "_heuristic_verdict": {
            "verdict_id": f"h-{call_id}",
            "call_id": call_id,
            "func_name": func_name,
            "risk_level": "medium",
            "confidence": 0.5,
            "recommendation": "review",
        },
    }


def _llm_verdict(
    call_id: str,
    *,
    recommendation: str = "approve",
    confidence: float = 0.99,
    tier: str = "llm",
) -> dict[str, Any]:
    return {
        "verdict_id": f"v-{call_id}",
        "call_id": call_id,
        "func_name": "bash",
        "risk_level": "low",
        "confidence": confidence,
        "recommendation": recommendation,
        "tier": tier,
        "intent_summary": "",
        "reasoning": "",
        "evidence": [],
    }


def test_smart_approval_clears_high_confidence_llm_approve() -> None:
    ui = _smart_ui()
    item = _pending_item("c1")
    ui._llm_verdicts["c1"] = _llm_verdict("c1", recommendation="approve", confidence=0.99)
    with _patch_get_storage(MagicMock()):
        remaining = ui._apply_smart_approvals([item])
    assert remaining == []  # nothing left for a human
    assert item["needs_approval"] is False
    assert item["auto_approved"] is True
    assert item["auto_approve_reason"] == "smart_approval"


def test_smart_approval_clears_at_exact_threshold() -> None:
    """``confidence >= threshold`` — the boundary value auto-approves."""
    ui = _smart_ui()
    item = _pending_item("c1")
    ui._llm_verdicts["c1"] = _llm_verdict("c1", recommendation="approve", confidence=0.95)
    with _patch_get_storage(MagicMock()):
        remaining = ui._apply_smart_approvals([item])
    assert remaining == []
    assert item["auto_approved"] is True


def test_smart_approval_holds_just_below_threshold() -> None:
    ui = _smart_ui()
    item = _pending_item("c1")
    ui._llm_verdicts["c1"] = _llm_verdict("c1", recommendation="approve", confidence=0.94)
    with _patch_get_storage(MagicMock()):
        remaining = ui._apply_smart_approvals([item])
    assert remaining == [item]
    assert item.get("auto_approved") is not True
    assert item["needs_approval"] is True


def test_smart_approval_holds_review_and_deny() -> None:
    """Only ``approve`` auto-approves; ``review`` / ``deny`` reach a human
    no matter how confident the judge is."""
    ui = _smart_ui()
    for rec in ("review", "deny"):
        item = _pending_item("c1")
        ui._llm_verdicts = {"c1": _llm_verdict("c1", recommendation=rec, confidence=1.0)}
        with _patch_get_storage(MagicMock()):
            remaining = ui._apply_smart_approvals([item])
        assert remaining == [item], rec
        assert item.get("auto_approved") is not True, rec


def test_smart_approval_holds_llm_fallback_even_if_approve() -> None:
    """A ``llm_fallback`` verdict means the LLM stage timed out / errored
    and the row is the heuristic carry-over.  Even if it reads ``approve``
    at full confidence it must reach a human — errors require attention."""
    ui = _smart_ui()
    item = _pending_item("c1")
    ui._llm_verdicts["c1"] = _llm_verdict(
        "c1", recommendation="approve", confidence=1.0, tier="llm_fallback"
    )
    with _patch_get_storage(MagicMock()):
        remaining = ui._apply_smart_approvals([item])
    assert remaining == [item]
    assert item.get("auto_approved") is not True


def test_smart_approval_holds_when_no_verdict_arrives() -> None:
    """Wait budget elapses with no verdict cached → fail closed to the
    human gate."""
    ui = _smart_ui()
    ui.smart_approval_wait_seconds = 0.05  # nothing will be delivered
    item = _pending_item("c1")
    with _patch_get_storage(MagicMock()):
        remaining = ui._apply_smart_approvals([item])
    assert remaining == [item]
    assert item["needs_approval"] is True


def test_smart_approval_batch_atomic_holds_whole_batch_on_one_failure() -> None:
    """Batch-atomic: a single non-qualifying call (here a review) in a
    parallel batch holds the ENTIRE batch for a human — including the call
    that individually qualified.  Parallel calls are one unit of intent."""
    ui = _smart_ui()
    a = _pending_item("c1")
    b = _pending_item("c2")
    ui._llm_verdicts = {
        "c1": _llm_verdict("c1", recommendation="approve", confidence=0.99),
        "c2": _llm_verdict("c2", recommendation="review", confidence=0.99),
    }
    with _patch_get_storage(MagicMock()):
        remaining = ui._apply_smart_approvals([a, b])
    assert remaining == [a, b]  # NONE auto-approved
    assert a.get("auto_approved") is not True
    assert b.get("auto_approved") is not True


def test_smart_approval_approves_full_batch_when_all_qualify() -> None:
    """When every call in a parallel batch qualifies, the whole batch is
    auto-approved and nothing is left for a human."""
    ui = _smart_ui()
    a = _pending_item("c1")
    b = _pending_item("c2")
    ui._llm_verdicts = {
        "c1": _llm_verdict("c1", recommendation="approve", confidence=0.99),
        "c2": _llm_verdict("c2", recommendation="approve", confidence=0.96),
    }
    with _patch_get_storage(MagicMock()):
        remaining = ui._apply_smart_approvals([a, b])
    assert remaining == []
    assert a["auto_approved"] is True and b["auto_approved"] is True
    assert a["needs_approval"] is False and b["needs_approval"] is False


def test_smart_approved_item_serializes_llm_verdict_not_heuristic() -> None:
    """The auto-approved tool row must carry the driving LLM verdict
    (llm/approve) as judge_verdict so the UI doesn't render a contradictory
    heuristic 'review/medium' chip beside the SMART_APPROVAL pill."""
    ui = _smart_ui()
    item = _pending_item("c1")  # heuristic verdict is review / medium
    ui._llm_verdicts["c1"] = _llm_verdict("c1", recommendation="approve", confidence=0.99)
    with _patch_get_storage(MagicMock()):
        ui._apply_smart_approvals([item])
    serialized = _ConcreteUI._serialize_approval_items([item])[0]
    assert serialized["auto_approved"] is True
    assert serialized["auto_approve_reason"] == "smart_approval"
    judge_verdict = serialized["judge_verdict"]
    assert judge_verdict["tier"] == "llm"
    assert judge_verdict["recommendation"] == "approve"
    # Heuristic still carried, but judge_verdict is what the row renders.
    assert serialized["heuristic_verdict"]["recommendation"] == "review"


def test_smart_approval_holds_batch_when_one_call_has_no_verdict() -> None:
    """A parallel batch where one call never gets a verdict (timeout) holds
    the whole batch, even though its sibling qualified."""
    ui = _smart_ui()
    ui.smart_approval_wait_seconds = 0.05
    a = _pending_item("c1")
    b = _pending_item("c2")
    ui._llm_verdicts = {"c1": _llm_verdict("c1", recommendation="approve", confidence=0.99)}
    # c2 has no verdict — the wait times out and the batch is held.
    with _patch_get_storage(MagicMock()):
        remaining = ui._apply_smart_approvals([a, b])
    assert remaining == [a, b]
    assert a.get("auto_approved") is not True


def test_smart_approval_skips_budget_override_pseudo_tool() -> None:
    """The synthetic ``__budget_override__`` must always reach a human,
    never smart-approved."""
    ui = _smart_ui()
    item = _pending_item("c1", func_name="__budget_override__")
    ui._llm_verdicts["c1"] = _llm_verdict("c1", recommendation="approve", confidence=1.0)
    with _patch_get_storage(MagicMock()):
        remaining = ui._apply_smart_approvals([item])
    assert remaining == [item]
    assert item.get("auto_approved") is not True


def test_smart_approval_stamps_verdict_user_decision() -> None:
    """The LLM verdict arrived during the wait (parked in
    ``_pending_verdicts`` as pending); the smart stage pulls it out so a
    sibling's resolve can't re-stamp it, and records ``smart_approval`` on
    both the cached dict and the persisted row."""
    storage = MagicMock()
    ui = _smart_ui()
    item = _pending_item("c1")
    verdict = _llm_verdict("c1", recommendation="approve", confidence=0.99)
    ui._llm_verdicts["c1"] = verdict
    ui._pending_verdicts = [verdict]  # as on_intent_verdict would have parked it
    with _patch_get_storage(storage):
        ui._apply_smart_approvals([item])
    assert ui._pending_verdicts == []
    assert ui._llm_verdicts["c1"]["user_decision"] == "smart_approval"
    storage.update_intent_verdict.assert_called_once_with("v-c1", user_decision="smart_approval")


def test_approve_tools_smart_approves_whole_batch_without_prompt() -> None:
    """End-to-end through approve_tools: the verdict is delivered after
    the cache reset (via _SeedingUI), the gate auto-approves, and the
    function returns approved without ever emitting an approval prompt."""
    storage = MagicMock()
    ui = _SeedingUI(ws_id="ws-1", user_id="u1")
    ui.smart_approvals_enabled = True
    ui.smart_approval_threshold = 0.95
    ui.smart_approval_wait_seconds = 1.0
    item = _pending_item("c1")
    ui.seed_verdicts = [_llm_verdict("c1", recommendation="approve", confidence=0.99)]
    lq = ui._register_listener()
    with _patch_get_storage(storage), _patch_policies({}):
        approved, feedback = ui.approve_tools([item])
    assert approved is True
    assert feedback is None
    assert item["auto_approved"] is True
    assert item["auto_approve_reason"] == "smart_approval"
    assert item["needs_approval"] is False
    assert ui._pending_approval is None  # operator was never prompted
    assert ui._pending_verdicts == []  # smart verdict pulled out + stamped
    # No approval prompt was fanned out to listeners.
    events = []
    while True:
        try:
            events.append(lq.get_nowait()["type"])
        except queue.Empty:
            break
    assert "approve_request" not in events


def test_approve_tools_skips_smart_stage_when_disabled() -> None:
    """With Smart Approvals off (the default), a confident approve verdict
    does NOT bypass the human — approve_tools blocks on the prompt as
    before."""
    storage = MagicMock()
    ui = _SeedingUI(ws_id="ws-1", user_id="u1")
    ui.smart_approvals_enabled = False
    ui.smart_approval_wait_seconds = 1.0
    item = _pending_item("c1")
    ui.seed_verdicts = [_llm_verdict("c1", recommendation="approve", confidence=0.99)]
    timer = resolve_when_pending(ui, True, "ok")
    timer.start()
    try:
        with _patch_get_storage(storage), _patch_policies({}):
            approved, _feedback = ui.approve_tools([item])
    finally:
        timer.cancel()
    assert approved is True  # the human approved, not the judge
    assert item.get("auto_approve_reason") != "smart_approval"
    assert item.get("auto_approved") is not True


def test_await_llm_verdicts_returns_when_verdict_delivered() -> None:
    """The wait wakes as soon as the last needed verdict lands, well
    before the budget elapses."""
    ui = _smart_ui()

    def _deliver() -> None:
        with _patch_get_storage(MagicMock()):
            ui.on_intent_verdict(_llm_verdict("c1"))

    timer = threading.Timer(0.02, _deliver)
    timer.start()
    try:
        # Generous budget; should return on the notify, not the timeout.
        ui._await_llm_verdicts({"c1"}, 5.0)
    finally:
        timer.cancel()
    assert "c1" in ui._llm_verdicts


def test_smart_approval_respects_heuristic_deny_floor() -> None:
    """A high-confidence LLM ``approve`` must NOT override a deterministic
    heuristic ``deny`` — the LLM may escalate the heuristic but never lower
    it.  The call reaches a human."""
    ui = _smart_ui()
    item = _pending_item("c1")
    item["_heuristic_verdict"]["recommendation"] = "deny"
    item["_heuristic_verdict"]["risk_level"] = "critical"
    ui._llm_verdicts["c1"] = _llm_verdict("c1", recommendation="approve", confidence=1.0)
    with _patch_get_storage(MagicMock()):
        remaining = ui._apply_smart_approvals([item])
    assert remaining == [item]
    assert item.get("auto_approved") is not True
    assert item["needs_approval"] is True


def test_smart_approval_respects_heuristic_critical_floor() -> None:
    """A heuristic ``critical`` risk_level blocks smart approval even when
    the heuristic recommendation itself isn't ``deny``."""
    ui = _smart_ui()
    item = _pending_item("c1")
    item["_heuristic_verdict"]["recommendation"] = "review"
    item["_heuristic_verdict"]["risk_level"] = "critical"
    ui._llm_verdicts["c1"] = _llm_verdict("c1", recommendation="approve", confidence=1.0)
    with _patch_get_storage(MagicMock()):
        remaining = ui._apply_smart_approvals([item])
    assert remaining == [item]
    assert item.get("auto_approved") is not True


def test_smart_approval_skips_oversized_batch() -> None:
    """A batch with more calls than the FIFO verdict-cache cap can't be
    reliably awaited (older verdicts evict before the wait sees them all),
    so the whole batch reaches a human rather than stalling on the wait."""
    ui = _smart_ui()
    n = ui._LLM_VERDICT_CACHE_MAX + 1
    items = [_pending_item(f"c{i}") for i in range(n)]
    for i in range(n):
        ui._llm_verdicts[f"c{i}"] = _llm_verdict(f"c{i}", recommendation="approve", confidence=1.0)
    with _patch_get_storage(MagicMock()):
        remaining = ui._apply_smart_approvals(items)
    assert remaining == items  # none auto-approved
    assert all(it.get("auto_approved") is not True for it in items)


def test_replay_pending_verdicts_reemits_cached_verdicts() -> None:
    """The streaming-fix helper re-fans-out each pending call's cached LLM
    verdict as an intent_verdict event."""
    ui = _smart_ui()
    item = _pending_item("c1")
    ui._llm_verdicts["c1"] = _llm_verdict("c1", recommendation="review", confidence=0.9)
    lq = ui._register_listener()
    ui._replay_pending_verdicts([item])
    intent_events = []
    while True:
        try:
            ev = lq.get_nowait()
        except queue.Empty:
            break
        if ev.get("type") == "intent_verdict":
            intent_events.append(ev)
    assert len(intent_events) == 1
    assert intent_events[0]["call_id"] == "c1"
    assert intent_events[0]["recommendation"] == "review"


def test_approve_tools_reemits_verdict_after_card_on_held_batch() -> None:
    """Streaming regression fix: when Smart Approvals holds a batch (e.g. a
    review verdict), the approve_request card is FOLLOWED by a re-emitted
    intent_verdict so the live chip updates without a browser reload."""
    storage = MagicMock()
    ui = _SeedingUI(ws_id="ws-1", user_id="u1")
    ui.smart_approvals_enabled = True
    ui.smart_approval_threshold = 0.95
    ui.smart_approval_wait_seconds = 1.0
    item = _pending_item("c1")
    ui.seed_verdicts = [_llm_verdict("c1", recommendation="review", confidence=0.99)]
    lq = ui._register_listener()
    timer = resolve_when_pending(ui, False, "no")
    timer.start()
    try:
        with _patch_get_storage(storage), _patch_policies({}):
            ui.approve_tools([item])
    finally:
        timer.cancel()
    events = []
    while True:
        try:
            events.append(lq.get_nowait())
        except queue.Empty:
            break
    types = [e.get("type") for e in events]
    assert "approve_request" in types
    # An intent_verdict is re-emitted AFTER the card (the live chip update).
    ar = types.index("approve_request")
    assert "intent_verdict" in types[ar + 1 :]
    # The wait already collected the verdict, so the card must not claim the
    # judge is still working — no spurious "judge pending" spinner / poll.
    assert events[ar].get("judge_pending") is False


def test_judge_pending_true_when_llm_verdict_not_yet_cached() -> None:
    """Normal async flow (Smart Approvals off): a judged call whose LLM
    verdict hasn't arrived yet → approve_request reports judge_pending=True."""
    ui = _make_ui()  # smart_approvals_enabled defaults False
    item = _pending_item("c1")  # carries _heuristic_verdict, no cached LLM verdict
    lq = ui._register_listener()
    timer = resolve_when_pending(ui, True, "ok")
    timer.start()
    try:
        with _patch_get_storage(MagicMock()), _patch_policies({}):
            ui.approve_tools([item])
    finally:
        timer.cancel()
    reqs = [e for e in _drain(lq) if e.get("type") == "approve_request"]
    assert reqs and reqs[0]["judge_pending"] is True


def test_auto_approve_reason_vocabulary_matches_js() -> None:
    """AutoApproveReason.ALL must stay in lockstep with the JS
    KNOWN_AUTO_APPROVE_REASONS set — a server-sent reason missing from the JS
    set degrades to the 'unknown' pill on the coordinator tree."""
    import re
    from pathlib import Path

    from turnstone.core.session_ui_base import AutoApproveReason

    js = Path(__file__).resolve().parents[1] / "turnstone/console/static/coordinator/coordinator.js"
    m = re.search(
        r"KNOWN_AUTO_APPROVE_REASONS\s*=\s*new Set\(\s*\[(.*?)\]",
        js.read_text(),
        re.S,
    )
    assert m, "KNOWN_AUTO_APPROVE_REASONS set not found in coordinator.js"
    js_reasons = set(re.findall(r'"([^"]+)"', m.group(1)))
    assert js_reasons == AutoApproveReason.ALL


def test_verdict_confidence_rejects_non_finite() -> None:
    """NaN/inf confidence is treated as malformed (0.0), not clamped to 1.0."""
    assert _ConcreteUI._verdict_confidence({"confidence": float("nan")}) == 0.0
    assert _ConcreteUI._verdict_confidence({"confidence": float("inf")}) == 0.0
    assert _ConcreteUI._verdict_confidence({"confidence": 0.97}) == 0.97


def test_smart_approval_holds_nan_confidence() -> None:
    """A NaN confidence (json.loads accepts NaN) must NOT clear the
    auto-approve bar even with recommendation=approve."""
    ui = _smart_ui()
    item = _pending_item("c1")
    ui._llm_verdicts["c1"] = _llm_verdict("c1", recommendation="approve", confidence=float("nan"))
    with _patch_get_storage(MagicMock()):
        remaining = ui._apply_smart_approvals([item])
    assert remaining == [item]
    assert item.get("auto_approved") is not True


def test_smart_approval_holds_batch_with_duplicate_call_ids() -> None:
    """Two pending calls sharing a call_id (some local models emit duplicate
    non-empty ids) must not both be cleared by the single shared verdict —
    hold the whole batch."""
    ui = _smart_ui()
    a = _pending_item("dup")
    b = _pending_item("dup")  # same call_id, distinct call
    ui._llm_verdicts["dup"] = _llm_verdict("dup", recommendation="approve", confidence=0.99)
    with _patch_get_storage(MagicMock()):
        remaining = ui._apply_smart_approvals([a, b])
    assert remaining == [a, b]
    assert a.get("auto_approved") is not True


def test_on_intent_verdict_skips_append_for_already_finalized_verdict() -> None:
    """Guards the audit-corruption race: a verdict already stamped with a
    final user_decision (e.g. ``_finalize_smart_verdicts`` ran between this
    verdict's notify and its append) is NOT re-parked in _pending_verdicts,
    so a later round's resolve_approval can't overwrite its audit row."""
    ui = _make_ui()
    verdict = {"verdict_id": "v1", "call_id": "c1", "user_decision": "smart_approval"}
    with _patch_get_storage(MagicMock()):
        ui.on_intent_verdict(verdict)
    assert ui._pending_verdicts == []
    assert ui._llm_verdicts["c1"]["user_decision"] == "smart_approval"


# ---------------------------------------------------------------------------
# Early-paint (tool_pending) — render the batch before the judge / gate
# ---------------------------------------------------------------------------


def test_tool_pending_is_first_event_and_precedes_tool_info() -> None:
    """``approve_tools`` emits ``tool_pending`` as its very first event,
    before the auto-approve fall-through emits ``tool_info`` — so the UI
    paints the pending call the instant it lands, not only once the gate
    resolves.  The payload carries the serialised items (keyed by call_id)
    that the later ``tool_info`` / ``approve_request`` upgrades in place."""
    ui = _make_ui()
    lq = ui._register_listener()
    with _patch_get_storage(MagicMock()):
        # needs_approval=False → auto fall-through, no human block.
        ui.approve_tools([{"call_id": "c1", "func_name": "ls", "needs_approval": False}])
    events = _drain(lq)
    types = [e["type"] for e in events]
    assert types[0] == "tool_pending", types
    assert "tool_info" in types
    assert types.index("tool_pending") < types.index("tool_info")
    assert events[0]["items"][0]["call_id"] == "c1"


def test_tool_pending_precedes_smart_approval_gate() -> None:
    """Regression for the #621 block: pre-fix the Smart Approvals verdict
    wait sat AHEAD of the card emit, so nothing painted until the judge
    ruled.  The announce now fires at the top of ``approve_tools`` — already
    on the wire by the time the gate runs — and carries the heuristic
    verdict attached before the gate."""
    ui = _smart_ui()
    lq = ui._register_listener()
    captured: list[str] = []

    def _spy(pending: list[dict[str, Any]]) -> list[dict[str, Any]]:
        # Snapshot what the UI has already been told at gate-entry.
        captured.extend(e["type"] for e in _drain(lq))
        return []  # simulate the gate clearing the whole batch (no human, no wait)

    with patch.object(ui, "_apply_smart_approvals", side_effect=_spy), _patch_get_storage(None):
        approved, _feedback = ui.approve_tools([_pending_item("c1")])

    assert approved is True
    assert captured and captured[0] == "tool_pending", captured


# ---------------------------------------------------------------------------
# Sub-agent step tagging (task_agent child events nest under the parent card)
# ---------------------------------------------------------------------------


class TestAgentChildTagging:
    """``note_agent_child`` makes ``_enqueue`` stamp ``parent_call_id`` on a
    sub-tool's events so the UI can nest a task agent's steps under its card.
    Keyed on the immutable child call_id (correct under the parent's parallel
    tool pool); cleared when the task agent finishes."""

    def test_registered_child_event_is_stamped(self) -> None:
        ui = _make_ui()
        lq = ui._register_listener()
        ui.note_agent_child("child-1", "task-A")
        ui._enqueue({"type": "tool_result", "call_id": "child-1", "name": "bash", "output": "ok"})
        assert lq.get_nowait()["parent_call_id"] == "task-A"

    def test_unregistered_call_id_is_not_stamped(self) -> None:
        ui = _make_ui()
        lq = ui._register_listener()
        ui.note_agent_child("child-1", "task-A")
        ui._enqueue({"type": "tool_result", "call_id": "other", "name": "x", "output": "y"})
        assert "parent_call_id" not in lq.get_nowait()

    def test_no_registry_no_stamp(self) -> None:
        """Empty registry short-circuits — events pass through untouched."""
        ui = _make_ui()
        lq = ui._register_listener()
        ui._enqueue({"type": "tool_result", "call_id": "child-1", "name": "x", "output": "y"})
        assert "parent_call_id" not in lq.get_nowait()

    def test_items_payload_is_stamped_per_entry(self) -> None:
        """approve_request / tool_pending carry an ``items`` list; each child
        entry is tagged independently, leaving non-child entries alone."""
        ui = _make_ui()
        lq = ui._register_listener()
        ui.note_agent_child("child-1", "task-A")
        ui._enqueue(
            {
                "type": "tool_pending",
                "items": [
                    {"call_id": "child-1", "func_name": "bash"},
                    {"call_id": "top-level", "func_name": "search"},
                ],
            }
        )
        items = lq.get_nowait()["items"]
        assert items[0]["parent_call_id"] == "task-A"
        assert "parent_call_id" not in items[1]

    def test_clear_agent_children_stops_stamping(self) -> None:
        ui = _make_ui()
        lq = ui._register_listener()
        ui.note_agent_child("child-1", "task-A")
        ui.clear_agent_children("task-A")
        ui._enqueue({"type": "tool_result", "call_id": "child-1", "name": "x", "output": "y"})
        assert "parent_call_id" not in lq.get_nowait()

    def test_clear_is_scoped_to_one_parent(self) -> None:
        """Two task agents in flight: clearing one leaves the other's children
        tagged — the parallel-pool invariant."""
        ui = _make_ui()
        lq = ui._register_listener()
        ui.note_agent_child("child-A", "task-A")
        ui.note_agent_child("child-B", "task-B")
        ui.clear_agent_children("task-A")
        ui._enqueue({"type": "tool_result", "call_id": "child-B", "name": "x", "output": "y"})
        assert lq.get_nowait()["parent_call_id"] == "task-B"


class TestAgentScopeInfoSuppression:
    """While a task agent runs, its ``on_info`` progress chatter ("[task done] N
    chars", a tool's "fetched N chars") carries no call_id, so it can't nest
    under the task card.  The web pane drops it for the duration rather than let
    it escape to the top level; the per-thread contextvar keeps it correct under
    the parent's parallel task pool (siblings in other threads aren't suppressed)."""

    @pytest.fixture(autouse=True)
    def _reset_scope(self):
        # The scope depth is a module-level contextvar that persists across tests
        # in the same thread; reset it around each so an unbalanced test (or a
        # leak from elsewhere) can't bleed suppression into another test.
        from turnstone.core.session_ui_base import _agent_scope_var

        token = _agent_scope_var.set(0)
        yield
        _agent_scope_var.reset(token)

    def test_on_info_suppressed_within_scope(self) -> None:
        ui = _make_ui()
        lq = ui._register_listener()
        ui.begin_agent_scope()
        ui.on_info("fetched 5663 chars, extracting...")
        ui.end_agent_scope()
        assert lq.empty()

    def test_on_info_passes_through_outside_scope(self) -> None:
        ui = _make_ui()
        lq = ui._register_listener()
        ui.on_info("top-level status")
        assert lq.get_nowait() == {
            "type": "info",
            "message": "top-level status",
            "ws_id": "ws-1",
            "_event_id": 1,
        }

    def test_nested_scopes_need_matching_exits(self) -> None:
        """Parallel task agents: info stays suppressed until the LAST one
        leaves (the depth returns to zero)."""
        ui = _make_ui()
        lq = ui._register_listener()
        ui.begin_agent_scope()
        ui.begin_agent_scope()
        ui.end_agent_scope()
        ui.on_info("still inside a sibling task agent")
        assert lq.empty()
        ui.end_agent_scope()
        ui.on_info("now top-level again")
        assert lq.get_nowait()["message"] == "now top-level again"

    def test_end_scope_floored_at_zero(self) -> None:
        """An unmatched ``end_agent_scope`` can't drive the depth negative and
        wedge suppression off."""
        ui = _make_ui()
        lq = ui._register_listener()
        ui.end_agent_scope()
        ui.begin_agent_scope()
        ui.on_info("suppressed")
        assert lq.empty()


class TestAgentTrajectoryStash:
    """The recall store retains a finished task agent's projected sub-trajectory
    keyed by call_id, LRU-bounded.  A miss is the honest "not retained" signal —
    /history then renders the flat parent record, never a fabricated 0-step card."""

    def test_stash_and_get_roundtrip(self) -> None:
        ui = _make_ui()
        steps = [
            {"id": "t1::c1", "name": "search", "arguments": "{}", "output": "ok", "is_error": False}
        ]
        ui.stash_agent_trajectory("t1", steps)
        assert ui.get_agent_trajectory("t1") == steps

    def test_missing_returns_none(self) -> None:
        assert _make_ui().get_agent_trajectory("nope") is None

    def test_empty_call_id_ignored(self) -> None:
        ui = _make_ui()
        ui.stash_agent_trajectory("", [{"id": "x"}])
        assert ui.get_agent_trajectory("") is None

    def test_restash_updates_value(self) -> None:
        ui = _make_ui()
        ui.stash_agent_trajectory("k", [{"id": "v1"}])
        ui.stash_agent_trajectory("k", [{"id": "v2"}])
        assert ui.get_agent_trajectory("k") == [{"id": "v2"}]

    def test_lru_evicts_oldest(self) -> None:
        from turnstone.core.session_ui_base import _AGENT_TRAJECTORY_CAP

        ui = _make_ui()
        for i in range(_AGENT_TRAJECTORY_CAP + 3):
            ui.stash_agent_trajectory(f"t{i}", [{"id": f"t{i}"}])
        # The three oldest fell out → honest None; the newest is retained.
        assert ui.get_agent_trajectory("t0") is None
        assert ui.get_agent_trajectory("t2") is None
        assert ui.get_agent_trajectory(f"t{_AGENT_TRAJECTORY_CAP + 2}") is not None