Merge pull request #6186 from franksong2702/franksong2702/investigate-3929-partial-work

Live Stream: recover interrupted Thinking after restart

api/models.py

@@ -805,6 +805,28 @@ def _clear_webui_deleted_session_tombstone(sid: str) -> None:
             logger.debug("Failed to remove empty webui deleted-session tombstone", exc_info=True)
 
 
+def _content_has_reasoning_only_parts(content) -> bool:
+    if not isinstance(content, list) or not content:
+        return False
+    saw_reasoning = False
+    for part in content:
+        if not isinstance(part, dict):
+            if str(part or '').strip():
+                return False
+            continue
+        part_type = str(part.get('type') or '').lower()
+        if part_type in {'thinking', 'reasoning'}:
+            text = part.get('thinking') or part.get('reasoning') or part.get('text') or ''
+            if str(text).strip():
+                saw_reasoning = True
+            continue
+        if part_type == 'text' and str(part.get('text') or part.get('content') or '').strip():
+            return False
+        if part_type not in {'text', 'thinking', 'reasoning'}:
+            return False
+    return saw_reasoning
+
+
 def _active_stream_ids():
     with STREAMS_LOCK:
         active_ids = set(STREAMS.keys())
@@ -819,13 +841,30 @@ def _active_stream_ids():
     return active_ids
 
 
-def _append_recovered_turn_to_context(session, recovered: dict) -> None:
-    context_messages = getattr(session, 'context_messages', None)
-    if not isinstance(context_messages, list) or not context_messages:
-        return
+def _recovered_model_context_projection(message: dict) -> dict | None:
+    if not isinstance(message, dict):
+        return None
+    projected = dict(message)
+    projected.pop('reasoning', None)
+    if projected.get('_error'):
+        return None
+    if _content_has_reasoning_only_parts(projected.get('content')):
+        if projected.get('tool_calls'):
+            projected['content'] = ''
+        else:
+            return None
+    projected_text = _normalize_journal_recovery_text(projected.get('content'))
+    if not projected_text and not projected.get('tool_call_id') and not projected.get('tool_calls'):
+        return None
+    return projected
+
+
+def _append_recovered_context_projection(
+    session,
+    context_messages: list,
+    recovered: dict,
+) -> None:
     recovered_text = _normalize_journal_recovery_text(recovered.get('content'))
-    if not recovered_text and not recovered.get('tool_call_id') and not recovered.get('tool_calls'):
-        return
     if recovered_text:
         if recovered.get('role') == 'user':
             if _message_matches_pending_checkpoint(
@@ -845,6 +884,27 @@ def _append_recovered_turn_to_context(session, recovered: dict) -> None:
     context_messages.append(dict(recovered))
 
 
+def _seed_recovered_context_from_messages(session, context_messages: list) -> None:
+    for message in getattr(session, 'messages', None) or []:
+        projected = _recovered_model_context_projection(message)
+        if projected is None:
+            continue
+        context_messages.append(projected)
+
+
+def _append_recovered_turn_to_context(session, recovered: dict) -> None:
+    context_messages = getattr(session, 'context_messages', None)
+    if not isinstance(context_messages, list):
+        context_messages = []
+        session.context_messages = context_messages
+    if not context_messages:
+        _seed_recovered_context_from_messages(session, context_messages)
+    projected = _recovered_model_context_projection(recovered)
+    if projected is None:
+        return
+    _append_recovered_context_projection(session, context_messages, projected)
+
+
 def _append_recovered_pending_turn(session, *, timestamp: int | None = None) -> dict | None:
     pending_text = str(session.pending_user_message or '')
     if not pending_text:
@@ -2306,11 +2366,23 @@ def _collapse_adjacent_duplicate_partials(messages) -> tuple[list, bool]:
     return collapsed, changed
 
 
-def _find_existing_assistant_for_journal_content(session, content: str) -> int | None:
+def _find_existing_assistant_for_journal_content(
+    session,
+    content: str,
+    *,
+    max_index: int | None = None,
+    excluded_indexes: set[int] | None = None,
+) -> int | None:
     candidate = _normalize_journal_recovery_text(content)
     if not candidate:
         return None
-    for idx, message in enumerate(session.messages or []):
+    messages = session.messages or []
+    stop = len(messages) if max_index is None else min(len(messages), max_index)
+    substring_match = None
+    for idx in range(stop):
+        if excluded_indexes and idx in excluded_indexes:
+            continue
+        message = messages[idx]
         if not isinstance(message, dict) or message.get('role') != 'assistant':
             continue
         if message.get('_error'):
@@ -2320,9 +2392,9 @@ def _find_existing_assistant_for_journal_content(session, content: str) -> int |
             continue
         if existing == candidate:
             return idx
-        if len(candidate) >= 24 and candidate in existing:
-            return idx
-    return None
+        if substring_match is None and len(candidate) >= 24 and candidate in existing:
+            substring_match = idx
+    return substring_match
 
 
 def _journal_tool_already_present(
@@ -2394,6 +2466,12 @@ def _run_journal_has_visible_output(session, stream_id: str | None) -> bool:
                 continue
             if str(payload.get('text') or '').strip():
                 return True
+        if event_name == 'reasoning':
+            reasoning_text = str(
+                payload.get('text') or payload.get('reasoning') or payload.get('thinking') or ''
+            )
+            if reasoning_text.strip():
+                return True
         if event_name == 'tool':
             return True
     return False
@@ -2454,9 +2532,10 @@ def _append_journaled_partial_output(
     """Recover already-emitted visible output from a dead stream journal.
 
     This repair path is intentionally conservative: it restores user-visible
-    assistant text and tool-card metadata that had already been emitted over
-    SSE before the WebUI process died. It does not restore hidden reasoning and
-    it does not try to continue execution.
+    assistant text, display-only reasoning, and tool-card metadata that had
+    already been emitted over SSE before the WebUI process died. Restored
+    reasoning stays out of ``context_messages`` so it cannot become provider-
+    facing history. The repair does not try to continue execution.
     """
     if not stream_id:
         return False
@@ -2479,24 +2558,120 @@ def _append_journaled_partial_output(
 
     appended_any = False
     assistant_parts: list[str] = []
+    reasoning_parts: list[str] = []
     assistant_started_at: float | None = None
     current_assistant_idx: int | None = None
     recovered_tool_calls: list[dict] = []
+    initial_message_count = len(session.messages or [])
+    claimed_existing_assistant_indexes: set[int] = set()
+
+    def content_match_can_receive_reasoning(existing_idx: int) -> bool:
+        messages = session.messages or []
+        owner_idx = None
+        for candidate_idx in range(existing_idx - 1, -1, -1):
+            candidate = messages[candidate_idx]
+            if isinstance(candidate, dict) and candidate.get('role') == 'user':
+                owner_idx = candidate_idx
+                break
+        if owner_idx is None:
+            return False
+
+        pending_text = _normalize_journal_recovery_text(session.pending_user_message)
+        if pending_text and not _message_matches_pending_checkpoint(
+            messages[owner_idx],
+            session.pending_user_message,
+            session.pending_started_at,
+            session.pending_user_source,
+            session.pending_attachments,
+        ):
+            return False
+
+        for candidate_idx in range(existing_idx + 1, initial_message_count):
+            candidate = messages[candidate_idx]
+            if not isinstance(candidate, dict) or candidate.get('role') != 'user':
+                continue
+            candidate_text = _normalize_journal_recovery_text(candidate.get('content'))
+            candidate_matches_checkpoint = pending_text and _message_matches_pending_checkpoint(
+                candidate,
+                session.pending_user_message,
+                session.pending_started_at,
+                session.pending_user_source,
+                session.pending_attachments,
+            )
+            if candidate_matches_checkpoint and candidate.get('_recovered'):
+                continue
+            if pending_text and candidate_text == pending_text:
+                return False
+            return False
+        return True
+
+    def append_context_projection(message: dict) -> None:
+        context_projection = dict(message)
+        context_projection.pop('reasoning', None)
+        _append_recovered_turn_to_context(session, context_projection)
+
+    def attach_display_reasoning(message: dict, reasoning: str) -> bool:
+        if not reasoning:
+            return False
+        existing = str(message.get('reasoning') or '').strip()
+        if existing:
+            return False
+        message['reasoning'] = reasoning
+        return True
 
     def flush_assistant() -> int | None:
-        nonlocal appended_any, assistant_parts, assistant_started_at, current_assistant_idx
+        nonlocal appended_any, assistant_parts, reasoning_parts
+        nonlocal assistant_started_at, current_assistant_idx
         content = ''.join(assistant_parts).strip()
+        reasoning = ''.join(reasoning_parts).strip()
         assistant_parts = []
-        if not content:
+        reasoning_parts = []
+        if not content and not reasoning:
             return current_assistant_idx
-        if dedupe_existing:
-            existing_idx = _find_existing_assistant_for_journal_content(session, content)
+        if dedupe_existing and content:
+            search_excluded = set(claimed_existing_assistant_indexes)
+            existing_idx = None
+            while True:
+                candidate_idx = _find_existing_assistant_for_journal_content(
+                    session,
+                    content,
+                    max_index=initial_message_count,
+                    excluded_indexes=search_excluded,
+                )
+                if candidate_idx is None:
+                    break
+                if not reasoning or content_match_can_receive_reasoning(candidate_idx):
+                    existing_idx = candidate_idx
+                    break
+                search_excluded.add(candidate_idx)
             if existing_idx is not None:
+                claimed_existing_assistant_indexes.add(existing_idx)
                 current_assistant_idx = existing_idx
                 assistant_started_at = None
                 if 0 <= existing_idx < len(session.messages):
-                    _append_recovered_turn_to_context(session, session.messages[existing_idx])
+                    existing_message = session.messages[existing_idx]
+                    append_context_projection(existing_message)
+                    if attach_display_reasoning(existing_message, reasoning):
+                        appended_any = True
                 return existing_idx
+        if dedupe_existing and reasoning and not content:
+            for existing_idx in range(initial_message_count):
+                if existing_idx in claimed_existing_assistant_indexes:
+                    continue
+                existing_message = session.messages[existing_idx]
+                if not isinstance(existing_message, dict):
+                    continue
+                if (
+                    existing_message.get('_recovered_from_run_journal')
+                    and existing_message.get('_recovered_stream_id') == stream_id
+                    and existing_message.get('role') == 'assistant'
+                    and not str(existing_message.get('content') or '').strip()
+                    and str(existing_message.get('reasoning') or '').strip() == reasoning
+                ):
+                    claimed_existing_assistant_indexes.add(existing_idx)
+                    current_assistant_idx = existing_idx
+                    assistant_started_at = None
+                    return existing_idx
         timestamp = int(assistant_started_at or time.time())
         recovered_assistant = {
             'role': 'assistant',
@@ -2505,8 +2680,9 @@ def _append_journaled_partial_output(
             '_recovered_from_run_journal': True,
             '_recovered_stream_id': stream_id,
         }
+        attach_display_reasoning(recovered_assistant, reasoning)
         session.messages.append(recovered_assistant)
-        _append_recovered_turn_to_context(session, recovered_assistant)
+        append_context_projection(recovered_assistant)
         current_assistant_idx = len(session.messages) - 1
         assistant_started_at = None
         appended_any = True
@@ -2541,6 +2717,7 @@ def _append_journaled_partial_output(
                 and _m.get('_recovered_stream_id') == stream_id
                 and _m.get('role') == 'assistant'
                 and not str(_m.get('content') or '').strip()
+                and not str(_m.get('reasoning') or '').strip()
             ):
                 current_assistant_idx = _existing_idx
                 return _existing_idx
@@ -2559,6 +2736,16 @@ def _append_journaled_partial_output(
         event_name = str(event.get('event') or event.get('type') or '')
         payload = event.get('payload') if isinstance(event.get('payload'), dict) else {}
         created_at = event.get('created_at') if isinstance(event.get('created_at'), (int, float)) else None
+        if event_name == 'reasoning':
+            text = str(
+                payload.get('text') or payload.get('reasoning') or payload.get('thinking') or ''
+            )
+            if not text:
+                continue
+            if not assistant_parts and not reasoning_parts and assistant_started_at is None:
+                assistant_started_at = created_at or time.time()
+            reasoning_parts.append(text)
+            continue
         if event_name == 'token':
             text = str(payload.get('text') or '')
             if not text:

tests/test_issue3929_partial_work_recovery.py

@@ -0,0 +1,810 @@
+"""Regression coverage for #3929 restart-time partial-work recovery."""
+
+import json
+import time
+
+import pytest
+
+import api.models as models
+from api.models import (
+    Session,
+    _append_journaled_partial_output,
+    _apply_core_sync_or_error_marker,
+)
+from api.run_journal import append_run_event
+import api.streaming as streaming
+from api.streaming import _sanitize_messages_for_api
+
+
+@pytest.fixture(autouse=True)
+def _isolate_session_state(tmp_path, monkeypatch):
+    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")
+    models.SESSIONS.clear()
+    yield
+    models.SESSIONS.clear()
+
+
+def test_restart_recovery_preserves_display_reasoning_without_provider_replay(
+    tmp_path,
+):
+    """A dead live turn must recover all user-visible work from its journal.
+
+    Display-only reasoning belongs in the settled Worklog after reload, but it
+    must not leak into the model-facing history used by the next turn.
+    """
+    session_id = "issue3929_restart"
+    stream_id = "stream_partial_work"
+    previous_messages = [
+        {"role": "user", "content": "Inspect the failure"},
+        {"role": "assistant", "content": "I will trace it."},
+    ]
+    session = Session(
+        session_id=session_id,
+        title="Partial work recovery",
+        messages=[dict(message) for message in previous_messages],
+        context_messages=[dict(message) for message in previous_messages],
+        pending_user_message="Continue until the root cause is clear",
+        pending_started_at=time.time() - 120,
+        active_stream_id=stream_id,
+    )
+
+    append_run_event(
+        session_id,
+        stream_id,
+        "reasoning",
+        {"text": "Checking the first failure boundary."},
+    )
+    append_run_event(
+        session_id,
+        stream_id,
+        "token",
+        {"text": "The first boundary is the live stream."},
+    )
+    append_run_event(
+        session_id,
+        stream_id,
+        "tool",
+        {
+            "name": "terminal",
+            "preview": "rg STREAM_PARTIAL_TEXT api/streaming.py",
+            "args": {"command": "rg STREAM_PARTIAL_TEXT api/streaming.py"},
+        },
+    )
+    append_run_event(
+        session_id,
+        stream_id,
+        "tool_complete",
+        {"name": "terminal", "duration": 0.4, "is_error": False},
+    )
+    append_run_event(
+        session_id,
+        stream_id,
+        "reasoning",
+        {"text": "Comparing the journal with the session sidecar."},
+    )
+    append_run_event(
+        session_id,
+        stream_id,
+        "token",
+        {"text": "The journal still has the emitted work."},
+    )
+
+    core_path = tmp_path / "missing-core-transcript.json"
+    assert _apply_core_sync_or_error_marker(
+        session,
+        core_path,
+        stream_id_for_recheck=stream_id,
+    ) is True
+
+    models.SESSIONS.clear()
+    reloaded = models.get_session(session_id)
+    recovered = [
+        message for message in reloaded.messages
+        if message.get("_recovered_from_run_journal")
+    ]
+
+    assert [message.get("content") for message in recovered] == [
+        "The first boundary is the live stream.",
+        "The journal still has the emitted work.",
+    ]
+    assert [message.get("reasoning") for message in recovered] == [
+        "Checking the first failure boundary.",
+        "Comparing the journal with the session sidecar.",
+    ]
+    assert reloaded.tool_calls[0]["name"] == "terminal"
+    assert reloaded.tool_calls[0]["done"] is True
+    assert any(
+        message.get("_error")
+        and message.get("type") == "interrupted"
+        and "partial output above was recovered" in message.get("content", "")
+        for message in reloaded.messages
+    )
+
+    serialized_context = json.dumps(
+        reloaded.context_messages, ensure_ascii=False,
+    )
+    assert "Checking the first failure boundary" not in serialized_context
+    assert "Comparing the journal with the session sidecar" not in serialized_context
+    sanitized = _sanitize_messages_for_api(reloaded.context_messages)
+    assert all("reasoning" not in message for message in sanitized)
+
+
+def test_core_sync_keeps_pending_owner_for_reasoning_only_partial(tmp_path):
+    """Reasoning-only live work still proves the pending user turn had activity."""
+    session_id = "issue3929_reasoning_only"
+    stream_id = "stream_reasoning_only"
+    session = Session(
+        session_id=session_id,
+        title="Reasoning-only recovery",
+        messages=[],
+        context_messages=[
+            {"role": "user", "content": "Earlier question"},
+            {"role": "assistant", "content": "Earlier answer"},
+        ],
+        pending_user_message="Continue the investigation",
+        pending_started_at=time.time() - 90,
+        active_stream_id=stream_id,
+    )
+    core_path = tmp_path / "core-transcript.json"
+    core_path.write_text(json.dumps({
+        "messages": [
+            {"role": "user", "content": "Earlier question"},
+            {"role": "assistant", "content": "Earlier answer"},
+        ],
+    }), encoding="utf-8")
+    append_run_event(
+        session_id,
+        stream_id,
+        "reasoning",
+        {"text": "The worker reached a visible Thinking step before restart."},
+    )
+
+    assert _apply_core_sync_or_error_marker(
+        session,
+        core_path,
+        stream_id_for_recheck=stream_id,
+    ) is True
+
+    recovered_users = [message for message in session.messages if message.get("_recovered")]
+    assert [message.get("content") for message in recovered_users] == [
+        "Continue the investigation",
+    ]
+    recovered_reasoning = [
+        message for message in session.messages
+        if message.get("_recovered_from_run_journal") and message.get("reasoning")
+    ]
+    assert [message.get("reasoning") for message in recovered_reasoning] == [
+        "The worker reached a visible Thinking step before restart.",
+    ]
+    assert any(
+        message.get("_error")
+        and "partial output above was recovered" in message.get("content", "")
+        for message in session.messages
+    )
+    assert not any(
+        "visible Thinking step" in json.dumps(message, ensure_ascii=False)
+        for message in session.context_messages
+    )
+
+
+def test_reasoning_backfill_is_idempotent_for_existing_recovered_text():
+    session_id = "issue3929_reasoning_dedupe"
+    stream_id = "stream_reasoning_dedupe"
+    recovered_text = "The visible partial was already restored."
+    session = Session(
+        session_id=session_id,
+        title="Reasoning backfill",
+        messages=[
+            {"role": "user", "content": "Continue"},
+            {"role": "assistant", "content": recovered_text},
+        ],
+        context_messages=[
+            {"role": "user", "content": "Continue"},
+            {"role": "assistant", "content": recovered_text},
+        ],
+    )
+    append_run_event(
+        session_id,
+        stream_id,
+        "reasoning",
+        {"text": "Backfill this Thinking row once."},
+    )
+    append_run_event(session_id, stream_id, "token", {"text": recovered_text})
+
+    assert _append_journaled_partial_output(
+        session, stream_id, dedupe_existing=True,
+    ) is True
+    assert _append_journaled_partial_output(
+        session, stream_id, dedupe_existing=True,
+    ) is False
+
+    matching = [
+        message for message in session.messages
+        if message.get("role") == "assistant" and message.get("content") == recovered_text
+    ]
+    assert len(matching) == 1
+    assert matching[0].get("reasoning") == "Backfill this Thinking row once."
+    assert "Backfill this Thinking row once" not in json.dumps(
+        session.context_messages, ensure_ascii=False,
+    )
+
+
+def test_reasoning_backfill_does_not_claim_matching_content_from_prior_turn():
+    session_id = "issue3929_reasoning_current_turn_scope"
+    stream_id = "stream_reasoning_current_turn_scope"
+    repeated_text = "The same sufficiently long partial output appears again."
+    prior_assistant = {"role": "assistant", "content": repeated_text}
+    session = Session(
+        session_id=session_id,
+        title="Current-turn reasoning scope",
+        messages=[
+            {"role": "user", "content": "Earlier request"},
+            prior_assistant,
+            {"role": "user", "content": "Current request"},
+        ],
+        context_messages=[
+            {"role": "user", "content": "Earlier request"},
+            {"role": "assistant", "content": repeated_text},
+            {"role": "user", "content": "Current request"},
+        ],
+    )
+    append_run_event(
+        session_id,
+        stream_id,
+        "reasoning",
+        {"text": "This Thinking belongs only to the current request."},
+    )
+    append_run_event(session_id, stream_id, "token", {"text": repeated_text})
+
+    assert _append_journaled_partial_output(
+        session, stream_id, dedupe_existing=True,
+    ) is True
+
+    assert prior_assistant.get("reasoning") is None
+    current_rows = [
+        message for message in session.messages[3:]
+        if message.get("content") == repeated_text
+    ]
+    assert len(current_rows) == 1
+    assert current_rows[0].get("reasoning") == (
+        "This Thinking belongs only to the current request."
+    )
+    assert "This Thinking belongs only to the current request" not in json.dumps(
+        session.context_messages, ensure_ascii=False,
+    )
+
+
+def test_repeated_pending_prompt_uses_checkpoint_owner_not_prior_same_text():
+    session_id = "issue3929_repeated_pending_prompt"
+    stream_id = "stream_repeated_pending_prompt"
+    repeated_text = "The same sufficiently long recovered answer appears."
+    prior_assistant = {"role": "assistant", "content": repeated_text}
+    pending_started_at = 2_000
+    session = Session(
+        session_id=session_id,
+        title="Repeated pending prompt",
+        messages=[
+            {"role": "user", "content": "Continue", "timestamp": 1_000},
+            prior_assistant,
+            {
+                "role": "user",
+                "content": "Continue",
+                "timestamp": pending_started_at,
+                "_recovered": True,
+            },
+        ],
+        context_messages=[
+            {"role": "user", "content": "Continue", "timestamp": 1_000},
+            {"role": "assistant", "content": repeated_text},
+            {
+                "role": "user",
+                "content": "Continue",
+                "timestamp": pending_started_at,
+                "_recovered": True,
+            },
+        ],
+        pending_user_message="Continue",
+        pending_started_at=pending_started_at,
+    )
+    append_run_event(
+        session_id,
+        stream_id,
+        "reasoning",
+        {"text": "This Thinking belongs to the second Continue turn."},
+    )
+    append_run_event(session_id, stream_id, "token", {"text": repeated_text})
+
+    assert _append_journaled_partial_output(
+        session, stream_id, dedupe_existing=True,
+    ) is True
+
+    assert prior_assistant.get("reasoning") is None
+    current_rows = [
+        message for message in session.messages[3:]
+        if message.get("content") == repeated_text
+    ]
+    assert len(current_rows) == 1
+    assert current_rows[0].get("reasoning") == (
+        "This Thinking belongs to the second Continue turn."
+    )
+    assert "second Continue turn" not in json.dumps(
+        session.context_messages, ensure_ascii=False,
+    )
+
+
+def test_empty_context_recovery_seeds_reasoning_free_model_context():
+    session_id = "issue3929_empty_context_reasoning"
+    stream_id = "stream_empty_context_reasoning"
+    session = Session(
+        session_id=session_id,
+        title="Empty context reasoning",
+        messages=[
+            {"role": "user", "content": "Continue", "timestamp": 1234},
+        ],
+        context_messages=[],
+    )
+    append_run_event(
+        session_id,
+        stream_id,
+        "reasoning",
+        {"text": "Display-only Thinking must not become model context."},
+    )
+
+    assert _append_journaled_partial_output(
+        session, stream_id, dedupe_existing=True,
+    ) is True
+
+    assert [message.get("content") for message in session.context_messages] == [
+        "Continue",
+    ]
+    serialized_context = json.dumps(
+        session.context_messages, ensure_ascii=False,
+    )
+    assert "Display-only Thinking" not in serialized_context
+    assert all("reasoning" not in message for message in session.context_messages)
+
+    next_turn_context = streaming._context_messages_for_new_turn(
+        session, "Now continue from there",
+    )
+    assert next_turn_context == session.context_messages
+
+
+def test_empty_context_recovery_omits_structured_reasoning_only_content(tmp_path):
+    session_id = "issue3929_empty_context_structured_reasoning"
+    stream_id = "stream_empty_context_structured_reasoning"
+    session = Session(
+        session_id=session_id,
+        title="Structured reasoning-only content",
+        messages=[
+            {"role": "user", "content": "Continue", "timestamp": 1234},
+            {
+                "role": "assistant",
+                "content": [
+                    {
+                        "type": "reasoning",
+                        "reasoning": "Structured Thinking must stay display-only.",
+                    },
+                ],
+            },
+        ],
+        context_messages=[],
+    )
+    append_run_event(
+        session_id,
+        stream_id,
+        "token",
+        {"text": "Recovered visible answer."},
+    )
+
+    assert _append_journaled_partial_output(
+        session, stream_id, dedupe_existing=True,
+    ) is True
+    session.save()
+
+    models.SESSIONS.clear()
+    reloaded = models.get_session(session_id)
+    context_projection = [
+        (message.get("role"), message.get("content"))
+        for message in reloaded.context_messages
+    ]
+    assert context_projection == [
+        ("user", "Continue"),
+        ("assistant", "Recovered visible answer."),
+    ]
+    serialized_context = json.dumps(
+        reloaded.context_messages, ensure_ascii=False,
+    )
+    assert "Structured Thinking" not in serialized_context
+    sanitized = _sanitize_messages_for_api(reloaded.context_messages)
+    assert [
+        (message.get("role"), message.get("content"))
+        for message in sanitized
+    ] == [
+        ("user", "Continue"),
+        ("assistant", "Recovered visible answer."),
+    ]
+
+
+def test_empty_context_recovery_preserves_tool_calls_with_reasoning_only_content():
+    session_id = "issue3929_empty_context_reasoning_tool_call"
+    stream_id = "stream_empty_context_reasoning_tool_call"
+    tool_call = {
+        "id": "call_issue3929",
+        "type": "function",
+        "function": {"name": "inspect_state", "arguments": "{}"},
+    }
+    session = Session(
+        session_id=session_id,
+        title="Reasoning-only tool boundary",
+        messages=[
+            {"role": "user", "content": "Inspect the state"},
+            {
+                "role": "assistant",
+                "content": [
+                    {
+                        "type": "reasoning",
+                        "reasoning": "Tool planning Thinking must stay display-only.",
+                    },
+                ],
+                "tool_calls": [tool_call],
+            },
+            {
+                "role": "tool",
+                "tool_call_id": "call_issue3929",
+                "content": "{\"ok\": true}",
+            },
+        ],
+        context_messages=[],
+    )
+    append_run_event(
+        session_id,
+        stream_id,
+        "token",
+        {"text": "Recovered visible answer."},
+    )
+
+    assert _append_journaled_partial_output(
+        session, stream_id, dedupe_existing=True,
+    ) is True
+    session.save()
+
+    models.SESSIONS.clear()
+    reloaded = models.get_session(session_id)
+    assert [message.get("role") for message in reloaded.context_messages] == [
+        "user",
+        "assistant",
+        "tool",
+        "assistant",
+    ]
+    assert reloaded.context_messages[1]["content"] == ""
+    assert reloaded.context_messages[1]["tool_calls"] == [tool_call]
+    assert reloaded.context_messages[2]["tool_call_id"] == "call_issue3929"
+    assert "Tool planning Thinking" not in json.dumps(
+        reloaded.context_messages, ensure_ascii=False,
+    )
+
+    sanitized = _sanitize_messages_for_api(reloaded.context_messages)
+    assert [message.get("role") for message in sanitized] == [
+        "user",
+        "assistant",
+        "tool",
+        "assistant",
+    ]
+    assert sanitized[1]["tool_calls"] == [tool_call]
+    assert sanitized[2]["tool_call_id"] == "call_issue3929"
+
+
+def test_empty_context_recovery_preserves_duplicate_historical_replies():
+    session_id = "issue3929_empty_context_duplicate_history"
+    stream_id = "stream_empty_context_duplicate_history"
+    session = Session(
+        session_id=session_id,
+        title="Duplicate historical replies",
+        messages=[
+            {"role": "user", "content": "First check"},
+            {"role": "assistant", "content": "Same status."},
+            {"role": "user", "content": "Second check"},
+            {"role": "assistant", "content": "Same status."},
+        ],
+        context_messages=[],
+    )
+    append_run_event(
+        session_id,
+        stream_id,
+        "reasoning",
+        {"text": "Display-only Thinking should not affect history shape."},
+    )
+
+    assert _append_journaled_partial_output(
+        session, stream_id, dedupe_existing=True,
+    ) is True
+
+    assert [
+        (message.get("role"), message.get("content"))
+        for message in session.context_messages
+    ] == [
+        ("user", "First check"),
+        ("assistant", "Same status."),
+        ("user", "Second check"),
+        ("assistant", "Same status."),
+    ]
+
+
+def test_reasoning_backfill_accepts_core_row_before_recovered_owner_echo():
+    session_id = "issue3929_reasoning_core_owner_echo"
+    stream_id = "stream_reasoning_core_owner_echo"
+    pending_text = "Current request"
+    pending_started_at = 3_000
+    recovered_text = "The core transcript already contains this partial output."
+    core_assistant = {"role": "assistant", "content": recovered_text}
+    session = Session(
+        session_id=session_id,
+        title="Core owner echo",
+        messages=[
+            {"role": "user", "content": pending_text, "timestamp": pending_started_at},
+            core_assistant,
+            {
+                "role": "user",
+                "content": pending_text,
+                "timestamp": pending_started_at,
+                "_recovered": True,
+            },
+        ],
+        context_messages=[
+            {"role": "user", "content": pending_text, "timestamp": pending_started_at},
+            {"role": "assistant", "content": recovered_text},
+        ],
+        pending_user_message=pending_text,
+        pending_started_at=pending_started_at,
+    )
+    append_run_event(
+        session_id,
+        stream_id,
+        "reasoning",
+        {"text": "Attach this Thinking to the current core row."},
+    )
+    append_run_event(session_id, stream_id, "token", {"text": recovered_text})
+
+    assert _append_journaled_partial_output(
+        session, stream_id, dedupe_existing=True,
+    ) is True
+
+    assert core_assistant.get("reasoning") == (
+        "Attach this Thinking to the current core row."
+    )
+    matching_rows = [
+        message for message in session.messages
+        if message.get("content") == recovered_text
+    ]
+    assert len(matching_rows) == 1
+    assert "Attach this Thinking to the current core row" not in json.dumps(
+        session.context_messages, ensure_ascii=False,
+    )
+
+
+def test_reasoning_only_recovery_is_idempotent_across_replays():
+    session_id = "issue3929_reasoning_only_dedupe"
+    stream_id = "stream_reasoning_only_dedupe"
+    session = Session(
+        session_id=session_id,
+        title="Reasoning-only replay",
+        messages=[{"role": "user", "content": "Continue"}],
+        context_messages=[{"role": "user", "content": "Continue"}],
+    )
+    append_run_event(
+        session_id,
+        stream_id,
+        "reasoning",
+        {"text": "Restore this Thinking row once."},
+    )
+
+    assert _append_journaled_partial_output(
+        session, stream_id, dedupe_existing=True,
+    ) is True
+    assert _append_journaled_partial_output(
+        session, stream_id, dedupe_existing=True,
+    ) is False
+
+    reasoning_rows = [
+        message for message in session.messages
+        if message.get("reasoning") == "Restore this Thinking row once."
+    ]
+    assert len(reasoning_rows) == 1
+
+
+def test_retry_growth_attaches_later_tool_to_reasoning_segment():
+    session_id = "issue3929_reasoning_then_tool_growth"
+    stream_id = "stream_reasoning_then_tool_growth"
+    session = Session(
+        session_id=session_id,
+        title="Reasoning then tool growth",
+        messages=[{"role": "user", "content": "Keep checking"}],
+        context_messages=[{"role": "user", "content": "Keep checking"}],
+    )
+    append_run_event(
+        session_id,
+        stream_id,
+        "reasoning",
+        {"text": "Inspect the first boundary."},
+    )
+
+    assert _append_journaled_partial_output(
+        session, stream_id, dedupe_existing=True,
+    ) is True
+    reasoning_idx = next(
+        idx for idx, message in enumerate(session.messages)
+        if message.get("reasoning") == "Inspect the first boundary."
+    )
+
+    append_run_event(
+        session_id,
+        stream_id,
+        "tool",
+        {"name": "terminal", "preview": "inspect the next boundary"},
+    )
+
+    assert _append_journaled_partial_output(
+        session, stream_id, dedupe_existing=True,
+    ) is True
+    reasoning_rows = [
+        message for message in session.messages
+        if message.get("reasoning") == "Inspect the first boundary."
+    ]
+    assert len(reasoning_rows) == 1
+    assert session.tool_calls[0]["assistant_msg_idx"] == reasoning_idx
+
+
+def test_tool_before_reasoning_keeps_own_anchor_on_regrowth():
+    session_id = "issue3929_tool_before_reasoning_regrowth"
+    stream_id = "stream_tool_before_reasoning_regrowth"
+    session = Session(
+        session_id=session_id,
+        title="Tool before reasoning regrowth",
+        messages=[
+            {"role": "user", "content": "Keep checking"},
+            {
+                "role": "assistant",
+                "content": "",
+                "_recovered_from_run_journal": True,
+                "_recovered_stream_id": stream_id,
+            },
+            {
+                "role": "assistant",
+                "content": "",
+                "reasoning": "Inspect the follow-up boundary.",
+                "_recovered_from_run_journal": True,
+                "_recovered_stream_id": stream_id,
+            },
+        ],
+        context_messages=[{"role": "user", "content": "Keep checking"}],
+    )
+    append_run_event(
+        session_id,
+        stream_id,
+        "tool",
+        {"name": "terminal", "preview": "inspect the first boundary"},
+    )
+
+    assert _append_journaled_partial_output(
+        session, stream_id, dedupe_existing=True,
+    ) is True
+
+    assert session.tool_calls[0]["assistant_msg_idx"] == 1
+    assert session.messages[1].get("reasoning") is None
+    assert session.messages[2].get("reasoning") == "Inspect the follow-up boundary."
+
+
+def test_identical_reasoning_segments_around_tool_remain_distinct():
+    session_id = "issue3929_identical_reasoning_segments"
+    stream_id = "stream_identical_reasoning_segments"
+    repeated_reasoning = "Checking the same condition again."
+    session = Session(
+        session_id=session_id,
+        title="Repeated Thinking segments",
+        messages=[{"role": "user", "content": "Check twice"}],
+        context_messages=[{"role": "user", "content": "Check twice"}],
+    )
+    append_run_event(
+        session_id,
+        stream_id,
+        "reasoning",
+        {"text": repeated_reasoning},
+    )
+    append_run_event(
+        session_id,
+        stream_id,
+        "tool",
+        {"name": "terminal", "preview": "first check"},
+    )
+    append_run_event(
+        session_id,
+        stream_id,
+        "tool_complete",
+        {"name": "terminal", "is_error": False},
+    )
+    append_run_event(
+        session_id,
+        stream_id,
+        "reasoning",
+        {"text": repeated_reasoning},
+    )
+
+    assert _append_journaled_partial_output(
+        session, stream_id, dedupe_existing=True,
+    ) is True
+
+    reasoning_rows = [
+        (idx, message)
+        for idx, message in enumerate(session.messages)
+        if message.get("reasoning") == repeated_reasoning
+    ]
+    assert len(reasoning_rows) == 2
+    assert reasoning_rows[0][0] != reasoning_rows[1][0]
+    assert session.tool_calls[0]["assistant_msg_idx"] == reasoning_rows[0][0]
+
+
+def test_identical_content_segments_claim_distinct_rows_on_replay():
+    session_id = "issue3929_identical_content_segments"
+    stream_id = "stream_identical_content_segments"
+    repeated_content = "The same sufficiently long process update appears twice."
+    session = Session(
+        session_id=session_id,
+        title="Repeated process segments",
+        messages=[{"role": "user", "content": "Check twice"}],
+        context_messages=[{"role": "user", "content": "Check twice"}],
+    )
+    append_run_event(
+        session_id,
+        stream_id,
+        "reasoning",
+        {"text": "First Thinking segment."},
+    )
+    append_run_event(
+        session_id,
+        stream_id,
+        "token",
+        {"text": repeated_content},
+    )
+    append_run_event(
+        session_id,
+        stream_id,
+        "tool",
+        {"name": "terminal", "preview": "separate the segments"},
+    )
+    append_run_event(
+        session_id,
+        stream_id,
+        "tool_complete",
+        {"name": "terminal", "is_error": False},
+    )
+    append_run_event(
+        session_id,
+        stream_id,
+        "reasoning",
+        {"text": "Second Thinking segment."},
+    )
+    append_run_event(
+        session_id,
+        stream_id,
+        "token",
+        {"text": repeated_content},
+    )
+
+    assert _append_journaled_partial_output(
+        session, stream_id, dedupe_existing=True,
+    ) is True
+    assert _append_journaled_partial_output(
+        session, stream_id, dedupe_existing=True,
+    ) is False
+
+    matching_rows = [
+        message for message in session.messages
+        if message.get("content") == repeated_content
+    ]
+    assert [message.get("reasoning") for message in matching_rows] == [
+        "First Thinking segment.",
+        "Second Thinking segment.",
+    ]

tests/test_session_sidecar_repair.py

@@ -763,9 +763,11 @@ class TestNonEmptyMessagesPendingCleared:
         assert "partial output above was recovered" in error_msgs[0]["content"]
         assert "no agent output was recovered" not in error_msgs[0]["content"]
 
-    def test_journal_recovery_does_not_materialize_reasoning_only_events(self, hermes_home, monkeypatch):
-        """Run-journal repair must not turn hidden reasoning into visible chat
-        transcript content."""
+    def test_journal_recovery_restores_reasoning_only_as_display_metadata(
+        self, hermes_home, monkeypatch,
+    ):
+        """Run-journal repair keeps live Thinking without making it chat prose
+        or provider-facing history."""
         s = _make_session(messages=[{"role": "user", "content": "existing turn"}])
         s.pending_user_message = "Keep going"
         s.pending_started_at = time.time() - 120
@@ -789,16 +791,14 @@ class TestNonEmptyMessagesPendingCleared:
         assert result is True
         contents = [m.get("content", "") for m in s.messages]
         assert not any("private scratchpad text" in c for c in contents)
+        reasoning_msgs = [m for m in s.messages if m.get("reasoning")]
+        assert [m.get("reasoning") for m in reasoning_msgs] == ["private scratchpad text"]
         error_msgs = [m for m in s.messages if m.get("_error")]
         assert len(error_msgs) == 1
-        # Reasoning-only events do not count as visible output, so the marker
-        # arms the lazy-retry hook (stream id is known, journal may have more
-        # events appear on a later read). The legacy "no agent output was
-        # recovered" wording is now reserved for the no-stream-id case.
-        assert error_msgs[0].get("_pending_journal_recovery") is True
-        assert error_msgs[0].get("_journal_retry_stream_id") == "reasoning_only_stream"
-        assert "no agent output was recovered" not in error_msgs[0]["content"]
-        assert "Recovering the partial output" in error_msgs[0]["content"]
+        assert error_msgs[0].get("_pending_journal_recovery") is not True
+        assert "partial output above was recovered" in error_msgs[0]["content"]
+        provider_history = streaming._sanitize_messages_for_api(s.messages)
+        assert "private scratchpad text" not in json.dumps(provider_history)
 
     def test_journal_recovery_keeps_consecutive_tools_on_one_anchor(self, hermes_home, monkeypatch):
         """Consecutive journaled tools without an intervening visible update