wuyang/zk-data-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Persist active run event history

Browse Source
This commit is contained in:
武阳
2026-05-07 17:40:46 +08:00
parent 847b74ffb8
commit d57570e527
5 changed files with 316 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/api/server.py
+107 -16
View File
@@ -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],
frontend/app/components/assistant-ui/thread-list.tsx
+100 -5
View File
@@ -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<string, ClawRunEvent>();
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<string>();
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,
src/agent_runtime.py
+26 -3
View File
@@ -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,
tests/test_agent_runtime.py
+46
View File
@@ -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 = [
{
tests/test_gui_server.py
+37
View File
@@ -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)