From de96c1d3849d25d552d77f2d17d342b01c0fa067 Mon Sep 17 00:00:00 2001 From: Frank Song Date: Mon, 20 Jul 2026 15:34:18 +0800 Subject: [PATCH] fix: harden terminal error worklog settlement --- api/gateway_chat.py | 13 ++----- api/routes.py | 20 +++++++++- api/streaming.py | 38 ++++++++++--------- static/messages.js | 18 ++++++++- tests/test_anchor_scene_persistence.py | 36 +++++++++++++++++- .../test_issue3929_error_preserves_partial.py | 18 +++++++++ ...test_live_to_final_anchor_visible_order.py | 9 ++++- 7 files changed, 119 insertions(+), 33 deletions(-) diff --git a/api/gateway_chat.py b/api/gateway_chat.py index 06053fcb..006ad390 100644 --- a/api/gateway_chat.py +++ b/api/gateway_chat.py @@ -558,6 +558,7 @@ def _settle_gateway_terminal_error(session_id, stream_id, workspace, model, mode _provider_error_payload, _session_payload_with_full_messages, _snapshot_and_append_partial_on_error, + _terminal_turn_duration, ) with _get_session_agent_lock(session_id): @@ -570,14 +571,7 @@ def _settle_gateway_terminal_error(session_id, stream_id, workspace, model, mode error_classification["type"], error_classification.get("hint", ""), ) - # Freeze turn duration before terminal cleanup clears pending_started_at (#6309) - _turn_duration_seconds = 0.0 - try: - _pending_ts = getattr(session, 'pending_started_at', None) - if _pending_ts: - _turn_duration_seconds = max(0.0, time.time() - float(_pending_ts)) - except Exception: - pass + turn_duration = _terminal_turn_duration(session) _materialize_pending_user_turn_before_error(session) session.active_stream_id = None session.pending_user_message = None @@ -596,8 +590,9 @@ def _settle_gateway_terminal_error(session_id, stream_id, workspace, model, mode ) + (f"\n\n*{error_payload['hint']}*" if error_payload.get("hint") else ""), "timestamp": int(time.time()), "_error": True, - "_turnDuration": round(_turn_duration_seconds, 3), } + if turn_duration is not None: + error_message["_turnDuration"] = turn_duration if error_payload.get("details"): error_message["provider_details"] = error_payload["details"] if not isinstance(session.messages, list): diff --git a/api/routes.py b/api/routes.py index 97e8044a..8a96dd4d 100644 --- a/api/routes.py +++ b/api/routes.py @@ -4567,7 +4567,17 @@ def _anchor_scene_row_has_live_identity(row) -> bool: values = [row.get("row_id"), row.get("local_id"), row.get("event_id")] identity = row.get("identity") if isinstance(row.get("identity"), dict) else {} values.extend([identity.get("local_id"), identity.get("event_id")]) - return any(str(value or "").startswith("live-") for value in values) + if any(str(value or "").startswith("live-") for value in values): + return True + group = row.get("group") if isinstance(row.get("group"), dict) else {} + has_stream_owner = bool( + row.get("stream_id") + or row.get("run_id") + or identity.get("stream_id") + or identity.get("run_id") + ) + has_assistant_message_index = group.get("assistant_msg_idx") is not None + return has_stream_owner and not has_assistant_message_index def _anchor_scene_settle_live_running_row(row, *, has_settled_thinking: bool): @@ -4584,6 +4594,14 @@ def _anchor_scene_settle_live_running_row(row, *, has_settled_thinking: bool): return None next_row = copy.deepcopy(row) next_row["status"] = "completed" + payload = next_row.get("payload") + if isinstance(payload, dict): + payload["status"] = "completed" + if role == "tool": + payload["done"] = True + tool = next_row.get("tool") + if role == "tool" and isinstance(tool, dict): + tool["done"] = True return next_row diff --git a/api/streaming.py b/api/streaming.py index 115b6fe5..8b294684 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -7,6 +7,7 @@ import contextlib import contextvars import json import logging +import math import mimetypes import os import queue @@ -6501,6 +6502,19 @@ def _materialize_pending_user_turn_before_error(session) -> bool: return True +def _terminal_turn_duration(session, *, now: float | None = None) -> float | None: + """Freeze a valid turn timer before terminal cleanup clears its origin.""" + started_at = getattr(session, 'pending_started_at', None) + try: + started = float(started_at) + ended = float(time.time() if now is None else now) + except (TypeError, ValueError): + return None + if not math.isfinite(started) or not math.isfinite(ended) or started <= 0: + return None + return round(max(0.0, ended - started), 3) + + def _build_partial_message(content_text, reasoning_text, tool_calls) -> dict | None: """Build a _partial assistant message from raw streaming buffers. @@ -9640,14 +9654,7 @@ def _run_agent_streaming( + 'send a message, switch the model/provider, or fix the credentials.' ) _error_payload['hint'] = _err_hint - # Freeze turn duration before terminal cleanup clears pending_started_at (#6309) - _turn_duration_seconds = 0.0 - try: - _pending_ts = getattr(s, 'pending_started_at', None) - if _pending_ts: - _turn_duration_seconds = max(0.0, time.time() - float(_pending_ts)) - except Exception: - pass + _turn_duration = _terminal_turn_duration(s) _materialize_pending_user_turn_before_error(s) s.active_stream_id = None s.pending_user_message = None @@ -9667,8 +9674,9 @@ def _run_agent_streaming( 'content': _error_content, 'timestamp': int(time.time()), '_error': True, - '_turnDuration': round(_turn_duration_seconds, 3), } + if _turn_duration is not None: + _error_message['_turnDuration'] = _turn_duration if _err_type == 'compression_exhausted': _recovery = stamp_compression_exhausted_recovery( s, @@ -10873,14 +10881,7 @@ def _run_agent_streaming( + 'send a message, switch the model/provider, or fix the credentials.' ) _error_payload['hint'] = _exc_hint - # Freeze turn duration before terminal cleanup clears pending_started_at (#6309) - _turn_duration_seconds = 0.0 - try: - _pending_ts = getattr(s, 'pending_started_at', None) - if _pending_ts: - _turn_duration_seconds = max(0.0, time.time() - float(_pending_ts)) - except Exception: - pass + _turn_duration = _terminal_turn_duration(s) _materialize_pending_user_turn_before_error(s) s.active_stream_id = None s.pending_user_message = None @@ -10896,8 +10897,9 @@ def _run_agent_streaming( 'content': f'**{_exc_label}:** {_error_payload.get("message") or err_str}' + (f'\n\n*{_exc_hint}*' if _exc_hint else ''), 'timestamp': int(time.time()), '_error': True, - '_turnDuration': round(_turn_duration_seconds, 3), } + if _turn_duration is not None: + _error_message['_turnDuration'] = _turn_duration if _exc_type == 'compression_exhausted': _recovery = stamp_compression_exhausted_recovery( s, diff --git a/static/messages.js b/static/messages.js index 9f1532f1..514c5899 100644 --- a/static/messages.js +++ b/static/messages.js @@ -3467,7 +3467,13 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ if(!row||typeof row!=='object') return false; const identity=row.identity&&typeof row.identity==='object'?row.identity:{}; const values=[row.row_id,row.local_id,row.event_id,identity.local_id,identity.event_id]; - return values.some(value=>String(value||'').startsWith('live-')); + if(values.some(value=>String(value||'').startsWith('live-'))) return true; + // Provider tool-call IDs do not use the live prefix. A stream owner without + // a settled assistant index still identifies a projected live row. + const group=row.group&&typeof row.group==='object'?row.group:{}; + const hasStreamOwner=!!(row.stream_id||row.run_id||identity.stream_id||identity.run_id); + const hasAssistantMessageIndex=group.assistant_msg_idx!==undefined&&group.assistant_msg_idx!==null; + return hasStreamOwner&&!hasAssistantMessageIndex; } function _anchorSceneMessageRowsHaveThinking(messageRows){ if(!(messageRows instanceof Map)) return false; @@ -3482,7 +3488,15 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ if(String(row.status||'').toLowerCase()!=='running') return row; if(!_anchorSceneRowHasLiveIdentity(row)) return row; if(row.role==='thinking'&&hasSettledThinking) return null; - return {...row,status:'completed'}; + const sealed={...row,status:'completed'}; + if(row.payload&&typeof row.payload==='object'){ + sealed.payload={...row.payload,status:'completed'}; + if(row.role==='tool') sealed.payload.done=true; + } + if(row.role==='tool'&&row.tool&&typeof row.tool==='object'){ + sealed.tool={...row.tool,done:true}; + } + return sealed; } function _anchorSceneRowLooksLikeFinalAnswer(rowTextKey, finalKey){ if(!rowTextKey||!finalKey) return false; diff --git a/tests/test_anchor_scene_persistence.py b/tests/test_anchor_scene_persistence.py index a9787092..fcd2a28f 100644 --- a/tests/test_anchor_scene_persistence.py +++ b/tests/test_anchor_scene_persistence.py @@ -2068,14 +2068,23 @@ def test_anchor_scene_hydration_seals_unmatched_live_running_activity_rows(): "text": "only live prose", }, { - "row_id": "live-tool:stream-1:call-1", - "local_id": "live-tool:stream-1:call-1", + "row_id": "tool:call-1:0", + "local_id": "call-1", "role": "tool", "kind": "tool_started", "source_event_type": "tool", "status": "running", + "run_id": "run-1", + "stream_id": "stream-1", + "identity": { + "local_id": "call-1", + "run_id": "run-1", + "stream_id": "stream-1", + }, + "group": {"group_key": "segment:1", "activity_segment_seq": 1}, "tool_call_id": "call-1", "tool": {"id": "call-1", "name": "terminal", "done": False}, + "payload": {"tid": "call-1", "status": "running", "done": False}, }, {"row_id": "done", "role": "terminal", "kind": "terminal_status", "source_event_type": "done"}, ], @@ -2089,6 +2098,29 @@ def test_anchor_scene_hydration_seals_unmatched_live_running_activity_rows(): activity_rows = [row for row in rows if row.get("role") in {"thinking", "prose", "tool"}] assert [row.get("status") for row in activity_rows] == ["completed", "completed", "completed"] assert [row.get("role") for row in activity_rows] == ["thinking", "prose", "tool"] + tool_row = activity_rows[-1] + assert tool_row["tool"]["done"] is True + assert tool_row["payload"]["status"] == "completed" + assert tool_row["payload"]["done"] is True + + +def test_anchor_scene_settlement_does_not_reclassify_transcript_owned_tool_row(): + from api import routes + + row = { + "row_id": "hydrated:stream-1:tool:call-1", + "local_id": "call-1", + "role": "tool", + "status": "running", + "stream_id": "stream-1", + "group": {"assistant_msg_idx": 1}, + "tool": {"id": "call-1", "done": False}, + "payload": {"tid": "call-1", "status": "running", "done": False}, + } + + assert routes._anchor_scene_settle_live_running_row(row, has_settled_thinking=False) is row + assert row["tool"]["done"] is False + assert row["payload"]["done"] is False def test_runtime_journal_anchor_scene_matches_settled_hydrated_visible_semantics(tmp_path, monkeypatch): diff --git a/tests/test_issue3929_error_preserves_partial.py b/tests/test_issue3929_error_preserves_partial.py index 6584cf24..520bb54d 100644 --- a/tests/test_issue3929_error_preserves_partial.py +++ b/tests/test_issue3929_error_preserves_partial.py @@ -4,6 +4,7 @@ from __future__ import annotations import queue import sys +import time import types import pytest from unittest import mock @@ -14,6 +15,19 @@ import api.streaming as streaming from api.models import Session +@pytest.mark.parametrize("started_at", [None, 0, -1, "invalid", float("nan"), float("inf")]) +def test_terminal_turn_duration_omits_invalid_origin(started_at): + session = types.SimpleNamespace(pending_started_at=started_at) + + assert streaming._terminal_turn_duration(session, now=100.0) is None + + +def test_terminal_turn_duration_freezes_valid_origin(): + session = types.SimpleNamespace(pending_started_at=88.7654) + + assert streaming._terminal_turn_duration(session, now=100.0) == 11.235 + + @pytest.fixture(autouse=True) def _isolate_session_dir(tmp_path, monkeypatch): """Redirect SESSION_DIR / SESSION_INDEX_FILE to an isolated temp dir.""" @@ -111,6 +125,7 @@ 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.pending_started_at = time.time() - 12 fake_session.active_stream_id = "test_stream_silent" fake_session.save() models.SESSIONS["test_sess_silent"] = fake_session @@ -161,12 +176,14 @@ def test_silent_failure_preserves_partials(tmp_path): err_msg = saved.messages[-1] assert err_msg.get("_error") is True + assert err_msg.get("_turnDuration", 0) >= 12 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.pending_started_at = time.time() - 9 fake_session.active_stream_id = "test_stream_exc" fake_session.save() models.SESSIONS["test_sess_exc"] = fake_session @@ -205,6 +222,7 @@ def test_exception_preserves_partials(tmp_path): err_msg = saved.messages[-1] assert err_msg.get("_error") is True assert "Fake provider crash!" in err_msg.get("content", "") + assert err_msg.get("_turnDuration", 0) >= 9 def test_empty_partials_do_not_create_spurious_messages(tmp_path): diff --git a/tests/test_live_to_final_anchor_visible_order.py b/tests/test_live_to_final_anchor_visible_order.py index 239bd253..f353be22 100644 --- a/tests/test_live_to_final_anchor_visible_order.py +++ b/tests/test_live_to_final_anchor_visible_order.py @@ -948,9 +948,16 @@ def test_settled_anchor_scene_does_not_persist_running_live_activity_rows(): assert "const hasSettledThinking=_anchorSceneMessageRowsHaveThinking(messageRows);" in complete assert "row=_anchorSceneSettleLiveRunningRow(row,hasSettledThinking);" in complete assert "String(value||'').startsWith('live-')" in live_identity + assert "const hasStreamOwner=!!(row.stream_id||row.run_id||identity.stream_id||identity.run_id);" in live_identity + assert "const hasAssistantMessageIndex=group.assistant_msg_idx!==undefined&&group.assistant_msg_idx!==null;" in live_identity + assert "return hasStreamOwner&&!hasAssistantMessageIndex;" in live_identity assert "String(row.status||'').toLowerCase()!=='running'" in settle_live assert "if(row.role==='thinking'&&hasSettledThinking) return null;" in settle_live - assert "return {...row,status:'completed'};" in settle_live + assert "const sealed={...row,status:'completed'};" in settle_live + assert "sealed.payload={...row.payload,status:'completed'};" in settle_live + assert "if(row.role==='tool') sealed.payload.done=true;" in settle_live + assert "sealed.tool={...row.tool,done:true};" in settle_live + assert "return sealed;" in settle_live def test_settled_anchor_scene_separates_final_answer_from_activity_rows():