diff --git a/CHANGELOG.md b/CHANGELOG.md index cd5c3883..106d0891 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/api/routes.py b/api/routes.py index ddeeb0de..25796f0b 100644 --- a/api/routes.py +++ b/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 ( diff --git a/api/streaming.py b/api/streaming.py index d6b27d1c..57b51d05 100644 --- a/api/streaming.py +++ b/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 diff --git a/static/messages.js b/static/messages.js index 2c1a02d0..3888c762 100644 --- a/static/messages.js +++ b/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){ diff --git a/tests/test_anchor_scene_persistence.py b/tests/test_anchor_scene_persistence.py index 552ec64c..5b1a9fbf 100644 --- a/tests/test_anchor_scene_persistence.py +++ b/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 diff --git a/tests/test_issue_progress_echo_dedupe.py b/tests/test_issue_progress_echo_dedupe.py index 4b0e5109..b5243111 100644 --- a/tests/test_issue_progress_echo_dedupe.py +++ b/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) diff --git a/tests/test_live_anchor_progress_echo.py b/tests/test_live_anchor_progress_echo.py new file mode 100644 index 00000000..f20438d4 --- /dev/null +++ b/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