Merge pull request #6312 from franksong2702/lifecycle-terminal-error-gate

fix: harden terminal-error Worklog duration + settlement (#6309)

Co-authored-by: Frank Song <franksong2702@gmail.com>

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):

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
 
 

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,22 @@ 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, OverflowError):
+        return None
+    if not math.isfinite(started) or not math.isfinite(ended) or started <= 0:
+        return None
+    elapsed = ended - started
+    if elapsed < 0:
+        return None
+    return round(elapsed, 3)
+
+
 def _build_partial_message(content_text, reasoning_text, tool_calls) -> dict | None:
     """Build a _partial assistant message from raw streaming buffers.
 
@@ -9640,14 +9657,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 +9677,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 +10884,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 +10900,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,

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;

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):

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"), 101.0, 10**400])
+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,65 @@ 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
+
+
+@pytest.mark.parametrize("started_at", [None, "future"])
+def test_result_error_omits_turn_duration_for_missing_or_future_origin(tmp_path, started_at):
+    """Result-error writeback must not manufacture durations from invalid timer origins."""
+    label = "future" if started_at == "future" else "missing"
+    fake_session = Session(session_id=f"test_sess_silent_{label}", title="Test Session")
+    fake_session.pending_user_message = "What is python?"
+    fake_session.pending_started_at = time.time() + 30 if started_at == "future" else started_at
+    fake_session.active_stream_id = f"test_stream_silent_{label}"
+    fake_session.save()
+    models.SESSIONS[fake_session.session_id] = fake_session
+
+    class SilentFailureAgent(MockAgent):
+        def run_conversation(self, **kwargs):
+            if self.stream_delta_callback:
+                self.stream_delta_callback("Python is a programming language.")
+            return {
+                "status": "error",
+                "error": "Silent failure details",
+                "messages": kwargs.get("conversation_history") or [],
+            }
+
+    fake_queue = queue.Queue()
+    streaming.STREAMS[fake_session.active_stream_id] = fake_queue
+    config.STREAM_PARTIAL_TEXT[fake_session.active_stream_id] = ""
+
+    with mock.patch.object(streaming, "get_session", return_value=fake_session), \
+         mock.patch.object(streaming, "_get_ai_agent", return_value=SilentFailureAgent), \
+         mock.patch.object(streaming, "resolve_model_provider", return_value=("test-model", "test-provider", None)), \
+         mock.patch("api.config.get_config", return_value={}), \
+         mock.patch("api.config._resolve_cli_toolsets", return_value=[]):
+
+        streaming._run_agent_streaming(
+            session_id=fake_session.session_id,
+            msg_text="What is python?",
+            model="test-model",
+            workspace=str(tmp_path),
+            stream_id=fake_session.active_stream_id,
+        )
+
+    saved = Session.load(fake_session.session_id)
+    assert saved is not None
+    assert saved.messages[-1].get("_error") is True
+    assert "_turnDuration" not in saved.messages[-1]
+
+    apperrors = [item[1] for item in list(fake_queue.queue) if item[0] == "apperror"]
+    assert apperrors
+    payload_error = apperrors[-1]["session"]["messages"][-1]
+    assert payload_error.get("_error") is True
+    assert "_turnDuration" not in payload_error
 
 
 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 +273,48 @@ 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
+
+
+@pytest.mark.parametrize("started_at", [None, "future"])
+def test_exception_error_omits_turn_duration_for_missing_or_future_origin(tmp_path, started_at):
+    """Exception writeback must not persist durations from invalid timer origins."""
+    label = "future" if started_at == "future" else "missing"
+    fake_session = Session(session_id=f"test_sess_exc_{label}", title="Test Session")
+    fake_session.pending_user_message = "Exception test"
+    fake_session.pending_started_at = time.time() + 30 if started_at == "future" else started_at
+    fake_session.active_stream_id = f"test_stream_exc_{label}"
+    fake_session.save()
+    models.SESSIONS[fake_session.session_id] = fake_session
+
+    class ExceptionAgent(MockAgent):
+        def run_conversation(self, **kwargs):
+            if self.stream_delta_callback:
+                self.stream_delta_callback("Stream before crash.")
+            raise RuntimeError("Fake provider crash!")
+
+    fake_queue = queue.Queue()
+    streaming.STREAMS[fake_session.active_stream_id] = fake_queue
+    config.STREAM_PARTIAL_TEXT[fake_session.active_stream_id] = ""
+
+    with mock.patch.object(streaming, "get_session", return_value=fake_session), \
+         mock.patch.object(streaming, "_get_ai_agent", return_value=ExceptionAgent), \
+         mock.patch.object(streaming, "resolve_model_provider", return_value=("test-model", "test-provider", None)), \
+         mock.patch("api.config.get_config", return_value={}), \
+         mock.patch("api.config._resolve_cli_toolsets", return_value=[]):
+
+        streaming._run_agent_streaming(
+            session_id=fake_session.session_id,
+            msg_text="Exception test",
+            model="test-model",
+            workspace=str(tmp_path),
+            stream_id=fake_session.active_stream_id,
+        )
+
+    saved = Session.load(fake_session.session_id)
+    assert saved is not None
+    assert saved.messages[-1].get("_error") is True
+    assert "_turnDuration" not in saved.messages[-1]
 
 
 def test_empty_partials_do_not_create_spurious_messages(tmp_path):

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():

tests/test_webui_gateway_chat_backend.py

@@ -4,6 +4,7 @@ from email.message import Message
 import json
 from pathlib import Path
 import re
+import time
 import urllib.error
 
 import api.gateway_chat as gateway_chat
@@ -519,6 +520,38 @@ def test_gateway_chat_worker_classifies_terminal_provider_error_without_text(tmp
     unknown_errors = [item[1] for item in unknown_events if item[0] == "apperror"]
     assert unknown_errors[-1]["type"] == "error"
     assert "Gateway provider failed" in unknown_errors[-1]["message"]
+    unknown_payload_error = unknown_errors[-1]["session"]["messages"][-1]
+    assert unknown_payload_error.get("_error") is True
+    assert "_turnDuration" not in unknown_payload_error
+
+    response_error[0] = error_text
+    future_stream_id = "stream-gateway-future-duration-terminal-error-test"
+    s = new_session()
+    s.active_stream_id = future_stream_id
+    s.pending_user_message = "Say hello"
+    s.pending_started_at = time.time() + 30
+    s.pending_attachments = []
+    s.save()
+    future_events = []
+    future_channel = MagicMock()
+    future_channel.put_nowait = lambda item: future_events.append(item)
+    STREAMS[future_stream_id] = future_channel
+    gateway_chat._run_gateway_chat_streaming(
+        s.session_id,
+        "Say hello",
+        "test-model",
+        str(tmp_path),
+        future_stream_id,
+        [],
+    )
+    future_errors = [item[1] for item in future_events if item[0] == "apperror"]
+    assert future_errors[-1]["type"] in {"model_not_found", "auth_mismatch"}
+    saved = models.get_session(s.session_id)
+    assert saved.messages[-1].get("_error") is True
+    assert "_turnDuration" not in saved.messages[-1]
+    future_payload_error = future_errors[-1]["session"]["messages"][-1]
+    assert future_payload_error.get("_error") is True
+    assert "_turnDuration" not in future_payload_error
 
     response_error[0] = "partial"
     partial_stream_id = "stream-gateway-partial-terminal-error-test"
@@ -553,6 +586,7 @@ def test_gateway_chat_worker_classifies_terminal_provider_error_without_text(tmp
     assert payload_messages[-2]["_partial"] is True
     assert payload_messages[-2]["content"] == "partial"
     assert payload_messages[-1]["_error"] is True
+    assert "_turnDuration" not in payload_messages[-1]
 
 
 def test_gateway_chat_worker_persists_reasoning_and_tool_state_on_terminal_error(tmp_path, monkeypatch):