fix: dedupe tool-only partial recovery markers

CHANGELOG.md

@@ -2,6 +2,9 @@
 
 ## [Unreleased]
 
+### Fixed
+
+- **PR #2593** by @Michaelyklam (closes #2592) — Deduplicate cancelled/recovered partial assistant markers using the full `(content, reasoning, partial tool calls)` payload instead of only non-empty text content. Tool-only failed turns no longer append identical empty-content `_partial` messages repeatedly, and full session loads collapse adjacent duplicate partial markers from already-bloated session files while preserving a backup.
 
 ## [v0.51.92] — 2026-05-19 — Release BP (stage-385 — 7-PR full sweep batch — RFC Slice 3c clarification + workspace tree icon alignment + project move cache refresh + auto-compression handoff metadata + Grok OAuth provider catalog + anonymous custom endpoint picker fallback + PWA standalone reload + pull-to-refresh)
 

api/models.py

@@ -562,7 +562,18 @@ class Session:
         p = SESSION_DIR / f'{sid}.json'
         if not p.exists():
             return None
-        return cls(**json.loads(p.read_text(encoding='utf-8')))
+        data = json.loads(p.read_text(encoding='utf-8'))
+        data['messages'], _collapsed_partials = _collapse_adjacent_duplicate_partials(data.get('messages'))
+        session = cls(**data)
+        if _collapsed_partials:
+            try:
+                # Self-heal bloated sessions on first full load without touching
+                # recency/index ordering; save() creates a .bak because this
+                # intentionally shrinks the transcript (#2592).
+                session.save(touch_updated_at=False, skip_index=True)
+            except Exception:
+                logger.debug("Failed to persist collapsed duplicate partials for %s", sid, exc_info=True)
+        return session
 
     @classmethod
     def load_metadata_only(cls, sid):
@@ -722,6 +733,57 @@ def _normalize_journal_recovery_text(value) -> str:
     return " ".join(str(value or "").split())
 
 
+def _partial_message_signature(message: dict) -> tuple:
+    """Return a stable identity for partial assistant markers recovered on load."""
+    if not isinstance(message, dict):
+        return ('', '', ())
+    tool_sig = []
+    for tool_call in message.get('_partial_tool_calls') or []:
+        if not isinstance(tool_call, dict):
+            continue
+        try:
+            args_sig = json.dumps(
+                tool_call.get('args') or {},
+                ensure_ascii=False,
+                sort_keys=True,
+                default=str,
+            )
+        except Exception:
+            args_sig = str(tool_call.get('args') or '')
+        tool_sig.append((
+            str(tool_call.get('name') or ''),
+            args_sig,
+            bool(tool_call.get('done', False)),
+            bool(tool_call.get('is_error', False)),
+            str(tool_call.get('preview') or tool_call.get('snippet') or ''),
+        ))
+    return (
+        str(message.get('content') or '').strip(),
+        str(message.get('reasoning') or '').strip(),
+        tuple(tool_sig),
+    )
+
+
+def _collapse_adjacent_duplicate_partials(messages) -> tuple[list, bool]:
+    """Collapse repeated identical partial markers from the same failed turn."""
+    if not isinstance(messages, list):
+        return messages, False
+    collapsed = []
+    changed = False
+    previous_partial_sig = None
+    for message in messages:
+        if isinstance(message, dict) and message.get('_partial'):
+            sig = _partial_message_signature(message)
+            if previous_partial_sig == sig:
+                changed = True
+                continue
+            previous_partial_sig = sig
+        else:
+            previous_partial_sig = None
+        collapsed.append(message)
+    return collapsed, changed
+
+
 def _find_existing_assistant_for_journal_content(session, content: str) -> int | None:
     candidate = _normalize_journal_recovery_text(content)
     if not candidate:

api/streaming.py

@@ -2590,6 +2590,56 @@ def _extract_tool_calls_from_messages(messages, live_tool_calls=None):
     return tool_calls
 
 
+def _partial_message_signature(message: dict) -> tuple:
+    """Return a stable identity for a persisted partial assistant marker."""
+    if not isinstance(message, dict):
+        return ('', '', ())
+    tool_sig = []
+    for tool_call in message.get('_partial_tool_calls') or []:
+        if not isinstance(tool_call, dict):
+            continue
+        try:
+            args_sig = json.dumps(
+                tool_call.get('args') or {},
+                ensure_ascii=False,
+                sort_keys=True,
+                default=str,
+            )
+        except Exception:
+            args_sig = str(tool_call.get('args') or '')
+        tool_sig.append((
+            str(tool_call.get('name') or ''),
+            args_sig,
+            bool(tool_call.get('done', False)),
+            bool(tool_call.get('is_error', False)),
+            str(tool_call.get('preview') or tool_call.get('snippet') or ''),
+        ))
+    return (
+        str(message.get('content') or '').strip(),
+        str(message.get('reasoning') or '').strip(),
+        tuple(tool_sig),
+    )
+
+
+def _partial_marker_already_present(messages, candidate: dict, *, before_idx: int | None = None) -> bool:
+    """Check for an equivalent partial marker in the current user turn only."""
+    if not isinstance(messages, list) or not isinstance(candidate, dict):
+        return False
+    end = before_idx if isinstance(before_idx, int) else len(messages)
+    end = max(0, min(end, len(messages)))
+    start = 0
+    for idx in range(end - 1, -1, -1):
+        msg = messages[idx]
+        if isinstance(msg, dict) and msg.get('role') == 'user':
+            start = idx + 1
+            break
+    candidate_sig = _partial_message_signature(candidate)
+    for msg in messages[start:end]:
+        if isinstance(msg, dict) and msg.get('_partial') and _partial_message_signature(msg) == candidate_sig:
+            return True
+    return False
+
+
 def _sse(handler, event, data):
     """Write one SSE event to the response stream."""
     payload = f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n"
@@ -5504,24 +5554,7 @@ def cancel_stream(stream_id: str) -> bool:
                         if any(pattern in _content for pattern in _CANCEL_MARKER_PATTERNS):
                             _cancel_marker_idx = _idx
                             break
-                _partial_already_present = False
-                if _stripped:
-                    for _m in _cs.messages:
-                        # Stage-350 Opus SHOULD-FIX (#2151): only dedup
-                        # against actual prior _partial markers from the
-                        # same stream, with exact content match. The original
-                        # substring check (`_stripped in _existing or
-                        # _existing in _stripped`) was too broad — any short
-                        # prior assistant reply (e.g. "OK", "Here is the
-                        # answer:") becomes a substring of many later partial
-                        # bodies and could silently drop the new partial,
-                        # resurrecting the #893 data-loss bug on long sessions.
-                        if not isinstance(_m, dict) or not _m.get('_partial'):
-                            continue
-                        if str(_m.get('content') or '').strip() == _stripped:
-                            _partial_already_present = True
-                            break
-                if (_stripped or _has_reasoning or _has_tools) and not _partial_already_present:
+                if _stripped or _has_reasoning or _has_tools:
                     _partial_msg: dict = {
                         'role': 'assistant',
                         'content': _stripped,  # may be empty for reasoning/tool-only turns
@@ -5548,7 +5581,16 @@ def cancel_stream(stream_id: str) -> bool:
                         # alongside the regular tool_calls path.
                         # (Opus pre-release review pass 2 of v0.50.251.)
                         _partial_msg['_partial_tool_calls'] = list(_cancel_tool_calls)
-                    _cs.messages.insert(_cancel_marker_idx, _partial_msg)
+                    # Deduplicate against the full partial payload, not just
+                    # non-empty content. Tool-only/reasoning-only partials have
+                    # empty content, so a content-gated check can append the same
+                    # failed turn repeatedly during cancel/replay recovery (#2592).
+                    if not _partial_marker_already_present(
+                        _cs.messages,
+                        _partial_msg,
+                        before_idx=_cancel_marker_idx,
+                    ):
+                        _cs.messages.insert(_cancel_marker_idx, _partial_msg)
                 # Cancel marker — flagged _error=True so it is stripped from conversation
                 # history on the next turn (prevents model from seeing "Task cancelled."
                 # as a prior assistant reply).

tests/test_issue2592_partial_dedupe.py

@@ -0,0 +1,87 @@
+import json
+
+
+def _tool_partial(reasoning="same reasoning", args=None, *, timestamp=123):
+    return {
+        "role": "assistant",
+        "content": "",
+        "_partial": True,
+        "timestamp": timestamp,
+        "reasoning": reasoning,
+        "_partial_tool_calls": [
+            {
+                "name": "execute_code",
+                "args": args or {"code": "raise RuntimeError('boom')"},
+                "done": True,
+                "is_error": True,
+                "duration": 3.87,
+            }
+        ],
+    }
+
+
+def test_tool_only_partial_dedupe_uses_reasoning_and_tool_signature():
+    from api.streaming import _partial_marker_already_present
+
+    existing = [
+        {"role": "user", "content": "run this"},
+        _tool_partial(),
+        {"role": "assistant", "content": "**Task cancelled.**", "_error": True},
+    ]
+
+    assert _partial_marker_already_present(existing, _tool_partial(), before_idx=2)
+    assert not _partial_marker_already_present(
+        existing,
+        _tool_partial(args={"code": "print('different tool body')"}),
+        before_idx=2,
+    )
+
+
+def test_tool_only_partial_dedupe_is_scoped_to_current_user_turn():
+    from api.streaming import _partial_marker_already_present
+
+    existing = [
+        {"role": "user", "content": "first run"},
+        _tool_partial(),
+        {"role": "assistant", "content": "**Task cancelled.**", "_error": True},
+        {"role": "user", "content": "repeat it"},
+    ]
+
+    assert not _partial_marker_already_present(existing, _tool_partial(), before_idx=len(existing))
+
+
+def test_session_load_collapses_adjacent_duplicate_partials(tmp_path, monkeypatch):
+    import api.models as models
+
+    sid = "abc123"
+    session_dir = tmp_path / "sessions"
+    session_dir.mkdir()
+    monkeypatch.setattr(models, "SESSION_DIR", session_dir)
+    monkeypatch.setattr(models, "SESSION_INDEX_FILE", session_dir / "_index.json")
+
+    payload = {
+        "session_id": sid,
+        "title": "bloated partials",
+        "workspace": str(tmp_path),
+        "model": "gpt-5.5",
+        "created_at": 100.0,
+        "updated_at": 200.0,
+        "messages": [
+            {"role": "user", "content": "run this"},
+            _tool_partial(timestamp=123),
+            _tool_partial(timestamp=123),
+            _tool_partial(timestamp=123),
+            {"role": "assistant", "content": "**Task cancelled.**", "_error": True},
+        ],
+        "tool_calls": [],
+    }
+    (session_dir / f"{sid}.json").write_text(json.dumps(payload), encoding="utf-8")
+
+    loaded = models.Session.load(sid)
+
+    assert loaded is not None
+    assert sum(1 for message in loaded.messages if message.get("_partial")) == 1
+    persisted = json.loads((session_dir / f"{sid}.json").read_text(encoding="utf-8"))
+    assert sum(1 for message in persisted["messages"] if message.get("_partial")) == 1
+    assert persisted["updated_at"] == 200.0
+    assert (session_dir / f"{sid}.json.bak").exists()