diff --git a/api/streaming.py b/api/streaming.py index 775cf784..35b04b0f 100644 --- a/api/streaming.py +++ b/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 = { diff --git a/tests/test_issue1298_cancel_and_activity.py b/tests/test_issue1298_cancel_and_activity.py index fe4b1b96..fe6e363d 100644 --- a/tests/test_issue1298_cancel_and_activity.py +++ b/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 ─────────