tests/test_progress_workflow.py

"""Tests for ProgressTracker + bridge integration + spec source-level guards.

See `projects/andy-ai-employee/specs/realtime-progress-events.md` §13 for the
acceptance criteria each test maps to.

M1a — ProgressTracker unit surface + static guards:
  T-1, T-2, T-9, T-10, T-11, T-13, T-15
M1b — bridge integration + duplication regression + cancellation guards:
  T-3, T-4, T-5, T-6, T-7, T-8, T-14, T-16
"""
from __future__ import annotations

import ast
import asyncio
import importlib
import inspect
import time
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Any, AsyncIterator

import pytest

from app.progress_workflow import ProgressTracker
from chatkit.types import (
    AssistantMessageContentPartTextDelta,
    AssistantMessageItem,
    CustomTask,
    ProgressUpdateEvent,
    ThreadItemAddedEvent,
    ThreadItemDoneEvent,
    ThreadItemUpdatedEvent,
    ThreadMetadata,
    ThreadUpdatedEvent,
    UserMessageItem,
    UserMessageTextContent,
    WorkflowItem,
    WorkflowTaskAdded,
    WorkflowTaskUpdated,
)


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


async def _drain(gen) -> list:
    out = []
    async for ev in gen:
        out.append(ev)
    return out


# --------------------------------------------------------------------------- #
# T-1 — Lazy-open + add_task contract                                         #
# --------------------------------------------------------------------------- #


@pytest.mark.asyncio
async def test_t1_first_add_task_yields_thread_item_added() -> None:
    """T-1: first add_task yields ThreadItemAddedEvent with WorkflowItem;
    a second add_task yields ThreadItemUpdatedEvent(WorkflowTaskAdded)."""
    tracker = ProgressTracker(thread_id="thread_test", enabled=True)

    first = await _drain(tracker.add_task(title="Thinking", icon="sparkle"))
    assert len(first) == 1
    ev = first[0]
    assert isinstance(ev, ThreadItemAddedEvent)
    assert isinstance(ev.item, WorkflowItem)
    assert ev.item.type == "workflow"
    assert ev.item.thread_id == "thread_test"
    assert ev.item.workflow.type == "custom"
    assert len(ev.item.workflow.tasks) == 1
    assert ev.item.workflow.tasks[0].title == "Thinking"
    assert ev.item.workflow.tasks[0].status_indicator == "loading"
    assert tracker.last_index == 0
    assert tracker.opened is True

    second = await _drain(tracker.add_task(title="Calling Bash", icon="bolt"))
    assert len(second) == 1
    ev2 = second[0]
    assert isinstance(ev2, ThreadItemUpdatedEvent)
    assert ev2.item_id == ev.item.id
    assert isinstance(ev2.update, WorkflowTaskAdded)
    assert ev2.update.task_index == 1
    assert ev2.update.task.title == "Calling Bash"
    assert tracker.last_index == 1


# --------------------------------------------------------------------------- #
# T-2 — update_task mutates in place                                          #
# --------------------------------------------------------------------------- #


@pytest.mark.asyncio
async def test_t2_update_task_mutates_in_place() -> None:
    """T-2: update_task yields ThreadItemUpdatedEvent(WorkflowTaskUpdated),
    same item_id, same task_index, mutated status_indicator."""
    tracker = ProgressTracker(thread_id="thread_test", enabled=True)
    added = await _drain(tracker.add_task(title="Calling Bash", icon="bolt"))
    item_id = added[0].item.id

    updated = await _drain(
        tracker.update_task(0, title="Bash done", status="complete")
    )
    assert len(updated) == 1
    ev = updated[0]
    assert isinstance(ev, ThreadItemUpdatedEvent)
    assert ev.item_id == item_id
    assert isinstance(ev.update, WorkflowTaskUpdated)
    assert ev.update.task_index == 0
    assert ev.update.task.status_indicator == "complete"
    assert ev.update.task.title == "Bash done"

    # Workflow tasks count must still be 1 — no new task created.
    assert len(tracker.item.workflow.tasks) == 1


# --------------------------------------------------------------------------- #
# T-9 — expanded=False at open and at done                                    #
# --------------------------------------------------------------------------- #


@pytest.mark.asyncio
async def test_t9_workflow_expanded_false_at_open_and_done() -> None:
    """T-9: workflow.expanded is False both at open and at end_workflow."""
    tracker = ProgressTracker(thread_id="thread_test", enabled=True)
    added = await _drain(tracker.add_task(title="Thinking", icon="sparkle"))
    assert added[0].item.workflow.expanded is False

    done = await _drain(tracker.end_workflow())
    assert len(done) == 1
    assert isinstance(done[0], ThreadItemDoneEvent)
    assert isinstance(done[0].item, WorkflowItem)
    assert done[0].item.workflow.expanded is False


# --------------------------------------------------------------------------- #
# T-13 — Lazy open: zero progress signals → no workflow item                  #
# --------------------------------------------------------------------------- #


@pytest.mark.asyncio
async def test_t13_no_add_task_means_no_workflow_emitted() -> None:
    """T-13: if no task is added, end_workflow yields nothing — the bridge
    must not emit a phantom WorkflowItem on instant-completion turns."""
    tracker = ProgressTracker(thread_id="thread_test", enabled=True)
    done = await _drain(tracker.end_workflow())
    assert done == []
    assert tracker.opened is False
    assert tracker.item is None


# --------------------------------------------------------------------------- #
# Disabled-path no-op                                                         #
# --------------------------------------------------------------------------- #


@pytest.mark.asyncio
async def test_disabled_tracker_yields_nothing() -> None:
    tracker = ProgressTracker(thread_id="thread_test", enabled=False)
    assert await _drain(tracker.add_task(title="Thinking")) == []
    assert await _drain(tracker.update_task(0, title="x", status="complete")) == []
    assert await _drain(tracker.end_workflow()) == []
    assert tracker.item is None
    assert tracker.opened is False


# --------------------------------------------------------------------------- #
# Coalescing — sub-1s repeats collapse, terminal status always emits          #
# --------------------------------------------------------------------------- #


@pytest.mark.asyncio
async def test_sub_1s_loading_updates_coalesce() -> None:
    """Rapid <1s loading updates on the same task are dropped (last-wins),
    but state is still mutated. A terminal `complete` update is always
    emitted."""
    tracker = ProgressTracker(thread_id="thread_test", enabled=True)
    await _drain(tracker.add_task(title="Delegated", icon="agent"))

    # Two updates within 1s — second should be coalesced.
    out1 = await _drain(tracker.update_task(0, title="Delegated · 5s"))
    out2 = await _drain(tracker.update_task(0, title="Delegated · 7s"))
    assert len(out1) == 1
    assert len(out2) == 0  # coalesced
    # State still last-wins.
    assert tracker.item.workflow.tasks[0].title == "Delegated · 7s"

    # Terminal status change must always emit, even within 1s.
    out3 = await _drain(tracker.update_task(0, status="complete"))
    assert len(out3) == 1
    assert out3[0].update.task.status_indicator == "complete"


# --------------------------------------------------------------------------- #
# end_workflow flips leftover loading → complete and sets DurationSummary     #
# --------------------------------------------------------------------------- #


@pytest.mark.asyncio
async def test_end_workflow_finalizes_state() -> None:
    tracker = ProgressTracker(thread_id="thread_test", enabled=True)
    await _drain(tracker.add_task(title="Thinking", icon="sparkle"))
    await _drain(tracker.add_task(title="Calling Bash", icon="bolt"))
    # Leave both in loading. end_workflow must flip them to complete.
    done = await _drain(tracker.end_workflow())
    item = done[0].item
    assert all(t.status_indicator == "complete" for t in item.workflow.tasks)
    # A second end_workflow is a no-op.
    assert await _drain(tracker.end_workflow()) == []


# --------------------------------------------------------------------------- #
# T-10 — Static check: no ChatKit subclassing / monkey-patching               #
# --------------------------------------------------------------------------- #


_BRIDGE_ROOT = Path(__file__).resolve().parent.parent
_WORKFLOW_PATH = _BRIDGE_ROOT / "app" / "progress_workflow.py"
_SERVER_PATH = _BRIDGE_ROOT / "app" / "openclaw_chatkit_server.py"


def _chatkit_type_names() -> set[str]:
    return {
        "WorkflowItem",
        "Workflow",
        "CustomTask",
        "SearchTask",
        "FileTask",
        "ThoughtTask",
        "ImageTask",
        "CustomSummary",
        "DurationSummary",
        "WorkflowTaskAdded",
        "WorkflowTaskUpdated",
        "ThreadItemAddedEvent",
        "ThreadItemUpdatedEvent",
        "ThreadItemDoneEvent",
        "ProgressUpdateEvent",
        "ThreadStreamEvent",
    }


def test_t10_no_chatkit_subclassing_or_monkey_patching() -> None:
    """T-10: progress_workflow.py and openclaw_chatkit_server.py MUST NOT
    subclass any ChatKit primitive, MUST NOT monkey-patch the chatkit
    module, and MUST NOT use setattr against chatkit attributes."""
    chatkit_names = _chatkit_type_names()
    # Substring "monkey" can appear in benign contexts (a docstring noting
    # that tests use pytest's `monkeypatch` fixture, for example), so the
    # check below is purely AST-level: no subclassing of ChatKit types and
    # no `setattr(chatkit, …)` calls.
    for path in (_WORKFLOW_PATH, _SERVER_PATH):
        src = path.read_text()
        tree = ast.parse(src)
        for node in ast.walk(tree):
            if isinstance(node, ast.ClassDef):
                for base in node.bases:
                    base_name = (
                        base.id
                        if isinstance(base, ast.Name)
                        else getattr(base, "attr", None)
                    )
                    assert base_name not in chatkit_names, (
                        f"{path.name}:{node.name} subclasses ChatKit type {base_name}"
                    )
            if isinstance(node, ast.Call):
                func = node.func
                fname = (
                    func.id
                    if isinstance(func, ast.Name)
                    else getattr(func, "attr", None)
                )
                if fname == "setattr" and node.args:
                    target = node.args[0]
                    target_name = (
                        target.id
                        if isinstance(target, ast.Name)
                        else getattr(target, "attr", None)
                    )
                    assert target_name != "chatkit", (
                        f"{path.name} uses setattr against chatkit module"
                    )


# --------------------------------------------------------------------------- #
# T-15 — No `::evt::` sentinel substring anywhere                             #
# --------------------------------------------------------------------------- #


def test_t15_no_sentinel_substring_in_bridge_or_workflow() -> None:
    """T-15: the prior sentinel-tail design is removed. Static check
    enforces the `::evt::` substring does not appear in either file."""
    for path in (_WORKFLOW_PATH, _SERVER_PATH):
        assert "::evt::" not in path.read_text(), (
            f"{path.name} contains the forbidden ::evt:: sentinel"
        )


# --------------------------------------------------------------------------- #
# T-11 — Microbench: 1k update_task calls p99 ≤ 0.5ms                         #
# --------------------------------------------------------------------------- #


@pytest.mark.asyncio
async def test_t11_update_task_microbench_warn_only() -> None:
    """T-11: 1k update_task emit cycles p99 ≤ 0.5ms.

    Marked warn-only (the spec says CI warn-only). We assert a generous
    upper bound to catch egregious regressions but print the measured
    p99 for visibility.
    """
    tracker = ProgressTracker(thread_id="thread_test", enabled=True)
    await _drain(tracker.add_task(title="Bench", icon="sparkle"))

    # Disable coalescing for accurate per-call measurement by using
    # alternating status flips (each call emits because complete is terminal,
    # then we reset to loading via title-only update which coalesces — to
    # bench the emit path we just hammer the title change with status=complete
    # interleaved with status=loading on a fresh task).
    samples_us: list[float] = []
    # New task per iteration would amortize state mutation; instead we
    # measure the pure update_task generator drive cost.
    for i in range(1000):
        # Force a fresh window so coalescing doesn't suppress emits.
        tracker._last_emit_at[0] = 0.0  # type: ignore[attr-defined]
        start = time.perf_counter()
        async for _ in tracker.update_task(0, title=f"t{i}"):
            pass
        samples_us.append((time.perf_counter() - start) * 1e6)

    samples_us.sort()
    p99 = samples_us[int(0.99 * len(samples_us)) - 1]
    print(f"\n[T-11] update_task p99 = {p99:.1f} µs over 1000 samples")
    # 5ms ceiling — well above the spec 0.5ms target — catches gross regressions
    # without flaking on slow CI runners. The print line is the canonical metric.
    assert p99 < 5000.0, f"update_task p99 exceeded 5ms: {p99:.1f} µs"


# =========================================================================== #
# M1b — Bridge integration                                                    #
# =========================================================================== #


# --------------------------------------------------------------------------- #
# Fake gateway stream supporting `tool_calls` deltas                          #
# --------------------------------------------------------------------------- #


@dataclass
class _Fn:
    name: str | None = None


@dataclass
class _ToolCall:
    function: _Fn = field(default_factory=_Fn)


@dataclass
class _Delta:
    content: str | None = None
    tool_calls: list[_ToolCall] = field(default_factory=list)


@dataclass
class _Choice:
    delta: _Delta


@dataclass
class _Chunk:
    choices: list[_Choice]


def _text_chunk(text: str) -> _Chunk:
    return _Chunk(choices=[_Choice(delta=_Delta(content=text))])


def _tool_chunk(name: str) -> _Chunk:
    return _Chunk(
        choices=[
            _Choice(
                delta=_Delta(content=None, tool_calls=[_ToolCall(function=_Fn(name=name))])
            )
        ]
    )


class _FakeStream:
    def __init__(self, chunks: list[_Chunk]) -> None:
        self._chunks = chunks

    def __aiter__(self) -> AsyncIterator[_Chunk]:
        return self._iter()

    async def _iter(self) -> AsyncIterator[_Chunk]:
        for c in self._chunks:
            yield c


@dataclass
class _NonStreamMessage:
    content: str


@dataclass
class _NonStreamChoice:
    message: _NonStreamMessage


@dataclass
class _NonStreamResponse:
    choices: list[_NonStreamChoice]


class _FakeCompletions:
    def __init__(self, chunks: list[_Chunk], *, raise_on_create: bool = False) -> None:
        self._chunks = chunks
        self._raise_on_create = raise_on_create
        self.calls = 0

    async def create(self, **kwargs: Any):
        self.calls += 1
        if kwargs.get("stream"):
            if self._raise_on_create:
                raise RuntimeError("simulated gateway failure")
            return _FakeStream(self._chunks)
        return _NonStreamResponse(
            choices=[_NonStreamChoice(message=_NonStreamMessage(content="T"))]
        )


class _FakeChat:
    def __init__(self, completions: _FakeCompletions) -> None:
        self.completions = completions


class _FakeClient:
    def __init__(self, chunks: list[_Chunk], **kw: Any) -> None:
        self.chat = _FakeChat(_FakeCompletions(chunks, **kw))


# --------------------------------------------------------------------------- #
# Bridge fixture — reloaded server with stubbed gateway-tool calls            #
# --------------------------------------------------------------------------- #


@pytest.fixture
def bridge(tmp_path, monkeypatch):
    monkeypatch.setenv("ARTIFACTS_DB_PATH", str(tmp_path / "art.db"))
    monkeypatch.setenv("OPENCLAW_GATEWAY_URL", "http://localhost:18789")
    monkeypatch.setenv("OPENCLAW_GATEWAY_TOKEN", "test-token")
    monkeypatch.setenv("ARTIFACT_EMISSION", "off")
    # Default both flags OFF so per-test toggle is explicit. The realtime
    # flag's production default is `poll` (Option 1 remediation 2026-05-13);
    # tests that want feed behavior must set it explicitly so we can keep
    # the M1 hard-preserve gates (T-7 etc.) testing the off path.
    monkeypatch.delenv("ACTIVITY_EVENTS", raising=False)
    monkeypatch.setenv("OPENCLAW_REALTIME_PROGRESS", "off")

    from app import artifact_store as art_mod
    importlib.reload(art_mod)
    from app import marker_extractor as mx_mod
    importlib.reload(mx_mod)
    from app import openclaw_chatkit_server as srv_mod
    importlib.reload(srv_mod)
    return srv_mod


def _user_message(text: str) -> UserMessageItem:
    return UserMessageItem(
        thread_id="thread-x",
        id="msg-user-1",
        created_at=datetime.now(),
        content=[UserMessageTextContent(text=text)],
        attachments=[],
        inference_options={},
    )


def _thread_meta() -> ThreadMetadata:
    return ThreadMetadata(
        id="thread-x",
        created_at=datetime.now(),
        status={"type": "active"},
        metadata={},
    )


async def _drive(server, chunks, *, raise_on_create: bool = False) -> list:
    server.client = _FakeClient(chunks, raise_on_create=raise_on_create)
    thread = _thread_meta()
    user_msg = _user_message("hello")
    await server.store.save_thread(thread, context={})
    await server.store.add_thread_item(thread.id, user_msg, context={})
    out: list = []
    async for ev in server.respond(thread, user_msg, context={}):
        out.append(ev)
    return out


def _kinds(events: list) -> list[str]:
    """Compact event-shape labels for ordering assertions."""
    out: list[str] = []
    for ev in events:
        name = type(ev).__name__
        update = getattr(ev, "update", None)
        item = getattr(ev, "item", None)
        if update is not None:
            out.append(f"{name}({type(update).__name__})")
        elif item is not None:
            out.append(f"{name}({type(item).__name__})")
        else:
            out.append(name)
    return out


# --------------------------------------------------------------------------- #
# T-3 — Workflow event sequence with flag ON                                  #
# --------------------------------------------------------------------------- #


@pytest.mark.asyncio
async def test_t3_workflow_event_sequence_with_tool_call_then_content(
    bridge, monkeypatch
) -> None:
    """T-3 (strict ordering — strengthened 2026-05-13 PM):
    With ACTIVITY_EVENTS=on, the event sequence MUST satisfy:
      - exactly ONE ThreadItemAddedEvent(WorkflowItem)
      - exactly ONE ThreadItemDoneEvent(WorkflowItem)
      - ≥1 task event in between
      - workflow added precedes all task events; all task events precede
        workflow done
      - **NEW:** WorkflowItem ADDED precedes AssistantMessageItem ADDED so
        the widget renders ABOVE the assistant bubble (receive order =
        render order in ChatKit React; per spec §4 U1-U3 and AC-3).
    """
    monkeypatch.setenv("ACTIVITY_EVENTS", "on")
    srv_mod = bridge
    server = srv_mod.server

    chunks = [
        _tool_chunk("bash"),
        _text_chunk("Listing files...\n"),
        _text_chunk("Done."),
    ]
    events = await _drive(server, chunks)

    workflow_added_idx = [
        i
        for i, e in enumerate(events)
        if isinstance(e, ThreadItemAddedEvent) and isinstance(e.item, WorkflowItem)
    ]
    workflow_done_idx = [
        i
        for i, e in enumerate(events)
        if isinstance(e, ThreadItemDoneEvent) and isinstance(e.item, WorkflowItem)
    ]
    assistant_added_idx = [
        i
        for i, e in enumerate(events)
        if isinstance(e, ThreadItemAddedEvent)
        and isinstance(e.item, AssistantMessageItem)
    ]
    task_evt_idx = [
        i
        for i, e in enumerate(events)
        if isinstance(e, ThreadItemUpdatedEvent)
        and isinstance(e.update, (WorkflowTaskAdded, WorkflowTaskUpdated))
    ]

    assert len(workflow_added_idx) == 1, f"expected 1 workflow added, kinds={_kinds(events)}"
    assert len(workflow_done_idx) == 1, f"expected 1 workflow done, kinds={_kinds(events)}"
    assert task_evt_idx, "expected at least one workflow task event"
    assert assistant_added_idx, "expected at least one assistant added"
    # Ordering invariants.
    assert workflow_added_idx[0] < min(task_evt_idx)
    assert max(task_evt_idx) < workflow_done_idx[0]
    # NEW strict ordering: workflow added precedes assistant added.
    # This is what makes the widget render ABOVE the assistant bubble on
    # first paint (SSE receive order).
    assert workflow_added_idx[0] < assistant_added_idx[0], (
        f"WorkflowItem ADDED must precede AssistantMessageItem ADDED for "
        f"the widget to render above the assistant bubble. "
        f"workflow_added@{workflow_added_idx[0]} vs "
        f"assistant_added@{assistant_added_idx[0]}. kinds={_kinds(events)}"
    )
    # Timestamp invariant: ChatKit React sorts thread items by `created_at`
    # on re-render (not just SSE arrival), so the workflow's timestamp
    # MUST be earlier than the assistant message's. Without this, the
    # widget renders above on first paint but slides below once the prose
    # streams in — the visible regression from 2026-05-13 PM.
    wf_added_evt = events[workflow_added_idx[0]]
    asst_added_evt = events[assistant_added_idx[0]]
    assert wf_added_evt.item.created_at < asst_added_evt.item.created_at, (
        "WorkflowItem.created_at MUST be earlier than "
        "AssistantMessageItem.created_at so the widget stays above the "
        "assistant bubble after re-render. "
        f"workflow={wf_added_evt.item.created_at} "
        f"assistant={asst_added_evt.item.created_at}"
    )

    # The opened WorkflowItem should expose at least 2 tasks (Thinking +
    # Calling bash), and the bash task should end in `complete`.
    wf_item = events[workflow_added_idx[0]].item
    titles = [t.title for t in wf_item.workflow.tasks]
    assert any("Thinking" in (t or "") for t in titles)
    assert any("bash" in (t or "").lower() for t in titles)
    # After end_workflow, every leftover loading flips to complete.
    final_item = events[workflow_done_idx[0]].item
    assert all(t.status_indicator == "complete" for t in final_item.workflow.tasks)


# --------------------------------------------------------------------------- #
# T-4 — DUPLICATION REGRESSION: subagent + 10 heartbeats + relay              #
#       (the explicit 2026-05-13 "aik post write krwao designing k lyia" fix) #
# --------------------------------------------------------------------------- #


@pytest.mark.asyncio
async def test_t4_subagent_heartbeats_collapse_to_single_task(
    bridge, monkeypatch
) -> None:
    """T-4 (non-negotiable): with ACTIVITY_EVENTS=on, a subagent spawn plus
    multiple heartbeat ticks plus a relay MUST produce exactly ONE
    delegation task on the workflow (status=complete after relay), NEVER
    one task per heartbeat tick. This is the explicit regression test for
    the 2026-05-13 duplication bug.
    """
    monkeypatch.setenv("ACTIVITY_EVENTS", "on")
    # Make the watch loop fire heartbeats fast and bail fast.
    import app.openclaw_chatkit_server as srv_mod
    monkeypatch.setattr(srv_mod, "_WATCH_POLL_INTERVAL_S", 0.01, raising=True)
    monkeypatch.setattr(srv_mod, "_WATCH_PILL_INTERVAL_S", 0.0, raising=True)
    monkeypatch.setattr(srv_mod, "_WATCH_INITIAL_QUIET_S", 0.05, raising=True)
    monkeypatch.setattr(srv_mod, "_WATCH_QUIET_AFTER_LAST_NEW_S", 0.05, raising=True)

    server = srv_mod.server

    # Fake gateway-tool plumbing.
    heartbeat_polls = {"n": 0}
    relayed_msg = {
        "role": "assistant",
        "content": "Here is the post copy for the designing service",
        "__openclaw": {"seq": 42},
    }

    async def fake_has_yield_signal(self, session_key: str) -> bool:
        return True

    async def fake_fetch(self, session_key: str, limit: int = 50):
        # First call = initial snapshot (no new messages).
        # After N polls, deliver the relay assistant message.
        heartbeat_polls["n"] += 1
        if heartbeat_polls["n"] <= 10:
            return []  # 10 heartbeat ticks see no new messages
        return [relayed_msg]

    active_calls = {"n": 0}

    async def fake_list_active(self, session_key: str) -> int:
        # Stay "active" while heartbeats run; report 0 after relay so the
        # watch loop can exit.
        active_calls["n"] += 1
        if heartbeat_polls["n"] <= 11:
            return 1
        return 0

    monkeypatch.setattr(
        srv_mod.OpenClawChatKitServer, "_has_yield_signal", fake_has_yield_signal
    )
    monkeypatch.setattr(
        srv_mod.OpenClawChatKitServer, "_fetch_session_messages", fake_fetch
    )
    monkeypatch.setattr(
        srv_mod.OpenClawChatKitServer, "_list_active_subagents", fake_list_active
    )

    # Hub's interim turn closes with stop. Use one prose chunk so the empty
    # check doesn't surface the "(no response)" fallback.
    chunks = [_text_chunk("Yielding to specialist…")]
    events = await _drive(server, chunks)

    # The workflow opens once.
    workflow_adds = [
        e for e in events
        if isinstance(e, ThreadItemAddedEvent) and isinstance(e.item, WorkflowItem)
    ]
    assert len(workflow_adds) == 1, f"expected 1 workflow added, got {len(workflow_adds)}"

    # Count NEW tasks added (WorkflowTaskAdded). Across the whole turn we
    # expect: 1 "Thinking" + 1 "Delegated to specialist" = 2 tasks added.
    # Critically, the 10 heartbeats must NOT have added new tasks.
    task_adds = [
        e for e in events
        if isinstance(e, ThreadItemUpdatedEvent)
        and isinstance(e.update, WorkflowTaskAdded)
    ]
    # First task came inside the WorkflowItem added event (Thinking); only
    # the delegation task should be WorkflowTaskAdded.
    assert len(task_adds) == 1, (
        f"DUPLICATION REGRESSION: expected exactly 1 WorkflowTaskAdded for "
        f"the delegation (heartbeats must mutate, not add). Got {len(task_adds)}."
    )
    # Note on the historical title — the WorkflowTaskAdded event's `task`
    # field is a reference to the same CustomTask the tracker continues to
    # mutate, so by the time this assertion runs it shows the LATEST title
    # ("Relay from specialist"). Over the wire, ChatKit serializes each
    # event at yield time, so the consumer sees "Delegated to specialist"
    # then later sees "Relay from specialist" via WorkflowTaskUpdated. We
    # verify the in-memory FINAL state instead (below).

    # Find the final workflow item to inspect.
    workflow_dones = [
        e for e in events
        if isinstance(e, ThreadItemDoneEvent) and isinstance(e.item, WorkflowItem)
    ]
    assert len(workflow_dones) == 1
    final_tasks = workflow_dones[0].item.workflow.tasks
    # Exactly two tasks total in the workflow: Thinking + Delegated. Never 11.
    assert len(final_tasks) == 2, (
        f"workflow has {len(final_tasks)} tasks; expected exactly 2 "
        f"(Thinking + Delegated). Titles: {[t.title for t in final_tasks]}"
    )
    # Delegation task ends complete.
    delegation = [t for t in final_tasks if "Delegated" in (t.title or "") or "Relay" in (t.title or "")]
    assert delegation, "no delegation/relay task on the final workflow"
    assert delegation[0].status_indicator == "complete"

    # The relayed assistant message did land on the stream.
    relay_items = [
        e for e in events
        if isinstance(e, ThreadItemAddedEvent)
        and isinstance(e.item, AssistantMessageItem)
        and (e.item.id or "").startswith("message_relay_")
    ]
    assert len(relay_items) == 1, "expected exactly one relay assistant item"


# --------------------------------------------------------------------------- #
# T-5 — Source-level cancellation guard                                       #
# --------------------------------------------------------------------------- #


def test_t5_cancellation_pattern_preserved() -> None:
    """T-5: respond() MUST keep the `asyncio.wait` + `pending_chunk` pattern
    from commit 1dde7f57. No `asyncio.wait_for` CALL around the chunk read.

    The existing comment in respond() still mentions `asyncio.wait_for` to
    document why the pattern exists — guard against the call, not the
    documentation. AST-level check rules out the false positive.
    """
    import textwrap
    from app.openclaw_chatkit_server import OpenClawChatKitServer
    src = textwrap.dedent(inspect.getsource(OpenClawChatKitServer.respond))
    tree = ast.parse(src)
    bad_calls: list[str] = []
    for node in ast.walk(tree):
        if isinstance(node, ast.Call):
            f = node.func
            if isinstance(f, ast.Attribute) and f.attr == "wait_for":
                # asyncio.wait_for OR loop.wait_for etc.
                bad_calls.append(ast.unparse(node)[:80])
    assert not bad_calls, (
        f"respond() contains forbidden wait_for call(s): {bad_calls}. "
        "Use asyncio.wait + pending_chunk (see commit 1dde7f57)."
    )
    assert "asyncio.wait" in src, "respond() missing asyncio.wait gating"
    assert "pending_chunk" in src, "respond() missing pending_chunk pattern"


# --------------------------------------------------------------------------- #
# T-6 — Source-level watch-loop termination guard                             #
# --------------------------------------------------------------------------- #


def test_t6_watch_loop_termination_policy_preserved() -> None:
    """T-6: _watch_for_handoff_relay MUST keep `should_try_exit` AND
    `_list_active_subagents` in the termination block (commit a8f8205f)."""
    from app.openclaw_chatkit_server import OpenClawChatKitServer
    src = inspect.getsource(OpenClawChatKitServer._watch_for_handoff_relay)
    assert "should_try_exit" in src
    assert "_list_active_subagents" in src


# --------------------------------------------------------------------------- #
# T-7 — Flag OFF: Phase-2 byte-identical ProgressUpdateEvent path             #
# --------------------------------------------------------------------------- #


@pytest.mark.asyncio
async def test_t7_flag_off_matches_phase2_pill_sequence(bridge, monkeypatch) -> None:
    """T-7: With ACTIVITY_EVENTS unset/off, the bridge emits the original
    `ProgressUpdateEvent` pills and zero workflow events."""
    monkeypatch.delenv("ACTIVITY_EVENTS", raising=False)
    server = bridge.server

    chunks = [
        _tool_chunk("bash"),
        _text_chunk("Listing files...\n"),
        _text_chunk("Done."),
    ]
    events = await _drive(server, chunks)

    # Zero workflow events.
    workflow_evts = [
        e for e in events
        if (isinstance(e, ThreadItemAddedEvent) and isinstance(e.item, WorkflowItem))
        or (isinstance(e, ThreadItemDoneEvent) and isinstance(e.item, WorkflowItem))
        or (
            isinstance(e, ThreadItemUpdatedEvent)
            and isinstance(e.update, (WorkflowTaskAdded, WorkflowTaskUpdated))
        )
    ]
    assert workflow_evts == [], (
        f"Flag-off path emitted workflow events: {_kinds(workflow_evts)}"
    )

    # Phase-2 pill text expected: at least "Thinking…" and "Calling bash…".
    pill_texts = [
        e.text for e in events if isinstance(e, ProgressUpdateEvent)
    ]
    assert any("Thinking" in (t or "") for t in pill_texts), pill_texts
    assert any("Calling bash" in (t or "") for t in pill_texts), pill_texts


# --------------------------------------------------------------------------- #
# T-8 — Flag ON: no non-empty ProgressUpdateEvent for thinking/tool/subagent  #
# --------------------------------------------------------------------------- #


@pytest.mark.asyncio
async def test_t8_flag_on_emits_no_progress_pills(bridge, monkeypatch) -> None:
    """T-8: With ACTIVITY_EVENTS=on, no ProgressUpdateEvent carries text for
    thinking / tool calls / subagent activity. The empty-text clear-pill
    end-of-turn yields are exempt (terminator markers)."""
    monkeypatch.setenv("ACTIVITY_EVENTS", "on")
    server = bridge.server

    chunks = [
        _tool_chunk("bash"),
        _text_chunk("Listing files...\n"),
        _text_chunk("Done."),
    ]
    events = await _drive(server, chunks)

    non_empty_pills = [
        e for e in events if isinstance(e, ProgressUpdateEvent) and (e.text or "")
    ]
    assert non_empty_pills == [], (
        f"Flag-on path emitted progress pills: {[e.text for e in non_empty_pills]}"
    )


# --------------------------------------------------------------------------- #
# T-14 — finally guarantees end_workflow even on exception                    #
# --------------------------------------------------------------------------- #


@pytest.mark.asyncio
async def test_t14_exception_still_finalizes_workflow(bridge, monkeypatch) -> None:
    """T-14: When respond() raises mid-turn, the workflow MUST still emit
    ThreadItemDoneEvent(WorkflowItem) in the finally path so the UI does
    not show a perpetually-spinning widget."""
    monkeypatch.setenv("ACTIVITY_EVENTS", "on")
    server = bridge.server
    # The first chunk creates the stream — raise INSIDE the stream so that
    # at least the pre-flight `Thinking` task has been added before the
    # exception. (raise_on_create raises BEFORE the iterator starts → no
    # tracker open, finally is a no-op which would be correct but not
    # exercise the gate.)
    class _RaisingStream:
        def __aiter__(self): return self
        async def __anext__(self): raise RuntimeError("gateway boom mid-stream")

    class _RaisingCompletions:
        async def create(self, **kw):
            if kw.get("stream"):
                return _RaisingStream()
            return _NonStreamResponse(choices=[_NonStreamChoice(message=_NonStreamMessage(content="T"))])

    class _RaisingChat:
        completions = _RaisingCompletions()

    class _RaisingClient:
        chat = _RaisingChat()

    server.client = _RaisingClient()
    thread = _thread_meta()
    user_msg = _user_message("trigger")
    await server.store.save_thread(thread, context={})
    await server.store.add_thread_item(thread.id, user_msg, context={})

    events: list = []
    async for ev in server.respond(thread, user_msg, context={}):
        events.append(ev)

    # Workflow added (pre-flight Thinking) AND workflow done MUST both be
    # present despite the gateway exception.
    assert any(
        isinstance(e, ThreadItemAddedEvent) and isinstance(e.item, WorkflowItem)
        for e in events
    ), "workflow added missing on exception path"
    assert any(
        isinstance(e, ThreadItemDoneEvent) and isinstance(e.item, WorkflowItem)
        for e in events
    ), "workflow done missing on exception path (T-14 regression)"


# --------------------------------------------------------------------------- #
# T-16 — Cancellation pattern is the source-level guard from T-5 (re-asserted) #
# --------------------------------------------------------------------------- #


def test_t16_no_wait_for_call_anywhere_in_bridge() -> None:
    """T-16: AST-level check — no `wait_for(...)` call appears anywhere in
    `openclaw_chatkit_server.py`. (The string `wait_for` may appear in
    comments documenting WHY the pattern is forbidden — that is fine; the
    regression risk is the call, not the docstring.)"""
    path = Path(__file__).resolve().parent.parent / "app" / "openclaw_chatkit_server.py"
    tree = ast.parse(path.read_text())
    bad_calls: list[str] = []
    for node in ast.walk(tree):
        if isinstance(node, ast.Call):
            f = node.func
            if isinstance(f, ast.Attribute) and f.attr == "wait_for":
                bad_calls.append(ast.unparse(node)[:80])
            if isinstance(f, ast.Name) and f.id == "wait_for":
                bad_calls.append(ast.unparse(node)[:80])
    assert not bad_calls, (
        f"openclaw_chatkit_server.py contains forbidden wait_for call(s): "
        f"{bad_calls}. Regression of commit 1dde7f57."
    )


# --------------------------------------------------------------------------- #
# Phase 2.4 — realtime feed integration                                       #
# --------------------------------------------------------------------------- #


@pytest.mark.asyncio
async def test_realtime_ws_fails_open_when_gateway_unreachable(
    bridge, monkeypatch
) -> None:
    """Phase 2.4: when OPENCLAW_REALTIME_PROGRESS=ws but the gateway WS is
    unreachable (no socket listening), respond() MUST still complete
    successfully. The feed's start() failure is caught; feed becomes None;
    the existing tracker-and-pill emit sites remain active.

    Also asserts the new env flag is recognised by the bridge module."""
    monkeypatch.setenv("OPENCLAW_REALTIME_PROGRESS", "ws")
    # Force a guaranteed-unreachable port for the gateway WS connect.
    monkeypatch.setenv("OPENCLAW_GATEWAY_URL", "http://127.0.0.1:1")
    # Reset the WS-client singleton so the new env is picked up.
    from app.gateway_ws_client import GatewayWsClient

    GatewayWsClient.reset_process_instance()
    server = bridge.server

    chunks = [_text_chunk("Hello "), _text_chunk("Raza.")]
    events = await _drive(server, chunks)

    # Tracker should still open (feed=ws implies tracker enabled per the
    # bridge's _tracker_enabled rule). Workflow open + done are present.
    assert any(
        isinstance(e, ThreadItemAddedEvent) and isinstance(e.item, WorkflowItem)
        for e in events
    )
    assert any(
        isinstance(e, ThreadItemDoneEvent) and isinstance(e.item, WorkflowItem)
        for e in events
    )
    # Assistant message ADDED + final DONE present too — the gateway-failure
    # safety net must not break the basic prose stream.
    assert any(
        isinstance(e, ThreadItemAddedEvent) and isinstance(e.item, AssistantMessageItem)
        for e in events
    )


def test_realtime_progress_mode_flag_parsing() -> None:
    """Source-level guard: the bridge reads OPENCLAW_REALTIME_PROGRESS via a
    helper that returns only canonical values. Default is `off` (reverted
    2026-05-13 EVENING after spec §13c finding). Typo-allowed values fall
    back to the safe default."""
    from app import openclaw_chatkit_server as srv_mod
    import os as _os

    prev = _os.environ.get("OPENCLAW_REALTIME_PROGRESS")
    try:
        for v, expected in [
            ("", "off"),        # default
            ("off", "off"),
            ("ws", "ws"),
            ("WS", "ws"),
            ("poll", "poll"),
            ("yes", "off"),     # not a valid value -> safe default
            ("true", "off"),
            ("1", "off"),
            (" ws ", "ws"),     # trimmed
        ]:
            _os.environ["OPENCLAW_REALTIME_PROGRESS"] = v
            assert srv_mod._realtime_progress_mode() == expected, (v, expected)
    finally:
        if prev is None:
            _os.environ.pop("OPENCLAW_REALTIME_PROGRESS", None)
        else:
            _os.environ["OPENCLAW_REALTIME_PROGRESS"] = prev