Improve replayed run status handling
This commit is contained in:
+56
-18
@@ -830,6 +830,13 @@ def create_app(state: AgentState) -> FastAPI:
|
||||
agent = state.agent_for(request.account_id, requested_session_id)
|
||||
config = state.config_for(request.account_id)
|
||||
session_directory = state.account_paths(request.account_id)['sessions']
|
||||
|
||||
def emit_agent_event(event: dict[str, object]) -> None:
|
||||
stage = _runtime_event_stage(event)
|
||||
if stage:
|
||||
state.run_manager.update(run_record.run_id, stage=stage)
|
||||
_emit_runtime_event(event_sink, event)
|
||||
|
||||
if request.resume_session_id is None:
|
||||
_save_in_progress_session(
|
||||
directory=session_directory,
|
||||
@@ -898,14 +905,14 @@ def create_app(state: AgentState) -> FastAPI:
|
||||
request.prompt.strip(),
|
||||
stored,
|
||||
runtime_context=request.runtime_context,
|
||||
event_sink=event_sink,
|
||||
event_sink=emit_agent_event,
|
||||
)
|
||||
else:
|
||||
result = agent.run(
|
||||
request.prompt.strip(),
|
||||
session_id=requested_session_id,
|
||||
runtime_context=request.runtime_context,
|
||||
event_sink=event_sink,
|
||||
event_sink=emit_agent_event,
|
||||
)
|
||||
except Exception as exc:
|
||||
_mark_session_interrupted(
|
||||
@@ -1164,23 +1171,11 @@ def _sanitize_stored_session_for_resume(
|
||||
stored: StoredAgentSession,
|
||||
) -> StoredAgentSession:
|
||||
trimmed, changed = _trim_incomplete_message_tail(stored.messages)
|
||||
messages = _without_run_status_messages(trimmed)
|
||||
if len(messages) != len(trimmed):
|
||||
changed = True
|
||||
if not changed:
|
||||
return stored
|
||||
messages = list(trimmed)
|
||||
if not _last_message_is_run_status(messages, 'interrupted'):
|
||||
messages.append(
|
||||
{
|
||||
'role': 'assistant',
|
||||
'content': _interrupted_session_message('interrupted'),
|
||||
'state': 'final',
|
||||
'stop_reason': 'interrupted',
|
||||
'metadata': {
|
||||
'kind': 'run_status',
|
||||
'status': 'interrupted',
|
||||
'created_at_ms': int(time.time() * 1000),
|
||||
},
|
||||
}
|
||||
)
|
||||
budget_state = (
|
||||
dict(stored.budget_state)
|
||||
if isinstance(stored.budget_state, dict)
|
||||
@@ -1189,7 +1184,7 @@ def _sanitize_stored_session_for_resume(
|
||||
budget_state['status'] = 'interrupted'
|
||||
return replace(
|
||||
stored,
|
||||
messages=tuple(messages),
|
||||
messages=messages,
|
||||
budget_state=budget_state,
|
||||
)
|
||||
|
||||
@@ -1232,6 +1227,21 @@ def _assistant_has_trailing_tool_call(message: dict[str, Any]) -> bool:
|
||||
return isinstance(tool_calls, list) and bool(tool_calls)
|
||||
|
||||
|
||||
def _without_run_status_messages(
|
||||
messages: tuple[dict[str, Any], ...],
|
||||
) -> tuple[dict[str, Any], ...]:
|
||||
return tuple(
|
||||
message
|
||||
for message in messages
|
||||
if not _is_run_status_message(message)
|
||||
)
|
||||
|
||||
|
||||
def _is_run_status_message(message: dict[str, Any]) -> bool:
|
||||
metadata = message.get('metadata')
|
||||
return isinstance(metadata, dict) and metadata.get('kind') == 'run_status'
|
||||
|
||||
|
||||
def _last_message_is_run_status(messages: list[dict[str, Any]], status: str) -> bool:
|
||||
if not messages:
|
||||
return False
|
||||
@@ -1251,6 +1261,34 @@ def _interrupted_session_message(status: str) -> str:
|
||||
return '上一次任务已中断,后台没有正在执行的进程。你可以继续回复,或重新发起任务。'
|
||||
|
||||
|
||||
def _runtime_event_stage(event: dict[str, object]) -> str:
|
||||
event_type = event.get('type')
|
||||
if event_type == 'tool_start':
|
||||
tool_name = str(event.get('tool_name') or '').strip()
|
||||
return f'调用工具 {tool_name}' if tool_name else '正在调用工具'
|
||||
if event_type == 'tool_result':
|
||||
tool_name = str(event.get('tool_name') or '').strip()
|
||||
return f'工具完成 {tool_name}' if tool_name else '工具调用完成'
|
||||
if event_type == 'final_text_start':
|
||||
return '正在整理回复'
|
||||
if event_type == 'final_text_end':
|
||||
return '回复整理完成'
|
||||
if event_type == 'user_review_required':
|
||||
return '等待用户 review'
|
||||
if event_type == 'continuation_request':
|
||||
return '继续补全回复'
|
||||
if event_type in {
|
||||
'prompt_length_check',
|
||||
'prompt_length_recovery',
|
||||
'auto_compact_summary',
|
||||
'auto_compact_circuit_breaker',
|
||||
}:
|
||||
return '整理上下文'
|
||||
if event_type == 'task_budget_exceeded':
|
||||
return '达到任务预算限制'
|
||||
return ''
|
||||
|
||||
|
||||
def _emit_runtime_event(
|
||||
event_sink: Any | None,
|
||||
event: dict[str, object],
|
||||
|
||||
Reference in New Issue
Block a user