fix(cancel): gate substring guard on pending_started_at timestamp (Opus review)

Pre-release Opus review on v0.50.246 caught a SHOULD-FIX in PR #1338's
cancel_stream synthesis: the symmetric substring guard
(_pending_user in _last_content OR _last_content in _pending_user) was too
loose. Common confirmation replies ("ok", "yes", "go") in the prior turn
would match longer follow-up prompts ("ok please continue"), the synthesis
would be skipped, and the user's typed text would be lost — exactly the
data-loss bug #1298 was supposed to fix.

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 of the pending must not short-circuit synthesis.

Also drops the symmetric (_last_content in _pending_user) branch — that
direction was the false-positive vector. Keeps the equality and prefix
match (workspace-prefix tolerance from the streaming thread).

Adds tests/test_issue1298_cancel_and_activity.py::
test_cancel_synthesizes_when_prior_turn_content_is_substring_of_pending —
regression for the exact 'ok' → 'ok please continue' scenario.

api/streaming.py

@@ -2607,6 +2607,7 @@ def cancel_stream(stream_id: str) -> bool:
                     _pending_user = getattr(_cs, 'pending_user_message', None)
                     _pending_atts_raw = getattr(_cs, 'pending_attachments', None)
                     _pending_atts = list(_pending_atts_raw) if isinstance(_pending_atts_raw, (list, tuple)) else []
+                    _pending_started = getattr(_cs, 'pending_started_at', None) or 0
                     _msgs_for_recovery = _cs.messages if isinstance(_cs.messages, list) else None
                     if _pending_user and _msgs_for_recovery is not None:
                         _last_user = None
@@ -2617,9 +2618,16 @@ def cancel_stream(stream_id: str) -> bool:
                         _already_persisted = False
                         if _last_user is not None:
                             _last_content = _last_user.get('content')
-                            if isinstance(_last_content, str):
+                            _last_ts = _last_user.get('timestamp') or 0
+                            # Only treat as already-persisted if the latest user turn
+                            # was created AT OR AFTER the current turn's pending_started_at.
+                            # An earlier turn whose content happens to be a substring
+                            # (e.g. prior reply was "ok", user now types "ok please continue")
+                            # must NOT short-circuit synthesis — that would re-introduce
+                            # the data-loss bug this guard is supposed to prevent.
+                            if isinstance(_last_content, str) and _last_ts >= _pending_started:
                                 # Tolerate the workspace prefix the streaming thread prepends.
-                                if _pending_user in _last_content or _last_content in _pending_user:
+                                if _pending_user == _last_content or _pending_user in _last_content:
                                     _already_persisted = True
                         if not _already_persisted:
                             _user_turn: dict = {

tests/test_issue1298_cancel_and_activity.py

@@ -213,6 +213,57 @@ class TestIssue1298CancelPreservesUserMessage:
         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 ─────────