turnstone/tests/test_webui_content.py

"""Tests for WebUI content accumulation — server-side single source of truth."""

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

import pytest

from turnstone.server import WebUI


@pytest.fixture(autouse=True)
def _reset_global_queue():
    """Ensure WebUI._global_queue is set for tests and cleaned up after."""
    WebUI._global_queue = queue.Queue()
    yield
    WebUI._global_queue = None


def _make_ui() -> WebUI:
    """Create a WebUI with a global queue for capturing broadcast events."""
    return WebUI(ws_id="ws-test")


def _drain_global() -> list[dict]:
    """Drain all events from the global queue."""
    events = []
    assert WebUI._global_queue is not None
    while not WebUI._global_queue.empty():
        events.append(WebUI._global_queue.get_nowait())
    return events


def test_public_intent_verdict_waits_for_storage_and_records_llm_metric() -> None:
    """The public UI hook keeps its synchronous persistence contract.

    Splitting live publication from audit I/O for the session's judge callback
    must not make direct WebUI callers fire-and-forget.  Its LLM metric remains
    part of live publication and the public call returns only after the UPSERT.
    """
    ui = _make_ui()
    storage = MagicMock()
    persistence_started = threading.Event()
    release_persistence = threading.Event()
    returned = threading.Event()

    def blocked_upsert(**_kwargs) -> None:
        persistence_started.set()
        if not release_persistence.wait(5):
            raise RuntimeError("test verdict persistence was not released")

    storage.upsert_intent_verdict.side_effect = blocked_upsert
    metrics = MagicMock()
    verdict = {
        "verdict_id": "v-web-llm",
        "call_id": "call-web-llm",
        "tier": "llm",
        "risk_level": "high",
        "latency_ms": 37,
    }

    def publish() -> None:
        ui.on_intent_verdict(verdict)
        returned.set()

    with (
        patch("turnstone.core.storage._registry.get_storage", return_value=storage),
        patch("turnstone.server._metrics", metrics),
    ):
        thread = threading.Thread(target=publish)
        thread.start()
        try:
            assert persistence_started.wait(2)
            metrics.record_judge_verdict.assert_called_once_with("llm", "high", 37)
            assert not returned.is_set()
            assert thread.is_alive()
        finally:
            release_persistence.set()
            thread.join(2)

    assert not thread.is_alive()
    assert returned.is_set()
    storage.upsert_intent_verdict.assert_called_once()


class TestContentAccumulation:
    """WebUI should accumulate content tokens and include in idle broadcast."""

    def test_content_token_accumulates(self, monkeypatch):
        """on_content_token should append to _ws_turn_content.

        Batch window forced to 0 (per-token flush) — this pins the
        accumulator wiring, not the emit-time batching cadence (which
        coalesces fragments; see test_sse_token_batching.py)."""

        monkeypatch.setattr("turnstone.core.session_ui_base._TOKEN_BATCH_WINDOW_SECS", 0.0)
        ui = _make_ui()
        ui.on_content_token("Hello ")
        ui.on_content_token("world")
        assert ui._ws_turn_content == ["Hello ", "world"]

    def test_idle_broadcast_includes_content(self):
        """_broadcast_state('idle') should include joined content and reset."""
        ui = _make_ui()
        ui.on_content_token("Hello ")
        ui.on_content_token("world")
        ui._broadcast_state("idle")

        events = _drain_global()
        idle_events = [e for e in events if e.get("state") == "idle"]
        assert len(idle_events) == 1
        assert idle_events[0]["content"] == "Hello world"
        # Accumulator should be reset
        assert ui._ws_turn_content == []
        assert ui._ws_turn_content_size == 0

    def test_error_broadcast_resets_without_content(self):
        """_broadcast_state('error') should reset accumulator without content in event."""
        ui = _make_ui()
        ui.on_content_token("partial")
        ui._broadcast_state("error")

        events = _drain_global()
        error_events = [e for e in events if e.get("state") == "error"]
        assert len(error_events) == 1
        assert "content" not in error_events[0]
        assert ui._ws_turn_content == []
        assert ui._ws_turn_content_size == 0

    def test_persistence_refresh_is_operator_only_and_non_consuming(self):
        ui = _make_ui()
        ui._ws_turn_content = ["preserve me"]
        ui._ws_turn_content_size = len("preserve me")
        session = MagicMock()
        session.conversation_persistence_status = lambda: {"state": "conflict"}
        ui.bind_session(session)
        ws = MagicMock()
        ws.state.value = "error"
        # The registry row deliberately disagrees: the persistence field
        # must come from the BOUND session (an id-reuse replacement row
        # reporting healthy must not launder the badge).
        ws.session.conversation_persistence_status = lambda: {"state": "healthy"}
        mgr = MagicMock()
        mgr.get.return_value = ws
        WebUI._workstream_mgr = mgr
        try:
            ui.on_persistence_state_changed()
        finally:
            WebUI._workstream_mgr = None

        event = _drain_global()[0]
        assert event["type"] == "ws_state"
        assert event["state"] == "error"
        assert event["persistence_state"] == "conflict"
        assert "content" not in event
        assert ui._ws_turn_content == ["preserve me"]
        assert ui._ws_turn_content_size == len("preserve me")

    def test_persistence_state_derives_from_bound_session_not_registry(self):
        """The badge keeps telling the truth while the row is out of the
        map: failed-delete tombstone retention and retirement pop the
        registry entry exactly when the journal needs the operator, and
        the old by-id lookup laundered that window into "healthy"."""
        ui = _make_ui()
        session = MagicMock()
        session.conversation_persistence_status = lambda: {"state": "conflict"}
        ui.bind_session(session)
        # No manager wired at all — the registry cannot be consulted.
        assert WebUI._workstream_mgr is None
        assert ui._current_persistence_state() == "conflict"
        # Unbound (or collected) still fails closed to healthy.
        assert _make_ui()._current_persistence_state() == "healthy"

    def test_thinking_broadcast_does_not_touch_accumulator(self):
        """_broadcast_state('thinking') should not affect the accumulator."""
        ui = _make_ui()
        ui.on_content_token("in progress")
        ui._broadcast_state("thinking")

        assert ui._ws_turn_content == ["in progress"]
        events = _drain_global()
        thinking_events = [e for e in events if e.get("state") == "thinking"]
        assert len(thinking_events) == 1
        assert "content" not in thinking_events[0]

    def test_multi_round_accumulation(self):
        """Content from multiple streaming rounds accumulates before idle."""
        ui = _make_ui()
        # Round 1
        ui.on_content_token("I'll check ")
        ui.on_content_token("that. ")
        # Round 2 (after tool execution)
        ui.on_content_token("Here's ")
        ui.on_content_token("the result.")
        ui._broadcast_state("idle")

        events = _drain_global()
        idle_events = [e for e in events if e.get("state") == "idle"]
        assert len(idle_events) == 1
        assert idle_events[0]["content"] == "I'll check that. Here's the result."

    def test_empty_content_on_idle_without_tokens(self):
        """idle with no content tokens should include empty content string."""
        ui = _make_ui()
        ui._broadcast_state("idle")

        events = _drain_global()
        idle_events = [e for e in events if e.get("state") == "idle"]
        assert len(idle_events) == 1
        assert idle_events[0]["content"] == ""

    def test_cancellation_preserves_partial_content(self):
        """Partial content accumulated before cancel should appear in idle event."""
        ui = _make_ui()
        ui.on_content_token("I'll ")
        ui.on_content_token("start by...")
        # Cancellation triggers idle broadcast with partial content
        ui._broadcast_state("idle")

        events = _drain_global()
        idle_events = [e for e in events if e.get("state") == "idle"]
        assert len(idle_events) == 1
        assert idle_events[0]["content"] == "I'll start by..."

    def test_consecutive_turns_isolated(self):
        """Content from turn 1 should not leak into turn 2."""
        ui = _make_ui()
        # Turn 1
        ui.on_content_token("first response")
        ui._broadcast_state("idle")
        _drain_global()

        # Turn 2
        ui.on_content_token("second response")
        ui._broadcast_state("idle")

        events = _drain_global()
        idle_events = [e for e in events if e.get("state") == "idle"]
        assert len(idle_events) == 1
        assert idle_events[0]["content"] == "second response"

    def test_content_cap_prevents_unbounded_growth(self):
        """Content exceeding the cap should stop accumulating."""
        # Constant lifted from turnstone.server to turnstone.core.session_ui_base
        # in the rich ws_state payload work so coord enforces the same ceiling.
        from turnstone.core.session_ui_base import _MAX_TURN_CONTENT_CHARS

        ui = _make_ui()
        # Fill to capacity
        chunk = "x" * 1024
        for _ in range(_MAX_TURN_CONTENT_CHARS // 1024 + 10):
            ui.on_content_token(chunk)

        assert ui._ws_turn_content_size <= _MAX_TURN_CONTENT_CHARS + 1024
        ui._broadcast_state("idle")

        events = _drain_global()
        idle_events = [e for e in events if e.get("state") == "idle"]
        assert len(idle_events) == 1
        # Content should be capped, not contain everything
        assert len(idle_events[0]["content"]) <= _MAX_TURN_CONTENT_CHARS + 1024


class TestPendingApprovalDetailNotPiggybacked:
    """Stage 3 cleanup — ``pending_approval_detail`` is no longer
    piggybacked on ``ws_state`` events. Approval items now arrive via
    bulk fetch when the coord tree's reducer sees the
    ``activity_state="approval"`` transition; verdicts via the explicit
    ``intent_verdict`` event class; resolution via
    ``approval_resolved``. These tests lock the no-piggyback contract
    down so a future regression doesn't silently re-introduce the
    duplicated path."""

    def test_state_broadcast_omits_field_when_no_approval_pending(self):
        ui = _make_ui()
        assert ui._pending_approval is None
        ui._broadcast_state("running")

        events = _drain_global()
        running_events = [e for e in events if e.get("state") == "running"]
        assert len(running_events) == 1
        assert "pending_approval_detail" not in running_events[0]

    def test_state_broadcast_omits_field_even_when_approval_pending(self):
        """The piggyback is gone: even when ``_pending_approval`` is set,
        the state broadcast must NOT carry ``pending_approval_detail``.
        The browser triggers a bulk fetch off the
        ``activity_state="approval"`` transition to get the items."""
        ui = _make_ui()
        ui._pending_approval = {
            "type": "approve_request",
            "items": [
                {
                    "call_id": "c1",
                    "header": "tool x",
                    "func_args": "{}",
                    "intent_summary": "do x",
                    "needs_approval": True,
                }
            ],
            "judge_pending": False,
        }
        ui._broadcast_state("attention")

        events = _drain_global()
        attn = [e for e in events if e.get("state") == "attention"]
        assert len(attn) == 1
        assert "pending_approval_detail" not in attn[0]

    def test_field_stays_absent_after_approval_resolves(self):
        ui = _make_ui()
        ui._pending_approval = {
            "type": "approve_request",
            "items": [{"call_id": "c1", "header": "x"}],
            "judge_pending": False,
        }
        ui._broadcast_state("attention")
        _drain_global()

        ui._pending_approval = None
        ui._broadcast_state("running")
        events = _drain_global()
        running = [e for e in events if e.get("state") == "running"]
        assert len(running) == 1
        assert "pending_approval_detail" not in running[0]


class TestBroadcastIntentVerdict:
    """Producer-side coverage for ``WebUI._broadcast_intent_verdict``.

    The collector-side test (``test_apply_delta_intent_verdict_*`` in
    test_console.py) covers consumption; this pins the event shape the
    producer puts on the global queue. A field rename or missed key
    here would slip past the consumer test because the consumer reads
    via ``data.get(...)``.
    """

    def test_pushes_intent_verdict_event_to_global_queue(self):
        ui = _make_ui()
        verdict = {
            "call_id": "c1",
            "risk_level": "low",
            "confidence": 0.92,
            "recommendation": "approve",
            "reasoning": "tool reads only",
        }
        ui._broadcast_intent_verdict(verdict)

        events = _drain_global()
        assert len(events) == 1
        ev = events[0]
        assert ev["type"] == "intent_verdict"
        assert ev["ws_id"] == "ws-test"
        assert ev["verdict"] == verdict

    def test_no_op_when_global_queue_unset(self):
        WebUI._global_queue = None
        ui = _make_ui()
        # Doesn't raise.
        ui._broadcast_intent_verdict({"call_id": "c1"})

    def test_queue_full_swallowed(self):
        # Force a tiny queue then fill it so the next put_nowait
        # raises queue.Full — the broadcast must absorb it without
        # propagating (matches _broadcast_state's queue.Full handling).
        WebUI._global_queue = queue.Queue(maxsize=1)
        WebUI._global_queue.put_nowait({"sentinel": True})
        ui = _make_ui()
        # Doesn't raise.
        ui._broadcast_intent_verdict({"call_id": "c1"})


class TestBroadcastApprovalResolved:
    """Producer-side coverage for ``WebUI._broadcast_approval_resolved``."""

    def test_pushes_approval_resolved_event_to_global_queue(self):
        ui = _make_ui()
        ui._broadcast_approval_resolved(True, "lgtm", always=False)

        events = _drain_global()
        assert len(events) == 1
        ev = events[0]
        assert ev["type"] == "approval_resolved"
        assert ev["ws_id"] == "ws-test"
        assert ev["approved"] is True
        assert ev["feedback"] == "lgtm"
        assert ev["always"] is False

    def test_normalises_none_feedback_to_empty_string(self):
        ui = _make_ui()
        ui._broadcast_approval_resolved(False, None)

        events = _drain_global()
        assert events[0]["feedback"] == ""
        assert events[0]["approved"] is False
        assert events[0]["always"] is False

    def test_always_kwarg_propagates(self):
        ui = _make_ui()
        ui._broadcast_approval_resolved(True, "ok", always=True)
        events = _drain_global()
        assert events[0]["always"] is True

    def test_no_op_when_global_queue_unset(self):
        WebUI._global_queue = None
        ui = _make_ui()
        # Doesn't raise.
        ui._broadcast_approval_resolved(True, None)


class TestBroadcastApproveRequest:
    """Producer-side coverage for ``WebUI._broadcast_approve_request`` —
    push path for the initial approval items so a coord parent's tree
    UI can render the inline approve/deny block immediately without
    waiting for a bulk-fetch round-trip."""

    def test_pushes_approve_request_event_to_global_queue(self):
        ui = _make_ui()
        detail = {
            "type": "approve_request",
            "items": [{"call_id": "c1", "header": "tool x"}],
            "judge_pending": True,
        }
        ui._broadcast_approve_request(detail)

        events = _drain_global()
        assert len(events) == 1
        ev = events[0]
        assert ev["type"] == "approve_request"
        assert ev["ws_id"] == "ws-test"
        assert ev["detail"] == detail

    def test_no_op_when_global_queue_unset(self):
        WebUI._global_queue = None
        ui = _make_ui()
        # Doesn't raise.
        ui._broadcast_approve_request({"items": []})