hermes-webui/tests/test_issue1298_cancel_and_activity.py

"""Regression tests for #1298 — Activity panel UI state and Stop/Cancel data loss.

Two distinct bugs reported in YanTianlong-01's bug report on v0.50.240:

  1. The expanded Activity list collapses automatically when new activity arrives.
  2. The latest user message disappears after clicking Stop/Cancel during streaming.

Bug 2 is server-side data loss (the message is gone from session JSON, not just
the in-memory client copy) caused by cancel_stream() clearing pending_user_message
without first persisting it to s.messages. This test suite locks down both fixes.
"""
import pathlib
import queue
import re
import threading
from unittest.mock import Mock

import pytest

import api.config as config
import api.models as models
import api.streaming as streaming
from api.models import Session
from api.streaming import cancel_stream

REPO_ROOT = pathlib.Path(__file__).parent.parent.resolve()


# ── Fixtures ────────────────────────────────────────────────────────────────

@pytest.fixture(autouse=True)
def _isolate_session_dir(tmp_path, monkeypatch):
    """Redirect SESSION_DIR / SESSION_INDEX_FILE to an isolated temp dir."""
    session_dir = tmp_path / "sessions"
    session_dir.mkdir()
    index_file = session_dir / "_index.json"
    monkeypatch.setattr(models, "SESSION_DIR", session_dir)
    monkeypatch.setattr(models, "SESSION_INDEX_FILE", index_file)
    models.SESSIONS.clear()
    yield
    models.SESSIONS.clear()


@pytest.fixture(autouse=True)
def _isolate_stream_state():
    config.STREAMS.clear()
    config.CANCEL_FLAGS.clear()
    config.AGENT_INSTANCES.clear()
    config.STREAM_PARTIAL_TEXT.clear()
    yield
    config.STREAMS.clear()
    config.CANCEL_FLAGS.clear()
    config.AGENT_INSTANCES.clear()
    config.STREAM_PARTIAL_TEXT.clear()


@pytest.fixture(autouse=True)
def _isolate_agent_locks():
    config.SESSION_AGENT_LOCKS.clear()
    yield
    config.SESSION_AGENT_LOCKS.clear()


def _make_pending_session(session_id="cancel_sid_1298",
                          pending_msg="Help me debug this issue",
                          messages=None,
                          attachments=None):
    """Build a session in mid-stream state: pending_user_message set, messages may be empty."""
    s = Session(
        session_id=session_id,
        title="Test Session",
        messages=messages or [],
    )
    s.pending_user_message = pending_msg
    s.pending_attachments = list(attachments or [])
    s.pending_started_at = None
    s.active_stream_id = "stream_1298"
    s.save()
    models.SESSIONS[session_id] = s
    return s


def _setup_cancel_stream_state(session_id, stream_id="stream_1298"):
    """Wire up STREAMS/CANCEL_FLAGS/AGENT_INSTANCES so cancel_stream() can run."""
    config.STREAMS[stream_id] = queue.Queue()
    config.CANCEL_FLAGS[stream_id] = threading.Event()
    mock_agent = Mock()
    mock_agent.session_id = session_id
    mock_agent.interrupt = Mock()
    config.AGENT_INSTANCES[stream_id] = mock_agent
    return stream_id, mock_agent


# ── Server-side: cancel preserves pending_user_message in s.messages ────────

class TestIssue1298CancelPreservesUserMessage:
    """Issue 2: Latest user message disappears after Stop/Cancel during streaming.

    Root cause: cancel_stream() at api/streaming.py:2575+ clears
    s.pending_user_message before the streaming thread's
    _merge_display_messages_after_agent_result() has a chance to merge the
    user turn into s.messages. The session is saved with neither
    pending_user_message nor a corresponding s.messages entry, so the user's
    typed text is lost permanently.

    Fix: synthesize a user turn from pending_user_message into s.messages when
    the most recent message isn't already that turn.
    """

    def test_cancel_synthesizes_user_message_when_messages_empty(self):
        """When the agent thread is killed before it can append the user turn,
        cancel_stream() must persist pending_user_message into s.messages so
        the typed text survives a session reload."""
        s = _make_pending_session(
            session_id="cancel_sid_empty",
            pending_msg="What's the weather forecast?",
            messages=[],
        )
        stream_id, _agent = _setup_cancel_stream_state(s.session_id)

        result = cancel_stream(stream_id)
        assert result is True

        # Reload from disk to confirm save happened
        s2 = models.SESSIONS[s.session_id]
        roles = [m.get("role") for m in s2.messages if isinstance(m, dict)]
        contents = [m.get("content") for m in s2.messages if isinstance(m, dict)]

        assert "user" in roles, (
            "Expected user turn synthesized into s.messages — "
            f"got roles={roles}"
        )
        assert "What's the weather forecast?" in contents, (
            "Expected pending_user_message text preserved verbatim in s.messages — "
            f"got contents={contents}"
        )
        assert s2.pending_user_message is None, (
            "pending_user_message must be cleared after cancel"
        )
        assert s2.active_stream_id is None

    def test_cancel_does_not_double_append_when_streaming_thread_already_merged(self):
        """If the streaming thread won the race and already merged the user turn
        into s.messages before cancel_stream() got the lock, cancel must not
        append a duplicate."""
        prior_user = {"role": "user", "content": "Run a tool for me"}
        s = _make_pending_session(
            session_id="cancel_sid_already_merged",
            pending_msg="Run a tool for me",
            messages=[prior_user],
        )
        stream_id, _agent = _setup_cancel_stream_state(s.session_id)

        cancel_stream(stream_id)

        s2 = models.SESSIONS[s.session_id]
        user_messages = [m for m in s2.messages
                         if isinstance(m, dict) and m.get("role") == "user"]
        # Exactly one user turn — no duplicate
        matching = [m for m in user_messages
                    if "Run a tool for me" in str(m.get("content") or "")]
        assert len(matching) == 1, (
            "Expected exactly one user turn matching pending_user_message — "
            f"got {len(matching)} ({user_messages})"
        )

    def test_cancel_synthesized_user_message_carries_attachments(self):
        """A cancelled turn that had attachments uploaded should keep them on
        the recovered user message."""
        s = _make_pending_session(
            session_id="cancel_sid_attachments",
            pending_msg="Look at this screenshot",
            messages=[],
            attachments=["bug_screenshot.png", "stack_trace.txt"],
        )
        stream_id, _agent = _setup_cancel_stream_state(s.session_id)

        cancel_stream(stream_id)

        s2 = models.SESSIONS[s.session_id]
        user_msgs = [m for m in s2.messages
                     if isinstance(m, dict) and m.get("role") == "user"]
        assert user_msgs, "User turn must be persisted on cancel"
        recovered = user_msgs[0]
        assert recovered.get("attachments") == [
            "bug_screenshot.png", "stack_trace.txt"
        ], (
            "Attachment list must be preserved on the synthesized user turn — "
            f"got {recovered.get('attachments')}"
        )

    def test_cancel_no_pending_user_message_does_nothing_extra(self):
        """When there is no pending_user_message (e.g. cancel after the agent
        has already returned), cancel_stream() must not synthesize a phantom
        user turn."""
        s = Session(
            session_id="cancel_sid_no_pending",
            title="Test",
            messages=[{"role": "user", "content": "earlier turn"}],
        )
        s.active_stream_id = "stream_1298"
        s.pending_user_message = None
        s.save()
        models.SESSIONS[s.session_id] = s
        stream_id, _agent = _setup_cancel_stream_state(s.session_id)

        cancel_stream(stream_id)

        s2 = models.SESSIONS[s.session_id]
        user_messages = [m for m in s2.messages
                         if isinstance(m, dict) and m.get("role") == "user"]
        # Still exactly one — the original earlier turn
        assert len(user_messages) == 1
        assert user_messages[0].get("content") == "earlier turn"

    def test_cancel_synthesizes_when_prior_turn_content_is_substring_of_pending(self):
        """Regression for Opus pre-release review of v0.50.246 (PR #1338):

        The substring guard in cancel_stream() was symmetric — it would skip
        synthesis if the prior user turn's content was a substring of the new
        pending message. Common confirmation replies ("ok", "yes", "go") would
        match longer follow-up prompts ("ok please continue") and the
        synthesis would be skipped, re-introducing the data-loss bug.

        The fix: gate the substring check on a timestamp comparison —
        only treat the latest user turn as "already merged by the streaming
        thread" if its timestamp is at or after pending_started_at. Earlier
        turns whose content happens to be a substring must not short-circuit
        the synthesis path.
        """
        import time as _time
        # Prior reply was "ok" (a common short reply).
        prior_ts = int(_time.time()) - 60  # 1 minute ago
        prior_user = {
            "role": "user",
            "content": "ok",
            "timestamp": prior_ts,
        }
        s = _make_pending_session(
            session_id="cancel_sid_substring_collision",
            pending_msg="ok please continue with the analysis",
            messages=[prior_user],
        )
        # The pending turn started AFTER the prior turn was logged.
        s.pending_started_at = prior_ts + 10
        s.save()
        models.SESSIONS[s.session_id] = s

        stream_id, _agent = _setup_cancel_stream_state(s.session_id)
        cancel_stream(stream_id)

        s2 = models.SESSIONS[s.session_id]
        user_messages = [m for m in s2.messages
                         if isinstance(m, dict) and m.get("role") == "user"]
        contents = [m.get("content") for m in user_messages]

        assert "ok please continue with the analysis" in contents, (
            "Pending user message must be synthesized — the substring 'ok' from a prior turn "
            "must NOT cause the synthesis to be skipped. "
            f"Got contents={contents}"
        )
        assert len(user_messages) == 2, (
            "Expected both the original prior turn AND the synthesized new turn — "
            f"got {len(user_messages)} user messages"
        )


# ── Client-side: ui.js source-level guards for activity-group state ─────────

class TestIssue1298ActivityGroupExpandPersistence:
    """Issue 1: Expanded Activity list collapses automatically when new
    activity arrives.

    Root cause:
      - ensureActivityGroup() (static/ui.js) creates the live activity group
        with `tool-call-group-collapsed` whenever it's missing
      - finalizeThinkingCard() force-adds `tool-call-group-collapsed` on every
        tool boundary, regardless of user intent
      - The user's manually-set expand state lives only on a DOM class list,
        so any destroy/recreate cycle (which fires on every thinking → tool →
        thinking transition) wipes it.

    Fix: track the user's last explicit toggle in a per-turn singleton, and
    skip the force-collapse when the user has explicitly expanded.
    """

    def test_ui_js_tracks_user_expand_intent_for_live_activity_group(self):
        src = (REPO_ROOT / "static" / "ui.js").read_text()
        assert "_liveActivityUserExpanded" in src, (
            "ui.js must declare a per-turn tracker for the user's expand intent "
            "on the live activity group (#1298)"
        )
        assert "_onLiveActivityToggle" in src, (
            "ui.js must expose a helper that records the user's manual toggle "
            "of the live activity group"
        )

    def test_ensure_activity_group_restores_expand_intent(self):
        """ensureActivityGroup() must consult _liveActivityUserExpanded when
        creating a fresh live group so the user's prior expand survives the
        destroy/recreate cycle."""
        src = (REPO_ROOT / "static" / "ui.js").read_text()
        # Find the ensureActivityGroup function body
        m = re.search(
            r"function ensureActivityGroup\(inner, opts\)\{(.*?)\n\}",
            src, re.DOTALL,
        )
        assert m, "ensureActivityGroup() must exist in ui.js"
        body = m.group(1)
        assert "_liveActivityUserExpanded" in body, (
            "ensureActivityGroup() body must reference the user-expand tracker "
            "to restore intent on re-create (#1298)"
        )
        assert "live" in body and "_liveActivityUserExpanded === true" in body, (
            "ensureActivityGroup() must override the default `collapsed` flag "
            "when the user previously expanded the live group"
        )

    def test_finalize_thinking_card_respects_user_expand(self):
        """finalizeThinkingCard() must not force-collapse the live Worklog."""
        src = (REPO_ROOT / "static" / "ui.js").read_text()
        m = re.search(
            r"function finalizeThinkingCard\(\)\{(.*?)\n\}",
            src, re.DOTALL,
        )
        assert m, "finalizeThinkingCard() must exist in ui.js"
        body = m.group(1)
        assert "tool-call-group-collapsed" not in body, (
            "Live Worklog must remain expanded until the settled render replaces "
            "it with the final collapsed L1 Activity summary."
        )
        assert "aria-expanded','false'" not in body

    def test_inline_onclick_records_user_intent(self):
        """The summary button's click path must call _onLiveActivityToggle
        so user clicks update the tracker (#1298)."""
        src = (REPO_ROOT / "static" / "ui.js").read_text()
        # The summary button is built inline inside ensureActivityGroup.
        assert "_onLiveActivityToggle" in src, (
            "_onLiveActivityToggle helper must be defined"
        )
        assert "function _toggleActivityGroup" in src, (
            "Activity summary clicks should route through the shared toggle helper"
        )
        # The inline onclick may delegate to _toggleActivityGroup(); that helper
        # must still call _onLiveActivityToggle(group) so user toggles are
        # captured into _liveActivityUserExpanded.
        m = re.search(r'class="tool-call-group-summary"[^`]*`', src)
        assert m, "live activity summary button template must be present"
        assert "onclick=\"_toggleActivityGroup(this)\"" in m.group(0), (
            "ensureActivityGroup() summary button should use the shared toggle helper"
        )
        toggle_body = re.search(
            r"function _toggleActivityGroup\(summary\)\{(.*?)\n\}",
            src,
            re.DOTALL,
        )
        assert toggle_body and "_onLiveActivityToggle(group)" in toggle_body.group(1), (
            "_toggleActivityGroup() must invoke _onLiveActivityToggle(group) "
            "so user clicks update the tracker"
        )

    def test_clear_live_tool_cards_resets_expand_intent(self):
        """clearLiveToolCards() — invoked between turns — must reset the
        per-turn user-expand tracker so the next turn starts collapsed by
        default (#1298)."""
        src = (REPO_ROOT / "static" / "ui.js").read_text()
        m = re.search(
            r"function clearLiveToolCards\(\)\{(.*?)\n\}",
            src, re.DOTALL,
        )
        assert m, "clearLiveToolCards() must exist"
        body = m.group(1)
        assert "_clearLiveActivityUserIntent" in body, (
            "clearLiveToolCards() must reset _liveActivityUserExpanded between "
            "turns so prior expand intent doesn't bleed into the next turn"
        )