fix(#3929): preserve partial work on turn-level errors

Extract _build_partial_message() shared by cancel_stream() and the new
_snapshot_and_append_partial_on_error(); the two provider/exception error paths
in _run_agent_streaming now snapshot accumulated text/reasoning/tool_calls under
STREAMS_LOCK and append a _partial assistant message instead of discarding it.

Closes #3929.

Co-authored-by: b3nw <150195942+b3nw@users.noreply.github.com>

api/streaming.py

@@ -4440,6 +4440,96 @@ def _materialize_pending_user_turn_before_error(session) -> bool:
     return True
 
 
+def _build_partial_message(content_text, reasoning_text, tool_calls) -> dict | None:
+    """Build a _partial assistant message from raw streaming buffers.
+
+    Shared by cancel_stream() and _snapshot_and_append_partial_on_error().
+    Strips thinking/reasoning markup, builds the dict, returns None when
+    there is nothing meaningful to preserve.
+    """
+    import re as _re
+    partial_text = (content_text or '').strip()
+    _stripped = ''
+    if partial_text:
+        # First pass: remove complete <thinking>...</thinking> blocks.
+        _stripped = _re.sub(r'<think(?:ing)?\b[^>]*>.*?</think(?:ing)?>',
+                            '', partial_text,
+                            flags=_re.DOTALL | _re.IGNORECASE).strip()
+        # Second pass: strip trailing UNCLOSED think/thinking block (the common
+        # cancel/error case — user stops mid-reasoning before the close tag appears).
+        _stripped = _re.sub(r'<think(?:ing)?\b[^>]*>.*',
+                            '', _stripped,
+                            flags=_re.DOTALL | _re.IGNORECASE).strip()
+    _has_reasoning = bool(reasoning_text and reasoning_text.strip())
+    _has_tools = bool(tool_calls)
+    if not (_stripped or _has_reasoning or _has_tools):
+        return None
+    _msg: dict = {
+        'role': 'assistant',
+        'content': _stripped,  # may be empty for reasoning/tool-only turns
+        '_partial': True,
+        'timestamp': int(time.time()),
+    }
+    if _has_reasoning:
+        _msg['reasoning'] = reasoning_text.strip()
+    if _has_tools:
+        _msg['_partial_tool_calls'] = list(tool_calls)
+    return _msg
+
+
+def _snapshot_and_append_partial_on_error(session, stream_id) -> dict | None:
+    """Snapshot streaming buffers under STREAMS_LOCK and append a _partial message.
+
+    Uses _build_partial_message() for the shared thinking-strip + dict-build logic.
+    """
+    from api import config as _live_config
+
+    streams_lock = STREAMS_LOCK
+    partial_texts = STREAM_PARTIAL_TEXT
+    reasoning_texts = STREAM_REASONING_TEXT
+    live_tool_calls = STREAM_LIVE_TOOL_CALLS
+
+    # Defensive check for live config (similar to cancel_stream)
+    if getattr(_live_config, 'STREAMS_LOCK', streams_lock) is not streams_lock:
+        streams_lock = _live_config.STREAMS_LOCK
+        partial_texts = getattr(_live_config, 'STREAM_PARTIAL_TEXT', partial_texts)
+        reasoning_texts = getattr(_live_config, 'STREAM_REASONING_TEXT', reasoning_texts)
+        live_tool_calls = getattr(_live_config, 'STREAM_LIVE_TOOL_CALLS', live_tool_calls)
+
+    _snap_partial_text = None
+    _snap_reasoning = None
+    _snap_tool_calls = None
+
+    with streams_lock:
+        _snap_partial_text = partial_texts.get(stream_id, '')
+        if not _snap_partial_text:
+            _live_partials = getattr(_live_config, 'STREAM_PARTIAL_TEXT', partial_texts)
+            if _live_partials is not partial_texts:
+                _snap_partial_text = _live_partials.get(stream_id, '')
+
+        _snap_reasoning = reasoning_texts.get(stream_id, '')
+        if not _snap_reasoning:
+            _live_reasoning = getattr(_live_config, 'STREAM_REASONING_TEXT', reasoning_texts)
+            if _live_reasoning is not reasoning_texts:
+                _snap_reasoning = _live_reasoning.get(stream_id, '')
+
+        _snap_tool_calls = list(live_tool_calls.get(stream_id, []) or [])
+        if not _snap_tool_calls:
+            _live_tools = getattr(_live_config, 'STREAM_LIVE_TOOL_CALLS', live_tool_calls)
+            if _live_tools is not live_tool_calls:
+                _snap_tool_calls = list(_live_tools.get(stream_id, []) or [])
+
+    _partial_msg = _build_partial_message(_snap_partial_text, _snap_reasoning, _snap_tool_calls)
+    if _partial_msg is None:
+        return None
+    if not isinstance(session.messages, list):
+        session.messages = []
+    if not _partial_marker_already_present(session.messages, _partial_msg):
+        session.messages.append(_partial_msg)
+        return _partial_msg
+    return None
+
+
 def _last_resort_sync_from_core(session, stream_id, agent_lock):
     """Final-exit guard: if the stream exits with pending_user_message still set,
     sync messages from the core transcript or add an error marker.
@@ -6787,16 +6877,15 @@ def _run_agent_streaming(
                             _err_type,
                             _err_hint,
                         )
-                        # Clear stream/pending state so the session does not appear
-                        # "agent_running" on reload after a silent failure.
-                        # Persist the error so it survives page reload.
-                        # _error=True ensures _sanitize_messages_for_api excludes it from
-                        # subsequent API calls so the LLM never sees its own error as prior context.
                         _materialize_pending_user_turn_before_error(s)
                         s.active_stream_id = None
                         s.pending_user_message = None
                         s.pending_attachments = []
                         s.pending_started_at = None
+                        try:
+                            _snapshot_and_append_partial_on_error(s, stream_id)
+                        except Exception:
+                            logger.debug("Failed to snapshot partials on error for %s", stream_id, exc_info=True)
                         _error_message = {
                             'role': 'assistant',
                             'content': f'**{_err_label}:** {_error_payload.get("message") or _err_label}\n\n*{_err_hint}*',
@@ -7741,11 +7830,16 @@ def _run_agent_streaming(
                         getattr(s, 'active_stream_id', None),
                     )
                     return
+
                 _materialize_pending_user_turn_before_error(s)
                 s.active_stream_id = None
                 s.pending_user_message = None
                 s.pending_attachments = []
                 s.pending_started_at = None
+                try:
+                    _snapshot_and_append_partial_on_error(s, stream_id)
+                except Exception:
+                    logger.debug("Failed to snapshot partials on error for %s", stream_id, exc_info=True)
                 _error_message = {
                     'role': 'assistant',
                     'content': f'**{_exc_label}:** {_error_payload.get("message") or err_str}' + (f'\n\n*{_exc_hint}*' if _exc_hint else ''),
@@ -8233,26 +8327,17 @@ def cancel_stream(stream_id: str) -> bool:
                 # accumulated in thread-local variables but invisible to the cancel path.
                 # This prevents paid-token data loss when cancelling mid-reasoning or
                 # mid-tool-execution.
-                partial_text = _cancel_partial_text.strip() if _cancel_partial_text else ''
-                _stripped = ''
-                if partial_text:
-                    import re as _re
-                    # Strip thinking/reasoning markup from partial content before saving.
-                    # First pass: remove complete <thinking>...</thinking> blocks.
-                    _stripped = _re.sub(r'<think(?:ing)?\b[^>]*>.*?</think(?:ing)?>',
-                                        '', partial_text,
-                                        flags=_re.DOTALL | _re.IGNORECASE).strip()
-                    # Second pass: strip trailing UNCLOSED think/thinking block (the common
-                    # cancel case — user stops mid-reasoning before the close tag appears).
-                    _stripped = _re.sub(r'<think(?:ing)?\b[^>]*>.*',
-                                        '', _stripped,
-                                        flags=_re.DOTALL | _re.IGNORECASE).strip()
-                # Determine whether there is anything to preserve beyond just the
-                # cancel marker.  Content text, reasoning trace, or tool calls all
-                # count (#1361 §C — previously only _stripped was checked, so a
-                # reasoning-only or tool-only stream produced NO partial message).
-                _has_reasoning = bool(_cancel_reasoning and _cancel_reasoning.strip())
-                _has_tools = bool(_cancel_tool_calls)
+                # NOTE on _partial_tool_calls: the captured entries use the WebUI
+                # internal shape {name, args, done, duration, is_error} — they do
+                # NOT carry the OpenAI/Anthropic API id + function: {name, arguments}
+                # envelope. Storing under 'tool_calls' would cause
+                # _sanitize_messages_for_api to forward them to the next-turn LLM
+                # call and strict providers would 400 on the malformed entries.
+                # The underscore-prefixed key is not in the whitelist, so sanitize
+                # strips it. The UI reads it via static/messages.js. (v0.50.251.)
+                _partial_msg = _build_partial_message(
+                    _cancel_partial_text, _cancel_reasoning, _cancel_tool_calls,
+                )
                 _cancel_marker_exists = _session_has_cancel_marker(_cs)
                 _cancel_marker_idx = len(_cs.messages)
                 if _cancel_marker_exists:
@@ -8264,33 +8349,7 @@ def cancel_stream(stream_id: str) -> bool:
                         if any(pattern in _content for pattern in _CANCEL_MARKER_PATTERNS):
                             _cancel_marker_idx = _idx
                             break
-                if _stripped or _has_reasoning or _has_tools:
-                    _partial_msg: dict = {
-                        'role': 'assistant',
-                        'content': _stripped,  # may be empty for reasoning/tool-only turns
-                        '_partial': True,
-                        'timestamp': int(time.time()),
-                    }
-                    if _has_reasoning:
-                        _partial_msg['reasoning'] = _cancel_reasoning.strip()
-                    if _has_tools:
-                        # NOTE: store under the private '_partial_tool_calls' key
-                        # (NOT 'tool_calls'). The captured entries use the WebUI
-                        # internal shape {name, args, done, duration, is_error}
-                        # — they do NOT carry the OpenAI/Anthropic API id +
-                        # function: {name, arguments} envelope. If we put them
-                        # under 'tool_calls', `_sanitize_messages_for_api`
-                        # (which whitelists 'tool_calls' via _API_SAFE_MSG_KEYS)
-                        # would forward them to the next-turn LLM call and
-                        # strict providers (OpenAI, Anthropic, Z.AI/GLM) would
-                        # 400 on the malformed entries — turning a "data lost
-                        # on cancel" bug into a "next message returns 400"
-                        # bug, which is worse. The underscore-prefixed key is
-                        # not in the whitelist, so sanitize strips it. The UI
-                        # reads it via static/messages.js and renders it
-                        # 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)
+                if _partial_msg is not None:
                     # 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

tests/test_issue3929_error_preserves_partial.py

@@ -0,0 +1,247 @@
+"""Tests for Issue #3929: Ensure turn-level error handling preserves partial work in the WebUI session sidecar."""
+
+from __future__ import annotations
+
+import queue
+import sys
+import types
+import pytest
+from unittest import mock
+
+import api.config as config
+import api.models as models
+import api.streaming as streaming
+from api.models import Session
+
+
+@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():
+    """Clear all shared streaming dicts before/after each test."""
+    config.STREAMS.clear()
+    config.CANCEL_FLAGS.clear()
+    config.AGENT_INSTANCES.clear()
+    config.STREAM_PARTIAL_TEXT.clear()
+    if hasattr(config, 'STREAM_REASONING_TEXT'):
+        config.STREAM_REASONING_TEXT.clear()
+    if hasattr(config, 'STREAM_LIVE_TOOL_CALLS'):
+        config.STREAM_LIVE_TOOL_CALLS.clear()
+    yield
+    config.STREAMS.clear()
+    config.CANCEL_FLAGS.clear()
+    config.AGENT_INSTANCES.clear()
+    config.STREAM_PARTIAL_TEXT.clear()
+    if hasattr(config, 'STREAM_REASONING_TEXT'):
+        config.STREAM_REASONING_TEXT.clear()
+    if hasattr(config, 'STREAM_LIVE_TOOL_CALLS'):
+        config.STREAM_LIVE_TOOL_CALLS.clear()
+
+
+@pytest.fixture(autouse=True)
+def _isolate_agent_locks():
+    config.SESSION_AGENT_LOCKS.clear()
+    yield
+    config.SESSION_AGENT_LOCKS.clear()
+
+
+@pytest.fixture(autouse=True)
+def _mock_hermes_modules(monkeypatch):
+    """Inject mock hermes modules to prevent side-effects during tests."""
+    fake_runtime_module = types.ModuleType("hermes_cli.runtime_provider")
+    fake_runtime_provider_fn = lambda requested=None: {
+        "provider": requested or "test-provider",
+        "api_key": "synthetic-key",
+        "base_url": None,
+    }
+    fake_runtime_module.resolve_runtime_provider = fake_runtime_provider_fn
+    fake_hermes_cli = types.ModuleType("hermes_cli")
+    fake_hermes_cli.runtime_provider = fake_runtime_module
+    fake_hermes_state = types.ModuleType("hermes_state")
+    fake_hermes_state.SessionDB = mock.Mock(return_value=None)
+
+    _injected = {
+        "hermes_cli": fake_hermes_cli,
+        "hermes_cli.runtime_provider": fake_runtime_module,
+        "hermes_state": fake_hermes_state,
+    }
+    _MISSING = object()
+    _saved = {k: sys.modules.get(k, _MISSING) for k in _injected}
+    sys.modules.update(_injected)
+    yield
+    for k, prev in _saved.items():
+        if prev is _MISSING:
+            sys.modules.pop(k, None)
+        else:
+            sys.modules[k] = prev
+
+
+class MockAgent:
+    def __init__(self, **kwargs):
+        self.session_id = kwargs.get("session_id")
+        self.stream_delta_callback = kwargs.get("stream_delta_callback")
+        self.reasoning_callback = kwargs.get("reasoning_callback")
+        self.tool_progress_callback = kwargs.get("tool_progress_callback")
+        self.session_prompt_tokens = 0
+        self.session_completion_tokens = 0
+        self.session_estimated_cost_usd = 0.0
+        self.context_compressor = None
+        self._last_error = None
+        self.ephemeral_system_prompt = None
+
+    def run_conversation(self, **kwargs):
+        pass
+
+    def interrupt(self, _message):
+        pass
+
+
+def test_silent_failure_preserves_partials(tmp_path):
+    """Test that a silent failure (agent returns no assistant reply) preserves partial streamed text."""
+    fake_session = Session(session_id="test_sess_silent", title="Test Session")
+    fake_session.pending_user_message = "What is python?"
+    fake_session.active_stream_id = "test_stream_silent"
+    fake_session.save()
+    models.SESSIONS["test_sess_silent"] = fake_session
+
+    class SilentFailureAgent(MockAgent):
+        def run_conversation(self, **kwargs):
+            if self.stream_delta_callback:
+                self.stream_delta_callback("Python is a programming language.")
+            # Return history without new assistant message, plus error status (causes silent failure)
+            return {
+                "status": "error",
+                "error": "Silent failure details",
+                "messages": kwargs.get("conversation_history") or []
+            }
+
+    fake_queue = queue.Queue()
+    streaming.STREAMS["test_stream_silent"] = fake_queue
+    config.STREAM_PARTIAL_TEXT["test_stream_silent"] = ""
+
+    with mock.patch.object(streaming, "get_session", return_value=fake_session), \
+         mock.patch.object(streaming, "_get_ai_agent", return_value=SilentFailureAgent), \
+         mock.patch.object(streaming, "resolve_model_provider", return_value=("test-model", "test-provider", None)), \
+         mock.patch("api.config.get_config", return_value={}), \
+         mock.patch("api.config._resolve_cli_toolsets", return_value=[]):
+        
+        streaming._run_agent_streaming(
+            session_id=fake_session.session_id,
+            msg_text="What is python?",
+            model="test-model",
+            workspace=str(tmp_path),
+            stream_id="test_stream_silent",
+        )
+
+    # Reload session from disk/cache and verify
+    saved = Session.load("test_sess_silent")
+    assert saved is not None
+    
+    # We should have:
+    # 1. The user turn (materialized by _materialize_pending_user_turn_before_error)
+    # 2. The _partial assistant message containing the streamed text
+    # 3. The error block
+    assert len(saved.messages) >= 3, f"Expected at least 3 messages. Got: {saved.messages}"
+    
+    partial_msg = next((m for m in saved.messages if m.get("_partial")), None)
+    assert partial_msg is not None, "Expected a partial message"
+    assert partial_msg["role"] == "assistant"
+    assert partial_msg["content"] == "Python is a programming language."
+
+    err_msg = saved.messages[-1]
+    assert err_msg.get("_error") is True
+
+
+def test_exception_preserves_partials(tmp_path):
+    """Test that an unhandled exception preserves partial streamed text."""
+    fake_session = Session(session_id="test_sess_exc", title="Test Session")
+    fake_session.pending_user_message = "Exception test"
+    fake_session.active_stream_id = "test_stream_exc"
+    fake_session.save()
+    models.SESSIONS["test_sess_exc"] = fake_session
+
+    class ExceptionAgent(MockAgent):
+        def run_conversation(self, **kwargs):
+            if self.stream_delta_callback:
+                self.stream_delta_callback("Stream before crash.")
+            raise RuntimeError("Fake provider crash!")
+
+    fake_queue = queue.Queue()
+    streaming.STREAMS["test_stream_exc"] = fake_queue
+    config.STREAM_PARTIAL_TEXT["test_stream_exc"] = ""
+
+    with mock.patch.object(streaming, "get_session", return_value=fake_session), \
+         mock.patch.object(streaming, "_get_ai_agent", return_value=ExceptionAgent), \
+         mock.patch.object(streaming, "resolve_model_provider", return_value=("test-model", "test-provider", None)), \
+         mock.patch("api.config.get_config", return_value={}), \
+         mock.patch("api.config._resolve_cli_toolsets", return_value=[]):
+        
+        streaming._run_agent_streaming(
+            session_id=fake_session.session_id,
+            msg_text="Exception test",
+            model="test-model",
+            workspace=str(tmp_path),
+            stream_id="test_stream_exc",
+        )
+
+    saved = Session.load("test_sess_exc")
+    assert saved is not None
+    
+    partial_msg = next((m for m in saved.messages if m.get("_partial")), None)
+    assert partial_msg is not None, "Expected a partial message before error"
+    assert partial_msg["content"] == "Stream before crash."
+
+    err_msg = saved.messages[-1]
+    assert err_msg.get("_error") is True
+    assert "Fake provider crash!" in err_msg.get("content", "")
+
+
+def test_empty_partials_do_not_create_spurious_messages(tmp_path):
+    """Test that if no content has been streamed, no _partial message is created on error."""
+    fake_session = Session(session_id="test_sess_empty", title="Test Session")
+    fake_session.pending_user_message = "Empty test"
+    fake_session.active_stream_id = "test_stream_empty"
+    fake_session.save()
+    models.SESSIONS["test_sess_empty"] = fake_session
+
+    class ExceptionAgent(MockAgent):
+        def run_conversation(self, **kwargs):
+            raise RuntimeError("Fake provider crash immediately!")
+
+    fake_queue = queue.Queue()
+    streaming.STREAMS["test_stream_empty"] = fake_queue
+    config.STREAM_PARTIAL_TEXT["test_stream_empty"] = ""
+
+    with mock.patch.object(streaming, "get_session", return_value=fake_session), \
+         mock.patch.object(streaming, "_get_ai_agent", return_value=ExceptionAgent), \
+         mock.patch.object(streaming, "resolve_model_provider", return_value=("test-model", "test-provider", None)), \
+         mock.patch("api.config.get_config", return_value={}), \
+         mock.patch("api.config._resolve_cli_toolsets", return_value=[]):
+        
+        streaming._run_agent_streaming(
+            session_id=fake_session.session_id,
+            msg_text="Empty test",
+            model="test-model",
+            workspace=str(tmp_path),
+            stream_id="test_stream_empty",
+        )
+
+    saved = Session.load("test_sess_empty")
+    assert saved is not None
+    
+    # We should have user message and error message, but NO _partial message
+    assert not any(m.get("_partial") for m in saved.messages)
+    
+    err_msg = saved.messages[-1]
+    assert err_msg.get("_error") is True