stage #4758 (franksong2702): dedupe live progress reasoning echoes

CHANGELOG.md

@@ -3,6 +3,10 @@
 
 ## [Unreleased]
 
+### Fixed
+
+- **Compact Worklog no longer shows the same live progress sentence as both Process and Thinking.** Some reasoning-heavy runtimes emit a user-facing progress update first through the reasoning callback and then through `interim_assistant`; the Anchor-backed live Worklog now treats that as one visible progress row, strips the duplicated reasoning tail from live/durable state, and keeps journal replay from rebuilding the duplicate Thinking row.
+
 ## [v0.51.598] — 2026-06-23 — Release VE (preserve scroll across live worklog rebuilds)
 
 ### Fixed

api/routes.py

@@ -2622,6 +2622,21 @@ def _run_journal_live_snapshot(stream_id: str | None) -> dict | None:
             call["activitySegmentSeq"] = current_activity_burst_id
         tool_calls.append(call)
 
+    def reasoning_echo_tail_matches(text: str) -> bool:
+        candidate = _compact_for_echo_compare(text)
+        if not candidate:
+            return False
+        return _compact_for_echo_compare(reasoning_text).endswith(candidate)
+
+    def strip_reasoning_echo_tail(text: str) -> bool:
+        nonlocal reasoning_text, reasoning_first_tool_count
+        next_reasoning, did_remove = _strip_compact_echo_suffix(reasoning_text, text)
+        if did_remove:
+            reasoning_text = next_reasoning
+            if not _compact_for_echo_compare(reasoning_text):
+                reasoning_first_tool_count = None
+        return did_remove
+
     for event in events:
         event_name = str(event.get("event") or event.get("type") or "")
         payload = event.get("payload") if isinstance(event.get("payload"), dict) else {}
@@ -2641,6 +2656,8 @@ def _run_journal_live_snapshot(stream_id: str | None) -> dict | None:
         if event_name == "interim_assistant":
             visible = str(payload.get("text") or "").strip()
             if visible:
+                if payload.get("reasoning_echo") or reasoning_echo_tail_matches(visible):
+                    strip_reasoning_echo_tail(visible)
                 if payload.get("already_streamed"):
                     if not assistant_text:
                         assistant_text = visible
@@ -6631,6 +6648,8 @@ from api.streaming import (
     cancel_stream,
     _materialize_pending_user_turn_before_error,
     generate_session_title_for_session,
+    _compact_for_echo_compare,
+    _strip_compact_echo_suffix,
 )
 from api.gateway_chat import _run_gateway_chat_streaming, webui_gateway_chat_enabled
 from api.run_journal import (

api/streaming.py

@@ -79,6 +79,20 @@ def _compact_for_echo_compare(value: str) -> str:
     return re.sub(r'\s+', '', str(value or ''))
 
 
+def _strip_compact_echo_suffix(value: str, suffix: str, *, search_window: int = 4096) -> tuple[str, bool]:
+    """Remove ``suffix`` from ``value`` when they match after whitespace folding."""
+    raw = str(value or '')
+    candidate = _compact_for_echo_compare(suffix)
+    if not raw or not candidate:
+        return raw, False
+    tail = raw[-max(len(str(suffix or '')) * 3, search_window):]
+    offset = len(raw) - len(tail)
+    for idx in range(len(tail) + 1):
+        if _compact_for_echo_compare(tail[idx:]) == candidate:
+            return raw[: offset + idx].rstrip(), True
+    return raw, False
+
+
 # Global lock for os.environ writes. Per-session locks (_agent_lock) prevent
 # concurrent runs of the SAME session, but two DIFFERENT sessions can still
 # interleave their os.environ writes. This global lock serializes the env
@@ -6375,6 +6389,38 @@ def _run_agent_streaming(
                 )
                 return bool(visible_tail and visible_tail.endswith(candidate))
 
+            def _strip_reasoning_output_echo(text: str) -> bool:
+                nonlocal _reasoning_segments
+                removed = False
+                if stream_id in STREAM_REASONING_TEXT:
+                    next_text, did_remove = _strip_compact_echo_suffix(
+                        STREAM_REASONING_TEXT.get(stream_id, ''),
+                        text,
+                    )
+                    if did_remove:
+                        STREAM_REASONING_TEXT[stream_id] = next_text
+                        removed = True
+                next_buffer, did_remove_buffer = _strip_compact_echo_suffix(_reasoning_buffer[0], text)
+                if did_remove_buffer:
+                    _reasoning_buffer[0] = next_buffer
+                    removed = True
+                for idx in (_current_reasoning_idx, _current_reasoning_idx - 1):
+                    if idx not in _reasoning_segments:
+                        continue
+                    next_segment, did_remove_segment = _strip_compact_echo_suffix(
+                        _reasoning_segments.get(idx, ''),
+                        text,
+                    )
+                    if not did_remove_segment:
+                        continue
+                    if next_segment:
+                        _reasoning_segments[idx] = next_segment
+                    else:
+                        _reasoning_segments.pop(idx, None)
+                    removed = True
+                    break
+                return removed
+
             def on_token(text):
                 nonlocal _token_sent
                 if text is None:
@@ -6448,11 +6494,15 @@ def _run_agent_streaming(
                 visible = str(text).strip()
                 if not visible:
                     return
+                reasoning_echo = _strip_reasoning_output_echo(visible)
                 already_streamed = bool(cb_kwargs.get('already_streamed', False)) or _is_visible_output_echo(visible)
-                put('interim_assistant', {
+                payload = {
                     'text': visible,
                     'already_streamed': already_streamed,
-                })
+                }
+                if reasoning_echo:
+                    payload['reasoning_echo'] = True
+                put('interim_assistant', payload)
 
             # Pre-initialise the activity counter here so on_tool (which
             # closes over it) never captures an unbound name even if this

static/messages.js

@@ -2791,6 +2791,65 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){
     },null);
     return _findAnchorActivityEventByLocalId(localId,'reasoning');
   }
+  function _compactVisibleEchoText(value){
+    return String(value||'').replace(/\s+/g,'');
+  }
+  function _stripCompactEchoSuffix(value, suffix){
+    const raw=String(value||'');
+    const candidate=_compactVisibleEchoText(suffix);
+    if(!raw||!candidate) return {text:raw,removed:false};
+    const windowSize=Math.max(String(suffix||'').length*3,4096);
+    const offset=Math.max(0,raw.length-windowSize);
+    const tail=raw.slice(offset);
+    for(let idx=0;idx<=tail.length;idx+=1){
+      if(_compactVisibleEchoText(tail.slice(idx))===candidate){
+        return {text:raw.slice(0,offset+idx).trimEnd(),removed:true};
+      }
+    }
+    return {text:raw,removed:false};
+  }
+  function _stripAnchorReasoningEcho(visible){
+    const events=_anchorActivityEvents();
+    if(!events||!visible) return false;
+    for(let i=events.length-1;i>=0;i-=1){
+      const event=events[i];
+      if(!event||event.source_event_type!=='reasoning') continue;
+      const payload=(event.payload&&typeof event.payload==='object')?event.payload:{};
+      const rawText=String(payload.text||payload.reasoning||payload.thinking||'');
+      const stripped=_stripCompactEchoSuffix(rawText, visible);
+      if(!stripped.removed) continue;
+      const nextText=String(stripped.text||'').trim();
+      if(nextText){
+        _replaceAnchorActivityEventByLocalId(event.local_id,'reasoning',{
+          payload:{text:nextText},
+        });
+      }else{
+        events.splice(i,1);
+      }
+      _renderAnchorLiveScene();
+      return true;
+    }
+    return false;
+  }
+  function _stripLiveReasoningEcho(visible){
+    let removed=false;
+    const durable=_stripCompactEchoSuffix(reasoningText, visible);
+    if(durable.removed){
+      reasoningText=durable.text;
+      removed=true;
+    }
+    const live=_stripCompactEchoSuffix(liveReasoningText, visible);
+    if(live.removed){
+      liveReasoningText=live.text;
+      removed=true;
+    }
+    const anchorRemoved=_stripAnchorReasoningEcho(visible);
+    if(removed) syncInflightAssistantMessage();
+    if((removed||anchorRemoved)&&!String(liveReasoningText||'').trim()&&typeof removeThinking==='function'){
+      removeThinking();
+    }
+    return removed||anchorRemoved;
+  }
   function _flushReasoningToAnchor(){
     if(_anchorReasoningFlushed||!reasoningText) return;
     _anchorReasoningFlushed=true;
@@ -3777,9 +3836,11 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){
       const d=JSON.parse(e.data);
       const visible=String(d&&d.text?d.text:'').trim();
       const alreadyStreamed=!!(d&&d.already_streamed);
+      const reasoningEcho=!!(d&&d.reasoning_echo);
       if(!visible){
         return;
       }
+      if(reasoningEcho) _stripLiveReasoningEcho(visible);
       liveReasoningText='';
       if(alreadyStreamed){
         if(!S.session||S.session.session_id!==activeSid){
@@ -3799,7 +3860,6 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){
         _resetAssistantSegment();
         return;
       }
-      _applyToAnchor('interim_assistant',d,e);
       assistantText += assistantText ? `\n\n${visible}` : visible;
       visibleInterimSnippets.push(visible);
       syncInflightAssistantMessage();
@@ -3814,6 +3874,7 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){
       _flushPendingSegmentRender({force:true});
       if(typeof finalizeThinkingCard==='function') finalizeThinkingCard();
       if(typeof closeCurrentLiveActivityGroup==='function') closeCurrentLiveActivityGroup();
+      _applyToAnchor('interim_assistant',d,e);
       // Collapse old interim notes once more than INTERIM_COLLAPSE_THRESHOLD accumulate.
       const INTERIM_COLLAPSE_THRESHOLD=3;
       if(visibleInterimSnippets.length>INTERIM_COLLAPSE_THRESHOLD&&assistantRow){

tests/test_anchor_scene_persistence.py

@@ -945,6 +945,51 @@ def test_runtime_journal_snapshot_includes_live_anchor_activity_scene(monkeypatc
     assert rows[3]["status"] == "running"
 
 
+def test_runtime_journal_snapshot_dedupes_reasoning_interim_progress_echo(monkeypatch):
+    from api import routes
+
+    stream_id = "stream-live-reasoning-interim-echo"
+    progress = "我先检查当前仓库状态，然后定位重复渲染路径。"
+    events = [
+        {
+            "event": "reasoning",
+            "seq": 1,
+            "event_id": f"{stream_id}:1",
+            "created_at": 1.0,
+            "payload": {"text": progress},
+        },
+        {
+            "event": "interim_assistant",
+            "seq": 2,
+            "event_id": f"{stream_id}:2",
+            "created_at": 2.0,
+            "payload": {"text": progress},
+        },
+    ]
+    monkeypatch.setattr(
+        routes,
+        "find_run_summary",
+        lambda sid: {
+            "session_id": "session-live-reasoning-interim-echo",
+            "last_seq": 2,
+            "last_event_id": f"{stream_id}:2",
+        },
+    )
+    monkeypatch.setattr(
+        routes,
+        "read_run_events",
+        lambda session_id, run_id: {"events": events},
+    )
+
+    snapshot = routes._run_journal_live_snapshot(stream_id)
+    rows = snapshot["anchor_activity_scene"]["activity_rows"]
+
+    assert snapshot["last_assistant_text"] == progress
+    assert snapshot["last_reasoning_text"] == ""
+    assert [row["role"] for row in rows] == ["prose"]
+    assert rows[0]["text"] == progress
+
+
 def test_runtime_journal_snapshot_has_running_anchor_row_before_first_token(monkeypatch):
     from api import routes
 

tests/test_issue_progress_echo_dedupe.py

@@ -134,11 +134,11 @@ def test_visible_progress_token_reasoning_and_interim_are_deduped(cleanup_test_s
         "credential_pool": None,
     }
     runtime_payload["api_" + "key"] = "***"
-    setattr(fake_runtime_module, "resolve_runtime_provider", mock.Mock(return_value=runtime_payload))
+    fake_runtime_module.resolve_runtime_provider = mock.Mock(return_value=runtime_payload)
     fake_hermes_cli = types.ModuleType("hermes_cli")
-    setattr(fake_hermes_cli, "runtime_provider", fake_runtime_module)
+    fake_hermes_cli.runtime_provider = fake_runtime_module
     fake_hermes_state = types.ModuleType("hermes_state")
-    setattr(fake_hermes_state, "SessionDB", mock.Mock(return_value=None))
+    fake_hermes_state.SessionDB = mock.Mock(return_value=None)
     injected = {
         "hermes_cli": fake_hermes_cli,
         "hermes_cli.runtime_provider": fake_runtime_module,
@@ -175,3 +175,171 @@ def test_visible_progress_token_reasoning_and_interim_are_deduped(cleanup_test_s
     assert not [payload for event, payload in events if event == "reasoning" and payload.get("text") == progress]
     interim = [payload for event, payload in events if event == "interim_assistant"]
     assert interim == [{"text": progress, "already_streamed": True}]
+
+
+def test_reasoning_then_interim_progress_marks_reasoning_echo(cleanup_test_sessions):
+    """A progress sentence mirrored as reasoning first must become prose, not Thinking.
+
+    Some reasoning-heavy runtimes emit the same user-facing status sentence first
+    through the reasoning callback and later through interim_assistant. The first
+    reasoning SSE may already be in the browser/journal, so the bridge must mark
+    the interim event and strip the durable reasoning tail before settlement.
+    """
+    import api.streaming as streaming
+
+    progress = "我先检查当前仓库状态，然后定位重复渲染路径。"
+
+    class FakeSession:
+        def __init__(self):
+            self.session_id = "issue_reasoning_interim_echo"
+            self.title = "Reasoning interim echo"
+            self.workspace = "/tmp"
+            self.model = "gpt-test"
+            self.model_provider = None
+            self.profile = None
+            self.personality = None
+            self.messages = []
+            self.context_messages = []
+            self.input_tokens = 0
+            self.output_tokens = 0
+            self.estimated_cost = 0
+            self.cache_read_tokens = 0
+            self.cache_write_tokens = 0
+            self.tool_calls = []
+            self.gateway_routing = None
+            self.gateway_routing_history = []
+            self.active_stream_id = ""
+            self.pending_user_message = None
+            self.pending_attachments = []
+            self.pending_started_at = None
+            self.context_length = 0
+            self.threshold_tokens = 0
+            self.last_prompt_tokens = 0
+            self.llm_title_generated = True
+
+        def save(self, *args, **kwargs):
+            pass
+
+        def compact(self):
+            return {
+                "session_id": self.session_id,
+                "title": self.title,
+                "workspace": self.workspace,
+                "model": self.model,
+                "created_at": 0,
+                "updated_at": 0,
+                "pinned": False,
+                "archived": False,
+                "project_id": None,
+                "profile": self.profile,
+                "input_tokens": self.input_tokens,
+                "output_tokens": self.output_tokens,
+                "estimated_cost": self.estimated_cost,
+                "cache_read_tokens": self.cache_read_tokens,
+                "cache_write_tokens": self.cache_write_tokens,
+                "personality": self.personality,
+            }
+
+    class ReasoningThenInterimAgent:
+        def __init__(
+            self,
+            model=None,
+            provider=None,
+            base_url=None,
+            platform=None,
+            quiet_mode=False,
+            enabled_toolsets=None,
+            fallback_model=None,
+            session_id=None,
+            session_db=None,
+            prefill_messages=None,
+            stream_delta_callback=None,
+            reasoning_callback=None,
+            tool_progress_callback=None,
+            clarify_callback=None,
+            interim_assistant_callback=None,
+            **_kwargs,
+        ):
+            self.reasoning_callback = cast(Callable[[str], None], reasoning_callback)
+            self.interim_assistant_callback = cast(Callable[[str], None], interim_assistant_callback)
+            self.context_compressor = None
+            self.session_prompt_tokens = 0
+            self.session_completion_tokens = 0
+            self.session_estimated_cost_usd = 0
+            self.session_cache_read_tokens = 0
+            self.session_cache_write_tokens = 0
+            self.reasoning_config = None
+            self.ephemeral_system_prompt = None
+            self._last_error = None
+
+        def run_conversation(self, **kwargs):
+            self.reasoning_callback(progress)
+            self.interim_assistant_callback(progress)
+            history = kwargs.get("conversation_history", [])
+            return {"messages": history + [
+                {"role": "user", "content": kwargs["persist_user_message"]},
+                {"role": "assistant", "content": progress},
+            ]}
+
+        def interrupt(self, _message):
+            pass
+
+    fake_session = FakeSession()
+    fake_stream_id = "stream_issue_reasoning_interim_echo"
+    fake_session.active_stream_id = fake_stream_id
+    fake_queue = queue.Queue()
+    fake_runtime_module = types.ModuleType("hermes_cli.runtime_provider")
+    runtime_payload = {
+        "provider": "openai",
+        "base_url": None,
+        "api_mode": "chat_completions",
+        "command": None,
+        "args": [],
+        "credential_pool": None,
+    }
+    runtime_payload["api_" + "key"] = "***"
+    fake_runtime_module.resolve_runtime_provider = mock.Mock(return_value=runtime_payload)
+    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,
+    }
+    saved = {k: sys.modules.get(k, _MISSING) for k in injected}
+    sys.modules.update(injected)
+    try:
+        with mock.patch.object(streaming, "get_session", return_value=fake_session), \
+             mock.patch.object(streaming, "_get_ai_agent", return_value=ReasoningThenInterimAgent), \
+             mock.patch.object(streaming, "resolve_model_provider", return_value=("gpt-test", "openai", None)), \
+             mock.patch("api.config.get_config", return_value={}), \
+             mock.patch("api.config._resolve_cli_toolsets", return_value=[]):
+            streaming.STREAMS[fake_stream_id] = fake_queue
+            streaming._run_agent_streaming(
+                session_id=fake_session.session_id,
+                msg_text="scan",
+                model="gpt-test",
+                workspace="/tmp",
+                stream_id=fake_stream_id,
+            )
+    finally:
+        streaming.STREAMS.pop(fake_stream_id, None)
+        for k, prev in saved.items():
+            if prev is _MISSING:
+                sys.modules.pop(k, None)
+            else:
+                sys.modules[k] = cast(types.ModuleType, prev)
+
+    events = list(fake_queue.queue)
+    interim = [payload for event, payload in events if event == "interim_assistant"]
+    assert interim == [{
+        "text": progress,
+        "already_streamed": False,
+        "reasoning_echo": True,
+    }]
+    done_payloads = [payload for event, payload in events if event == "done"]
+    assert done_payloads, "run should settle"
+    final_messages = done_payloads[-1]["session"]["messages"]
+    assert not any(message.get("reasoning") == progress for message in final_messages)

tests/test_live_anchor_progress_echo.py

@@ -0,0 +1,52 @@
+"""Regression guards for Anchor-owned live progress echo cleanup."""
+
+from __future__ import annotations
+
+import pathlib
+import re
+
+
+REPO = pathlib.Path(__file__).resolve().parent.parent
+MESSAGES = (REPO / "static" / "messages.js").read_text(encoding="utf-8")
+UI = (REPO / "static" / "ui.js").read_text(encoding="utf-8")
+
+
+def _interim_listener_body() -> str:
+    match = re.search(
+        r"source\.addEventListener\('interim_assistant'\s*,\s*(?:e|ev)\s*=>\s*\{(.*?)\n\s*\}\);",
+        MESSAGES,
+        re.DOTALL,
+    )
+    assert match, "interim_assistant listener not found"
+    return match.group(1)
+
+
+def test_interim_reasoning_echo_cleans_live_and_anchor_thinking():
+    body = _interim_listener_body()
+
+    assert "const reasoningEcho=!!(d&&d.reasoning_echo);" in body
+    assert "if(reasoningEcho) _stripLiveReasoningEcho(visible);" in body
+    assert "function _stripAnchorReasoningEcho(visible)" in MESSAGES
+    assert "events.splice(i,1);" in MESSAGES
+    assert "reasoningText=durable.text;" in MESSAGES
+    assert "liveReasoningText=live.text;" in MESSAGES
+
+
+def test_interim_anchor_render_runs_after_legacy_segment_flush():
+    body = _interim_listener_body()
+
+    flush_idx = body.index("_flushPendingSegmentRender({force:true});")
+    anchor_idx = body.index("_applyToAnchor('interim_assistant',d,e);")
+    assert flush_idx < anchor_idx, (
+        "Anchor live scene must render after the legacy interim segment is flushed, "
+        "so renderLiveAnchorActivityScene can hide that source segment immediately."
+    )
+
+
+def test_live_anchor_scene_hides_legacy_live_assistant_sources():
+    start = UI.index("function renderLiveAnchorActivityScene")
+    body = UI[start : UI.index("function _renderLiveAnchorActivitySceneForStream", start)]
+
+    assert "blocks.querySelectorAll('[data-live-assistant=\"1\"]').forEach" in body
+    assert "assistant-segment-worklog-source" in body
+    assert "aria-hidden" in body