From d57570e5271681a56cf922f536eaae1185ce62a7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=AD=A6=E9=98=B3?= Date: Thu, 7 May 2026 17:40:46 +0800 Subject: [PATCH] Persist active run event history --- backend/api/server.py | 123 +++++++++++++++--- .../components/assistant-ui/thread-list.tsx | 105 ++++++++++++++- src/agent_runtime.py | 29 ++++- tests/test_agent_runtime.py | 46 +++++++ tests/test_gui_server.py | 37 ++++++ 5 files changed, 316 insertions(+), 24 deletions(-) diff --git a/backend/api/server.py b/backend/api/server.py index efaa5be..ea14086 100644 --- a/backend/api/server.py +++ b/backend/api/server.py @@ -14,7 +14,7 @@ import shutil import threading import time import venv -from dataclasses import dataclass, replace +from dataclasses import dataclass, field, replace from pathlib import Path from threading import Lock, RLock from typing import Any @@ -214,6 +214,7 @@ class RunRecord: process_registry: RunProcessRegistry current_stage: str = '' error: str = '' + events: list[dict[str, Any]] = field(default_factory=list) class RunManager: @@ -252,6 +253,20 @@ class RunManager: record.error = error record.updated_at = time.time() + def record_event(self, run_id: str, event: dict[str, object]) -> None: + normalized = _normalize_run_event(event) + if normalized is None: + return + normalized['recorded_at'] = time.time() + with self._lock: + record = self._runs.get(run_id) + if record is None: + return + record.events.append(normalized) + if len(record.events) > 160: + del record.events[: len(record.events) - 160] + record.updated_at = time.time() + def finish(self, run_id: str, status: str) -> None: self.update(run_id, status=status) @@ -286,6 +301,7 @@ class RunManager: 'started_at': record.started_at, 'updated_at': record.updated_at, 'error': record.error, + 'events': [dict(event) for event in record.events], } def _select_session_record( @@ -835,6 +851,7 @@ def create_app(state: AgentState) -> FastAPI: stage = _runtime_event_stage(event) if stage: state.run_manager.update(run_record.run_id, stage=stage) + state.run_manager.record_event(run_record.run_id, event) _emit_runtime_event(event_sink, event) if request.resume_session_id is None: @@ -849,14 +866,16 @@ def create_app(state: AgentState) -> FastAPI: requested_session_id, ) if run_lock.locked(): + queued_event = { + 'type': 'run_queued', + 'run_id': run_record.run_id, + 'session_id': requested_session_id or '', + 'account_id': request.account_id or '', + } + state.run_manager.record_event(run_record.run_id, queued_event) _emit_runtime_event( event_sink, - { - 'type': 'run_queued', - 'run_id': run_record.run_id, - 'session_id': requested_session_id or '', - 'account_id': request.account_id or '', - }, + queued_event, ) with run_lock: if run_record.cancel_event.is_set(): @@ -872,15 +891,14 @@ def create_app(state: AgentState) -> FastAPI: 'stop_reason': 'cancelled', } state.run_manager.update(run_record.run_id, status='running', stage='后端开始执行') - _emit_runtime_event( - event_sink, - { - 'type': 'run_started', - 'run_id': run_record.run_id, - 'session_id': requested_session_id or '', - 'account_id': request.account_id or '', - }, - ) + started_event = { + 'type': 'run_started', + 'run_id': run_record.run_id, + 'session_id': requested_session_id or '', + 'account_id': request.account_id or '', + } + state.run_manager.record_event(run_record.run_id, started_event) + _emit_runtime_event(event_sink, started_event) agent.tool_context = replace( agent.tool_context, cancel_event=run_record.cancel_event, @@ -1289,6 +1307,79 @@ def _runtime_event_stage(event: dict[str, object]) -> str: return '' +_RUN_EVENT_TYPES = { + 'run_queued', + 'run_started', + 'tool_start', + 'tool_result', + 'final_text_start', + 'final_text_end', + 'user_review_required', + 'continuation_request', + 'prompt_length_check', + 'prompt_length_recovery', + 'auto_compact_summary', + 'auto_compact_circuit_breaker', + 'task_budget_exceeded', +} + + +def _normalize_run_event(event: dict[str, object]) -> dict[str, Any] | None: + event_type = event.get('type') + if not isinstance(event_type, str) or event_type not in _RUN_EVENT_TYPES: + return None + allowed = { + 'type', + 'run_id', + 'session_id', + 'account_id', + 'tool_name', + 'tool_call_id', + 'arguments', + 'assistant_content', + 'message_id', + 'ok', + 'metadata', + 'reason', + 'continuation_index', + 'turn_index', + 'strategy', + 'projected_input_tokens', + 'soft_input_limit_tokens', + 'hard_input_limit_tokens', + 'exceeds_hard_limit', + 'exceeds_soft_limit', + } + normalized = { + key: _json_safe_limited(value) + for key, value in event.items() + if key in allowed and value is not None + } + normalized['type'] = event_type + return normalized + + +def _json_safe_limited(value: Any, *, depth: int = 0) -> Any: + if depth > 4: + return str(value)[:500] + if isinstance(value, str): + return value[:2000] + if isinstance(value, (int, float, bool)) or value is None: + return value + if isinstance(value, dict): + return { + str(key)[:200]: _json_safe_limited(item, depth=depth + 1) + for key, item in list(value.items())[:80] + } + if isinstance(value, (list, tuple)): + return [_json_safe_limited(item, depth=depth + 1) for item in value[:80]] + try: + json.dumps(value) + return value + except TypeError: + return str(value)[:1000] + + 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 24899b3..c0d953f 100644 --- a/frontend/app/components/assistant-ui/thread-list.tsx +++ b/frontend/app/components/assistant-ui/thread-list.tsx @@ -64,12 +64,25 @@ export type ClawRunStatus = { | "interrupted"; current_stage?: string; started_at?: number; + events?: ClawRunEvent[]; }; type ClawActiveRunStatus = ClawRunStatus & { status: "queued" | "running"; }; +type ClawRunEvent = { + type?: string; + tool_name?: string; + tool_call_id?: string; + arguments?: unknown; + assistant_content?: string; + ok?: boolean; + metadata?: { + output_preview?: unknown; + }; +}; + type ClawStoredToolCall = { id?: string; name?: string; @@ -471,11 +484,7 @@ export function toReplayRepository( } if (isActiveRunStatus(runStatus)) { const id = `${fallbackSessionId}-run-${runStatus.run_id ?? "active"}`; - const text = - runStatus.status === "queued" - ? "当前会话已有任务在执行，本轮正在排队。" - : runStatus.current_stage || "后端正在执行这个会话。"; - const parts = [{ type: "reasoning", text }] as UIMessage["parts"]; + const parts = buildActiveRunParts(runStatus) as UIMessage["parts"]; const uiMessage: UIMessage = { id, role: "assistant", @@ -520,6 +529,92 @@ function isActiveRunStatus( return runStatus?.status === "queued" || runStatus?.status === "running"; } +function buildActiveRunParts(runStatus: ClawActiveRunStatus) { + const eventLines = summarizeRunEvents(runStatus); + const fallback = + runStatus.status === "queued" + ? "当前会话已有任务在执行，本轮正在排队。" + : runStatus.current_stage || "后端正在执行这个会话。"; + const reasoningText = eventLines.length ? eventLines.join("\n") : fallback; + return [ + { type: "reasoning", text: reasoningText }, + ...buildRunToolParts(runStatus.events ?? []), + ]; +} + +function summarizeRunEvents(runStatus: ClawActiveRunStatus) { + const lines: string[] = []; + for (const event of runStatus.events ?? []) { + if (event.type === "run_queued") { + lines.push("当前会话已有任务在执行，本轮正在排队。"); + } + if (event.type === "run_started") { + lines.push("后端已开始执行本轮任务。"); + } + if (event.type === "tool_start") { + lines.push( + event.tool_name ? `调用工具 ${event.tool_name}` : "正在调用工具", + ); + } + if (event.type === "tool_result") { + lines.push( + event.tool_name ? `工具完成 ${event.tool_name}` : "工具调用完成", + ); + } + if (event.type === "final_text_start") { + lines.push("正在整理回复"); + } + if (event.type === "user_review_required") { + lines.push("等待用户 review"); + } + } + if (runStatus.current_stage && lines.at(-1) !== runStatus.current_stage) { + lines.push(runStatus.current_stage); + } + return lines.slice(-40); +} + +function buildRunToolParts(events: readonly ClawRunEvent[]) { + const resultById = new Map(); + for (const event of events) { + if (event.type === "tool_result" && event.tool_call_id) { + resultById.set(event.tool_call_id, event); + } + } + const seen = new Set(); + return events.flatMap((event) => { + if ( + event.type !== "tool_start" || + !event.tool_call_id || + !event.tool_name || + seen.has(event.tool_call_id) + ) { + return []; + } + seen.add(event.tool_call_id); + const result = resultById.get(event.tool_call_id); + const outputPreview = result?.metadata?.output_preview; + const output = + typeof outputPreview === "string" + ? outputPreview + : result + ? result.ok === false + ? "工具执行失败，等待最终结果汇总。" + : "工具调用完成，等待最终结果汇总。" + : undefined; + return [ + { + type: "dynamic-tool", + toolName: event.tool_name, + toolCallId: event.tool_call_id, + state: output === undefined ? "input-available" : "output-available", + input: attachStageNote(event.arguments ?? {}, event.assistant_content), + ...(output === undefined ? {} : { output }), + }, + ]; + }); +} + function getReplayMessages( messages: readonly ClawStoredMessage[], runStatus?: ClawRunStatus | null, diff --git a/src/agent_runtime.py b/src/agent_runtime.py index 95381f5..ed6e43c 100644 --- a/src/agent_runtime.py +++ b/src/agent_runtime.py @@ -1270,9 +1270,15 @@ class LocalCodingAgent: self.last_run_result = result return result - final_output = ( - last_content - or 'Stopped: max turns reached before the model produced a final answer.' + final_output = self._build_max_turns_output( + max_turns=self.runtime_config.max_turns, + last_content=last_content, + tool_calls=tool_calls, + ) + session.append_assistant( + final_output, + message_id=f'assistant_{len(session.messages)}', + stop_reason='max_turns', ) _append_final_text_stream_events(stream_events, final_output) result = AgentRunResult( @@ -1299,6 +1305,23 @@ class LocalCodingAgent: self.last_run_result = result return result + @staticmethod + def _build_max_turns_output( + *, + max_turns: int, + last_content: str, + tool_calls: int, + ) -> str: + lines = [ + f'已达到本轮最大步骤上限（{max_turns} 轮），为了避免继续空转，当前任务已暂停。', + ] + if tool_calls: + lines.append('最后一轮已经执行完工具，但还没有来得及整理成最终回复。') + if last_content.strip(): + lines.append(f'最后一次阶段说明：{last_content.strip()}') + lines.append('你可以继续补充指令，我会基于当前会话结果接着处理。') + return '\n\n'.join(lines) + def _query_model( self, session: AgentSessionState, diff --git a/tests/test_agent_runtime.py b/tests/test_agent_runtime.py index 1945b29..26bfad2 100644 --- a/tests/test_agent_runtime.py +++ b/tests/test_agent_runtime.py @@ -285,6 +285,52 @@ class AgentRuntimeTests(unittest.TestCase): self.assertGreaterEqual(len(result.transcript), 5) self.assertGreaterEqual(len(result.file_history), 0) + def test_agent_persists_max_turns_explanation_after_tool_result(self) -> None: + responses = [ + { + 'choices': [ + { + 'message': { + 'role': 'assistant', + 'content': '我先读取文件。', + 'tool_calls': [ + { + 'id': 'call_1', + 'type': 'function', + 'function': { + 'name': 'read_file', + 'arguments': '{"path": "hello.txt"}', + }, + } + ], + }, + 'finish_reason': 'tool_calls', + } + ] + } + ] + with tempfile.TemporaryDirectory() as tmp_dir: + workspace = Path(tmp_dir) + (workspace / 'hello.txt').write_text('hello world\n', encoding='utf-8') + with patch( + 'src.openai_compat.request.urlopen', + side_effect=make_urlopen_side_effect(responses), + ): + agent = LocalCodingAgent( + model_config=ModelConfig( + model='Qwen/Qwen3-Coder-30B-A3B-Instruct', + base_url='http://127.0.0.1:8000/v1', + ), + runtime_config=AgentRuntimeConfig(cwd=workspace, max_turns=1), + ) + result = agent.run('Inspect hello.txt') + + self.assertEqual(result.stop_reason, 'max_turns') + self.assertIn('最大步骤上限', result.final_output) + self.assertEqual(result.transcript[-1]['role'], 'assistant') + self.assertEqual(result.transcript[-1]['stop_reason'], 'max_turns') + self.assertIn('最大步骤上限', result.transcript[-1]['content']) + def test_agent_stops_after_data_agent_goal_requires_review(self) -> None: responses = [ { diff --git a/tests/test_gui_server.py b/tests/test_gui_server.py index bde6883..7db31d4 100644 --- a/tests/test_gui_server.py +++ b/tests/test_gui_server.py @@ -262,6 +262,43 @@ class GuiServerTests(unittest.TestCase): self.assertEqual(payload['status'], 'running') self.assertEqual(payload['current_stage'], 'first stage') + def test_latest_run_includes_runtime_event_history(self) -> None: + with tempfile.TemporaryDirectory() as d: + client, state = _build_client(Path(d)) + record = state.run_manager.start('alice', 'thread-1') + state.run_manager.update(record.run_id, status='running') + state.run_manager.record_event( + record.run_id, + { + 'type': 'tool_start', + 'tool_name': 'python_exec', + 'tool_call_id': 'call-1', + 'arguments': {'code': 'print(1)'}, + 'assistant_content': '我用 Python 分析一下。', + }, + ) + state.run_manager.record_event( + record.run_id, + { + 'type': 'tool_result', + 'tool_name': 'python_exec', + 'tool_call_id': 'call-1', + 'ok': True, + 'metadata': {'output_preview': 'exit_code=0'}, + }, + ) + + latest = client.get( + '/api/runs/latest', + params={'account_id': 'alice', 'session_id': 'thread-1'}, + ) + + self.assertEqual(latest.status_code, 200) + events = latest.json()['events'] + self.assertEqual([event['type'] for event in events], ['tool_start', 'tool_result']) + self.assertEqual(events[0]['tool_name'], 'python_exec') + self.assertEqual(events[1]['metadata']['output_preview'], 'exit_code=0') + def test_cancel_run_marks_streaming_session_tail_cancelled(self) -> None: with tempfile.TemporaryDirectory() as d: root = Path(d)