David/HERMES-GIELIE
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix: harden terminal error worklog settlement

Browse source
This commit is contained in:
Frank Song 2026-07-20 15:34:18 +08:00
parent 1dd5cc43ce
commit de96c1d384
7 changed files with 119 additions and 33 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
api/gateway_chat.py
View file

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

20
api/routes.py
View file

@ -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

38
api/streaming.py
View file

@ -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,

18
static/messages.js
View file

@ -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;

36
tests/test_anchor_scene_persistence.py
View file

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

18
tests/test_issue3929_error_preserves_partial.py
View file

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

9
tests/test_live_to_final_anchor_visible_order.py
View file

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