diff --git a/backend/api/server.py b/backend/api/server.py index 124a47a..efaa5be 100644 --- a/backend/api/server.py +++ b/backend/api/server.py @@ -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], diff --git a/frontend/app/components/assistant-ui/thread-list.tsx b/frontend/app/components/assistant-ui/thread-list.tsx index 9beef27..24899b3 100644 --- a/frontend/app/components/assistant-ui/thread-list.tsx +++ b/frontend/app/components/assistant-ui/thread-list.tsx @@ -39,6 +39,7 @@ type ClawStoredMessage = { stop_reason?: string; metadata?: { elapsed_ms?: unknown; + kind?: unknown; placeholder?: unknown; phase?: unknown; status?: unknown; @@ -523,8 +524,11 @@ function getReplayMessages( messages: readonly ClawStoredMessage[], runStatus?: ClawRunStatus | null, ) { - if (isActiveRunStatus(runStatus)) return [...messages]; - return trimIncompleteReplayTail(messages); + const visibleMessages = isActiveRunStatus(runStatus) + ? messages.filter((message) => !isRunStatusMessage(message)) + : [...messages]; + if (isActiveRunStatus(runStatus)) return visibleMessages; + return trimIncompleteReplayTail(visibleMessages); } function shouldShowInterruptedNotice( @@ -532,6 +536,7 @@ function shouldShowInterruptedNotice( runStatus?: ClawRunStatus | null, ) { if (isActiveRunStatus(runStatus)) return false; + if (hasRunStatusMessage(messages, runStatus?.status)) return false; return ( runStatus?.status === "interrupted" || runStatus?.status === "cancelled" || @@ -574,6 +579,21 @@ function hasTrailingAssistantToolCall(message?: ClawStoredMessage) { return message?.role === "assistant" && Boolean(message.tool_calls?.length); } +function hasRunStatusMessage( + messages: readonly ClawStoredMessage[], + status?: string, +) { + return messages.some( + (message) => + isRunStatusMessage(message) && + (!status || message.metadata?.status === status), + ); +} + +function isRunStatusMessage(message?: ClawStoredMessage) { + return message?.metadata?.kind === "run_status"; +} + function normalizeRunTimestampMs(value: unknown) { if (typeof value !== "number" || !Number.isFinite(value)) return undefined; return value < 10_000_000_000 ? Math.round(value * 1000) : Math.round(value); diff --git a/tests/test_gui_server.py b/tests/test_gui_server.py index d685fbb..bde6883 100644 --- a/tests/test_gui_server.py +++ b/tests/test_gui_server.py @@ -16,7 +16,12 @@ from pathlib import Path from fastapi.testclient import TestClient -from backend.api.server import AgentState, create_app +from backend.api.server import ( + AgentState, + create_app, + _runtime_event_stage, + _sanitize_stored_session_for_resume, +) from src.agent_types import AgentRunResult from src.session_store import StoredAgentSession, load_agent_session, save_agent_session @@ -360,6 +365,52 @@ class GuiServerTests(unittest.TestCase): self.assertEqual(response.status_code, 200) self.assertEqual(response.json()['status'], 'interrupted') + def test_resume_sanitizer_removes_ui_run_status_messages(self) -> None: + stored = StoredAgentSession( + session_id='thread-1', + model_config={'model': 'test-model'}, + runtime_config={}, + system_prompt_parts=('system prompt',), + user_context={}, + system_context={}, + messages=( + {'role': 'user', 'content': 'hello'}, + { + 'role': 'assistant', + 'content': '上一次任务已中断，后台没有正在执行的进程。', + 'state': 'final', + 'stop_reason': 'interrupted', + 'metadata': {'kind': 'run_status', 'status': 'interrupted'}, + }, + {'role': 'user', 'content': 'continue'}, + ), + turns=1, + tool_calls=0, + usage={}, + total_cost_usd=0.0, + file_history=(), + budget_state={'status': 'interrupted'}, + plugin_state={}, + ) + + sanitized = _sanitize_stored_session_for_resume(stored) + + self.assertEqual( + [message['content'] for message in sanitized.messages], + ['hello', 'continue'], + ) + + def test_runtime_event_stage_names_long_running_work(self) -> None: + self.assertEqual( + _runtime_event_stage({'type': 'tool_start', 'tool_name': 'python_exec'}), + '调用工具 python_exec', + ) + self.assertEqual( + _runtime_event_stage({'type': 'final_text_start'}), + '正在整理回复', + ) + self.assertEqual(_runtime_event_stage({'type': 'server_heartbeat'}), '') + def test_chat_uses_account_scoped_model_config(self) -> None: with tempfile.TemporaryDirectory() as d: client, _ = _build_client(Path(d))