diff --git a/README.md b/README.md index 45454e6..84ea58f 100644 --- a/README.md +++ b/README.md @@ -118,7 +118,7 @@ cd frontend pnpm test ``` -当前回归基线为后端 76 项测试、前端 26 项测试，且生产构建通过。测试数量会随功能增长，以本地实际输出和 CI 为准。 +当前回归基线为后端 80 项测试、前端 27 项测试，且生产构建通过。测试数量会随功能增长，以本地实际输出和 CI 为准。 构建产物位于 `frontend/dist`，该目录不提交到 Git。 diff --git a/backend/app/agent/permissions.py b/backend/app/agent/permissions.py index aee818f..b0d7c8c 100644 --- a/backend/app/agent/permissions.py +++ b/backend/app/agent/permissions.py @@ -104,6 +104,11 @@ class PermissionManager: ticket.future.set_result(decision) return True + def get_ticket(self, run_id: str, request_id: str) -> PermissionTicket | None: + """只读返回待确认票据，供 Trace 记录权限类型；不暴露 Future 给接口层。""" + + return self._pending.get((run_id, request_id)) + def cancel_run(self, run_id: str) -> None: for key, ticket in list(self._pending.items()): if ticket.run_id == run_id: diff --git a/backend/app/agent/runtime.py b/backend/app/agent/runtime.py index 78af004..3daf87d 100644 --- a/backend/app/agent/runtime.py +++ b/backend/app/agent/runtime.py @@ -7,17 +7,20 @@ import json from collections.abc import AsyncIterator from dataclasses import dataclass, field from datetime import datetime, timezone +from time import perf_counter from typing import TYPE_CHECKING from uuid import uuid4 from app.agent.permissions import PermissionManager, PermissionMode from app.agent.tools import ToolExecutionContext, ToolNotFoundError, ToolRegistry +from app.agent.trace_repository import AgentTraceRepository, sanitize_trace_value from app.contracts import ( AgentEvent, AgentEventType, AgentRun, AgentRunCreateRequest, AgentRunStatus, + AgentTraceResponse, Citation, Message, MessageRole, @@ -61,6 +64,7 @@ class RunRecord: events: list[AgentEvent] = field(default_factory=list) subscribers: set[asyncio.Queue[AgentEvent]] = field(default_factory=set) task: asyncio.Task[None] | None = None + next_sequence: int = 0 class AgentRuntime: @@ -72,11 +76,13 @@ class AgentRuntime: tools: ToolRegistry, permissions: PermissionManager, skills: SkillRuntime | None = None, + trace_repository: AgentTraceRepository | None = None, ) -> None: self.providers = providers self.tools = tools self.permissions = permissions self.skills = skills + self.trace_repository = trace_repository or AgentTraceRepository() self._records: dict[str, RunRecord] = {} async def create_run(self, request: AgentRunCreateRequest) -> AgentRun: @@ -116,22 +122,38 @@ class AgentRuntime: skill_config=skill_config, allowed_tools=allowed_tools, ) + self.trace_repository.create_run( + run, + request, + self._config_snapshot(record), + ) self._records[run.run_id] = record record.task = asyncio.create_task(self._execute(record), name=run.run_id) return run.model_copy(deep=True) def get_run(self, run_id: str) -> AgentRun: - return self._get_record(run_id).run.model_copy(deep=True) + record = self._records.get(run_id) + if record is not None: + return record.run.model_copy(deep=True) + run = self.trace_repository.recover_interrupted(run_id) + if run is None: + raise AgentRunNotFoundError(run_id) + return run.model_copy(deep=True) def list_runs(self, limit: int, offset: int) -> tuple[list[AgentRun], int]: - records = sorted( - self._records.values(), key=lambda item: item.run.created_at, reverse=True - ) - items = [item.run.model_copy(deep=True) for item in records[offset : offset + limit]] - return items, len(records) + items, total = self.trace_repository.list_runs(limit=limit, offset=offset) + recovered = [ + self.trace_repository.recover_interrupted(item.run_id) or item + if item.run_id not in self._records + else self._records[item.run_id].run.model_copy(deep=True) + for item in items + ] + return recovered, total async def cancel(self, run_id: str) -> AgentRun: - record = self._get_record(run_id) + record = self._records.get(run_id) + if record is None: + return self.get_run(run_id) if record.run.status in TERMINAL_STATUSES: return record.run.model_copy(deep=True) record.run.cancelled = True @@ -144,23 +166,53 @@ class AgentRuntime: return record.run.model_copy(deep=True) def resolve_permission(self, run_id: str, request_id: str, decision: str) -> bool: - self._get_record(run_id) - return self.permissions.resolve(run_id, request_id, decision) + record = self._records.get(run_id) + if record is None: + return False + ticket = self.permissions.get_ticket(run_id, request_id) + resolved = self.permissions.resolve(run_id, request_id, decision) + if resolved: + self._publish( + record, + AgentEventType.permission_resolved, + { + "request_id": request_id, + "permission": ticket.permission if ticket else None, + "decision": decision, + }, + ) + return resolved - async def events(self, run_id: str) -> AsyncIterator[AgentEvent]: - record = self._get_record(run_id) - # 先回放快照再订阅实时事件，使晚加入的 SSE 客户端也能恢复界面状态。 - # TODO(agent): 持久化事件并支持 Last-Event-ID，进程重启后仍可续传。 + async def events( + self, run_id: str, *, after_sequence: int = -1 + ) -> AsyncIterator[AgentEvent]: + record = self._records.get(run_id) + run = self.get_run(run_id) + if record is None: + for event in self.trace_repository.list_events( + run_id, after_sequence=after_sequence + ): + yield event + return + + # 先注册订阅再读持久化历史；同一事件循环内没有 await，不会丢失交界事件。 queue: asyncio.Queue[AgentEvent] = asyncio.Queue() record.subscribers.add(queue) - history = [event.model_copy(deep=True) for event in record.events] + history = self.trace_repository.list_events( + run_id, after_sequence=after_sequence + ) + last_sequence = after_sequence try: for event in history: + last_sequence = event.sequence yield event - if record.run.status in TERMINAL_STATUSES: + if run.status in TERMINAL_STATUSES: return while True: event = await queue.get() + if event.sequence <= last_sequence: + continue + last_sequence = event.sequence yield event.model_copy(deep=True) if event.event in { AgentEventType.run_completed, @@ -172,7 +224,9 @@ class AgentRuntime: record.subscribers.discard(queue) async def wait(self, run_id: str) -> AgentRun: - record = self._get_record(run_id) + record = self._records.get(run_id) + if record is None: + return self.get_run(run_id) if record.task: try: await asyncio.shield(record.task) @@ -180,6 +234,17 @@ class AgentRuntime: pass return record.run.model_copy(deep=True) + def get_trace( + self, run_id: str, *, after_sequence: int, limit: int + ) -> AgentTraceResponse: + self.get_run(run_id) + trace = self.trace_repository.get_trace( + run_id, after_sequence=after_sequence, limit=limit + ) + if trace is None: + raise AgentRunNotFoundError(run_id) + return trace + async def _execute(self, record: RunRecord) -> None: try: async with asyncio.timeout(record.request.run_timeout_seconds): @@ -210,15 +275,51 @@ class AgentRuntime: for step in range(1, record.request.max_steps + 1): record.run.current_step = step record.run.updated_at = datetime.now(timezone.utc) - turn = await provider.complete( - ModelRequest( - provider_id=record.request.provider_id, - model=record.request.model, - system=(record.skill_config.system_prompt if record.skill_config else None), - messages=messages, - tools=allowed_tools, - metadata=self._request_metadata(record), + model_call_id = f"model_call_{uuid4().hex}" + started_at = perf_counter() + self._publish( + record, + AgentEventType.model_call_started, + { + "model_call_id": model_call_id, + "step": step, + "provider_id": record.request.provider_id, + "model": record.request.model, + }, + ) + try: + turn = await provider.complete( + ModelRequest( + provider_id=record.request.provider_id, + model=record.request.model, + system=(record.skill_config.system_prompt if record.skill_config else None), + messages=messages, + tools=allowed_tools, + metadata=self._request_metadata(record), + ) ) + except Exception as exc: + self._publish( + record, + AgentEventType.model_call_failed, + { + "model_call_id": model_call_id, + "duration_ms": int((perf_counter() - started_at) * 1000), + "error_code": getattr(exc, "code", type(exc).__name__), + }, + ) + raise + self._publish( + record, + AgentEventType.model_call_completed, + { + "model_call_id": model_call_id, + "duration_ms": int((perf_counter() - started_at) * 1000), + "finish_reason": "tool_calls" if turn.tool_calls else "stop", + "input_tokens": turn.input_tokens, + "output_tokens": turn.output_tokens, + "tool_call_count": len(turn.tool_calls), + }, ) record.run.token_usage += turn.input_tokens + turn.output_tokens self._publish( @@ -257,7 +358,7 @@ class AgentRuntime: async def execute(call: ToolCall) -> ToolResult: async with semaphore: - return await self._execute_tool(record, call) + return await self._execute_tool(record, call, model_call_id) results = await asyncio.gather(*(execute(call) for call in calls)) for call, result in zip(calls, results): @@ -290,8 +391,13 @@ class AgentRuntime: self._fail(record, "MAX_STEPS_EXCEEDED", "Agent reached its maximum step count.") - async def _execute_tool(self, record: RunRecord, call: ToolCall) -> ToolResult: - self._publish(record, AgentEventType.tool_call, call.model_dump(mode="json")) + async def _execute_tool( + self, record: RunRecord, call: ToolCall, parent_model_call_id: str + ) -> ToolResult: + started_at = perf_counter() + call_data = call.model_dump(mode="json") + call_data["parent_model_call_id"] = parent_model_call_id + self._publish(record, AgentEventType.tool_call, call_data) try: registered = self.tools.get(call.name) except ToolNotFoundError: @@ -305,7 +411,9 @@ class AgentRuntime: error_code="TOOL_NOT_ALLOWED", error_message="Tool is not included in allowed_tools.", ) - self._publish(record, AgentEventType.tool_result, result.model_dump(mode="json")) + self._publish_tool_result( + record, result, parent_model_call_id, started_at + ) return result permission = registered.definition.permission if registered else None @@ -317,7 +425,9 @@ class AgentRuntime: error_code="NETWORK_NOT_ALLOWED", error_message="Agent run does not allow network tools.", ) - self._publish(record, AgentEventType.tool_result, result.model_dump(mode="json")) + self._publish_tool_result( + record, result, parent_model_call_id, started_at + ) return result mode = self.permissions.mode_for(permission) if mode == PermissionMode.deny: @@ -348,11 +458,13 @@ class AgentRuntime: error_code="PERMISSION_TIMEOUT", error_message="Tool permission confirmation timed out.", ) - self._publish( - record, AgentEventType.tool_result, result.model_dump(mode="json") + self._publish_tool_result( + record, result, parent_model_call_id, started_at ) return result record.run.status = AgentRunStatus.running + record.run.updated_at = datetime.now(timezone.utc) + self.trace_repository.save_run(record.run) result = ( await self._invoke_tool(record, call) if decision in {"allow_once", "allow_session"} @@ -361,9 +473,21 @@ class AgentRuntime: else: result = await self._invoke_tool(record, call) - self._publish(record, AgentEventType.tool_result, result.model_dump(mode="json")) + self._publish_tool_result(record, result, parent_model_call_id, started_at) return result + def _publish_tool_result( + self, + record: RunRecord, + result: ToolResult, + parent_model_call_id: str, + started_at: float, + ) -> None: + data = result.model_dump(mode="json") + data["parent_model_call_id"] = parent_model_call_id + data["duration_ms"] = int((perf_counter() - started_at) * 1000) + self._publish(record, AgentEventType.tool_result, data) + async def _invoke_tool(self, record: RunRecord, call: ToolCall) -> ToolResult: try: return await asyncio.wait_for( @@ -411,15 +535,19 @@ class AgentRuntime: def _publish( self, record: RunRecord, event_type: AgentEventType, data: dict[str, object] ) -> None: + sanitized = sanitize_trace_value(data) + assert isinstance(sanitized, dict) event = AgentEvent( event=event_type, run_id=record.run.run_id, - sequence=len(record.events), - data=data, + sequence=record.next_sequence, + data=sanitized, timestamp=datetime.now(timezone.utc), ) + record.next_sequence += 1 record.events.append(event) - # 内存事件只保留最近窗口；完整审计轨迹应由后续持久化层承担。 + self.trace_repository.append_event(record.run, event) + # 内存只保留实时订阅窗口；完整审计轨迹由 SQLite 保存。 if len(record.events) > MAX_EVENTS_PER_RUN: del record.events[: len(record.events) - MAX_EVENTS_PER_RUN] for queue in record.subscribers: @@ -433,6 +561,21 @@ class AgentRuntime: metadata["retrieval"] = record.skill_config.retrieval.model_dump(mode="json") return metadata + def _config_snapshot(self, record: RunRecord) -> dict[str, object]: + provider = self.providers.get(record.request.provider_id).config + return { + "provider_id": record.request.provider_id, + "provider_type": provider.provider_type.value, + "model": record.request.model, + "capabilities": [item.value for item in provider.capabilities], + "skill_id": record.request.skill_id, + "allowed_tools": list(record.allowed_tools), + "max_steps": record.request.max_steps, + "token_budget": record.request.token_budget, + "allow_network": record.request.allow_network, + "metadata": record.request.metadata, + } + def _collect_citations(self, record: RunRecord, result: ToolResult) -> None: if not result.success or not isinstance(result.output, dict): return diff --git a/backend/app/agent/trace_repository.py b/backend/app/agent/trace_repository.py new file mode 100644 index 0000000..b56b96a --- /dev/null +++ b/backend/app/agent/trace_repository.py @@ -0,0 +1,359 @@ +"""Agent Run/Event 持久化与 Trace 查询。 + +SQLite 中的事件是 SSE、前端 Trace 和 Benchmark 的共同事实来源。写入前统一脱敏和 +限长，避免 Secret 或无限大的 Tool Result 进入审计数据。 +""" + +from __future__ import annotations + +import json +import re +from datetime import datetime, timezone +from typing import Any + +from app.contracts import ( + AgentEvent, + AgentEventType, + AgentRun, + AgentRunCreateRequest, + AgentRunStatus, + AgentTraceResponse, + AgentTraceSummary, +) +from app.database.db import connect, transaction + +MAX_TRACE_STRING = 4_096 +MAX_TRACE_COLLECTION = 100 +MAX_TRACE_DEPTH = 8 +_SECRET_KEYS = { + "api_key", + "apikey", + "authorization", + "access_token", + "refresh_token", + "client_secret", + "password", + "secret", + "token", +} +_SECRET_KEY_SUFFIXES = ("_api_key", "_password", "_secret") +_TERMINAL_VALUES = { + AgentRunStatus.completed.value, + AgentRunStatus.failed.value, + AgentRunStatus.cancelled.value, +} +_BEARER_PATTERN = re.compile(r"(?i)\bBearer\s+[^\s,;]+") +_API_KEY_PATTERN = re.compile(r"\bsk-[A-Za-z0-9_-]{8,}\b") + + +def sanitize_trace_value(value: Any, *, depth: int = 0) -> Any: + """递归净化 Trace 数据；键名疑似 Secret 时不保留原值。""" + + if depth >= MAX_TRACE_DEPTH: + return "[MAX_DEPTH]" + if isinstance(value, dict): + sanitized: dict[str, Any] = {} + for index, (key, item) in enumerate(value.items()): + if index >= MAX_TRACE_COLLECTION: + sanitized["__truncated__"] = True + break + normalized = str(key).casefold().replace("-", "_") + sanitized[str(key)] = ( + "[REDACTED]" + if normalized in _SECRET_KEYS + or normalized.endswith(_SECRET_KEY_SUFFIXES) + else sanitize_trace_value(item, depth=depth + 1) + ) + return sanitized + if isinstance(value, (list, tuple)): + items = [ + sanitize_trace_value(item, depth=depth + 1) + for item in value[:MAX_TRACE_COLLECTION] + ] + if len(value) > MAX_TRACE_COLLECTION: + items.append("[TRUNCATED]") + return items + if isinstance(value, str): + value = _BEARER_PATTERN.sub("Bearer [REDACTED]", value) + value = _API_KEY_PATTERN.sub("[REDACTED]", value) + if len(value) > MAX_TRACE_STRING: + return f"{value[:MAX_TRACE_STRING]}...[TRUNCATED]" + return value + if value is None or isinstance(value, (str, int, float, bool)): + return value + return sanitize_trace_value(str(value), depth=depth + 1) + + +class AgentTraceRepository: + def create_run( + self, + run: AgentRun, + request: AgentRunCreateRequest, + config_snapshot: dict[str, Any], + ) -> None: + conn = connect() + try: + with transaction(conn): + conn.execute( + """ + INSERT INTO agent_runs( + run_id, status, run_json, request_json, config_snapshot_json, + created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?) + """, + ( + run.run_id, + run.status.value, + self._serialize_run(run), + json.dumps( + sanitize_trace_value(request.model_dump(mode="json")), + ensure_ascii=False, + ), + json.dumps( + sanitize_trace_value(config_snapshot), ensure_ascii=False + ), + run.created_at.isoformat(), + run.updated_at.isoformat(), + ), + ) + finally: + conn.close() + + def save_run(self, run: AgentRun) -> None: + conn = connect() + try: + with transaction(conn): + self._update_run(conn, run) + finally: + conn.close() + + def append_event(self, run: AgentRun, event: AgentEvent) -> None: + """在同一事务中保存最新 Run 和事件；复写同一序号时保持幂等。""" + + conn = connect() + try: + with transaction(conn): + self._update_run(conn, run) + conn.execute( + """ + INSERT INTO agent_events(run_id, sequence, event, data_json, timestamp) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(run_id, sequence) DO NOTHING + """, + ( + event.run_id, + event.sequence, + event.event.value, + json.dumps(event.data, ensure_ascii=False), + event.timestamp.isoformat(), + ), + ) + finally: + conn.close() + + def get_run(self, run_id: str) -> AgentRun | None: + conn = connect() + try: + row = conn.execute( + "SELECT run_json FROM agent_runs WHERE run_id = ?", (run_id,) + ).fetchone() + return AgentRun.model_validate_json(row["run_json"]) if row else None + finally: + conn.close() + + def list_runs(self, limit: int, offset: int) -> tuple[list[AgentRun], int]: + conn = connect() + try: + total = int(conn.execute("SELECT COUNT(*) FROM agent_runs").fetchone()[0]) + rows = conn.execute( + """ + SELECT run_json FROM agent_runs + ORDER BY created_at DESC LIMIT ? OFFSET ? + """, + (limit, offset), + ).fetchall() + return [AgentRun.model_validate_json(row["run_json"]) for row in rows], total + finally: + conn.close() + + def list_events( + self, run_id: str, *, after_sequence: int = -1, limit: int | None = None + ) -> list[AgentEvent]: + conn = connect() + try: + sql = """ + SELECT event, sequence, data_json, timestamp + FROM agent_events + WHERE run_id = ? AND sequence > ? + ORDER BY sequence + """ + params: tuple[Any, ...] = (run_id, after_sequence) + if limit is not None: + sql += " LIMIT ?" + params += (limit,) + return [self._event_from_row(run_id, row) for row in conn.execute(sql, params)] + finally: + conn.close() + + def get_trace( + self, run_id: str, *, after_sequence: int, limit: int + ) -> AgentTraceResponse | None: + conn = connect() + try: + row = conn.execute( + """ + SELECT run_json, config_snapshot_json + FROM agent_runs WHERE run_id = ? + """, + (run_id,), + ).fetchone() + if row is None: + return None + run = AgentRun.model_validate_json(row["run_json"]) + event_rows = conn.execute( + """ + SELECT event, sequence, data_json, timestamp + FROM agent_events + WHERE run_id = ? AND sequence > ? + ORDER BY sequence LIMIT ? + """, + (run_id, after_sequence, limit + 1), + ).fetchall() + has_more = len(event_rows) > limit + items = [ + self._event_from_row(run_id, item) for item in event_rows[:limit] + ] + counts = { + item["event"]: int(item["count"]) + for item in conn.execute( + """ + SELECT event, COUNT(*) AS count + FROM agent_events WHERE run_id = ? GROUP BY event + """, + (run_id,), + ) + } + tool_errors = int( + conn.execute( + """ + SELECT COUNT(*) FROM agent_events + WHERE run_id = ? AND event = 'ToolResult' + AND json_extract(data_json, '$.success') = 0 + """, + (run_id,), + ).fetchone()[0] + ) + errors = ( + counts.get(AgentEventType.run_failed.value, 0) + + counts.get(AgentEventType.model_call_failed.value, 0) + + tool_errors + ) + duration_ms = max( + 0, int((run.updated_at - run.created_at).total_seconds() * 1000) + ) + return AgentTraceResponse( + run_id=run_id, + status=run.status, + items=items, + next_sequence=items[-1].sequence if items else after_sequence, + has_more=has_more, + summary=AgentTraceSummary( + model_calls=counts.get(AgentEventType.model_call_started.value, 0), + tool_calls=counts.get(AgentEventType.tool_call.value, 0), + duration_ms=duration_ms, + token_usage=run.token_usage, + errors=errors, + ), + config_snapshot=json.loads(row["config_snapshot_json"]), + ) + finally: + conn.close() + + def recover_interrupted(self, run_id: str) -> AgentRun | None: + """把上个进程遗留的非终态 Run 收束为失败，并追加可回放终止事件。""" + + conn = connect() + try: + with transaction(conn): + row = conn.execute( + "SELECT run_json, status FROM agent_runs WHERE run_id = ?", (run_id,) + ).fetchone() + if row is None: + return None + run = AgentRun.model_validate_json(row["run_json"]) + if row["status"] in _TERMINAL_VALUES: + return run + run.status = AgentRunStatus.failed + run.error_code = "AGENT_PROCESS_RESTARTED" + run.error_message = "Agent process restarted before the run completed." + run.updated_at = datetime.now(timezone.utc) + next_sequence = int( + conn.execute( + """ + SELECT COALESCE(MAX(sequence), -1) + 1 + FROM agent_events WHERE run_id = ? + """, + (run_id,), + ).fetchone()[0] + ) + event = AgentEvent( + event=AgentEventType.run_failed, + run_id=run_id, + sequence=next_sequence, + data={ + "code": run.error_code, + "message": run.error_message, + }, + timestamp=run.updated_at, + ) + self._update_run(conn, run) + conn.execute( + """ + INSERT INTO agent_events(run_id, sequence, event, data_json, timestamp) + VALUES (?, ?, ?, ?, ?) + """, + ( + run_id, + next_sequence, + event.event.value, + json.dumps(event.data, ensure_ascii=False), + event.timestamp.isoformat(), + ), + ) + return run + finally: + conn.close() + + @staticmethod + def _update_run(conn, run: AgentRun) -> None: + cursor = conn.execute( + """ + UPDATE agent_runs + SET status = ?, run_json = ?, updated_at = ? + WHERE run_id = ? + """, + ( + run.status.value, + AgentTraceRepository._serialize_run(run), + run.updated_at.isoformat(), + run.run_id, + ), + ) + if cursor.rowcount != 1: + raise LookupError(run.run_id) + + @staticmethod + def _event_from_row(run_id: str, row) -> AgentEvent: + return AgentEvent( + event=AgentEventType(row["event"]), + run_id=run_id, + sequence=int(row["sequence"]), + data=json.loads(row["data_json"]), + timestamp=datetime.fromisoformat(row["timestamp"]), + ) + + @staticmethod + def _serialize_run(run: AgentRun) -> str: + return json.dumps( + sanitize_trace_value(run.model_dump(mode="json")), ensure_ascii=False + ) diff --git a/backend/app/contracts.py b/backend/app/contracts.py index e7c9be8..bb8bf90 100644 --- a/backend/app/contracts.py +++ b/backend/app/contracts.py @@ -329,6 +329,10 @@ class AgentEventType(str, Enum): permission_required = "PermissionRequired" usage = "Usage" citation = "Citation" + model_call_started = "ModelCallStarted" + model_call_completed = "ModelCallCompleted" + model_call_failed = "ModelCallFailed" + permission_resolved = "PermissionResolved" run_completed = "RunCompleted" run_failed = "RunFailed" run_cancelled = "RunCancelled" @@ -342,6 +346,24 @@ class AgentEvent(Contract): timestamp: datetime +class AgentTraceSummary(Contract): + model_calls: int = 0 + tool_calls: int = 0 + duration_ms: int = 0 + token_usage: int = 0 + errors: int = 0 + + +class AgentTraceResponse(Contract): + run_id: str + status: AgentRunStatus + items: list[AgentEvent] = Field(default_factory=list) + next_sequence: int + has_more: bool = False + summary: AgentTraceSummary = Field(default_factory=AgentTraceSummary) + config_snapshot: dict[str, Any] = Field(default_factory=dict) + + class PermissionDecisionRequest(Contract): decision: Literal["allow_once", "allow_session", "deny"] diff --git a/backend/app/database/migrations.py b/backend/app/database/migrations.py index 771e69a..02cdb33 100644 --- a/backend/app/database/migrations.py +++ b/backend/app/database/migrations.py @@ -69,6 +69,33 @@ MIGRATIONS: list[str] = [ ); CREATE INDEX IF NOT EXISTS idx_tasks_status_due ON tasks(status, due_at); """, + # v3: 第二阶段 Agent Trace；Run 与事件事实持久化，供 SSE 恢复和 Benchmark 复用。 + """ + CREATE TABLE IF NOT EXISTS agent_runs ( + run_id TEXT PRIMARY KEY, + status TEXT NOT NULL, + run_json TEXT NOT NULL, + request_json TEXT NOT NULL, + config_snapshot_json TEXT NOT NULL DEFAULT '{}', + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_agent_runs_created + ON agent_runs(created_at DESC); + CREATE INDEX IF NOT EXISTS idx_agent_runs_status + ON agent_runs(status, updated_at DESC); + + CREATE TABLE IF NOT EXISTS agent_events ( + run_id TEXT NOT NULL REFERENCES agent_runs(run_id) ON DELETE CASCADE, + sequence INTEGER NOT NULL, + event TEXT NOT NULL, + data_json TEXT NOT NULL DEFAULT '{}', + timestamp TEXT NOT NULL, + PRIMARY KEY (run_id, sequence) + ); + CREATE INDEX IF NOT EXISTS idx_agent_events_type + ON agent_events(run_id, event, sequence); + """, ] diff --git a/backend/app/routes.py b/backend/app/routes.py index 31ff060..e5e2d07 100644 --- a/backend/app/routes.py +++ b/backend/app/routes.py @@ -2,13 +2,14 @@ from collections.abc import AsyncIterator from datetime import datetime, timezone from uuid import uuid4 -from fastapi import APIRouter, Query +from fastapi import APIRouter, Header, Query from fastapi.responses import StreamingResponse from app.contracts import ( AgentRun, AgentRunCreateRequest, AgentRunListResponse, + AgentTraceResponse, ChatRequest, CredentialStatus, CredentialWriteRequest, @@ -81,8 +82,9 @@ def utc_now() -> datetime: return datetime.now(timezone.utc) -def as_sse(event: str, payload: str) -> str: - return f"event: {event}\ndata: {payload}\n\n" +def as_sse(event: str, payload: str, *, event_id: int | None = None) -> str: + id_line = f"id: {event_id}\n" if event_id is not None else "" + return f"{id_line}event: {event}\ndata: {payload}\n\n" def provider_or_404(provider_id: str): @@ -314,16 +316,65 @@ async def cancel_agent_run(run_id: str) -> OperationResponse: }, tags=["Agent"], ) -async def agent_events(run_id: str) -> StreamingResponse: +async def agent_events( + run_id: str, + after_sequence: int | None = Query(default=None, ge=-1), + last_event_id: str | None = Header(default=None, alias="Last-Event-ID"), +) -> StreamingResponse: agent_run_or_404(run_id) + cursor = after_sequence + if cursor is None and last_event_id is not None: + try: + cursor = int(last_event_id) + except ValueError as exc: + raise ApiError( + 400, + "TRACE_CURSOR_INVALID", + "Last-Event-ID must be an integer sequence.", + {"last_event_id": last_event_id}, + ) from exc + if cursor < -1: + raise ApiError( + 400, + "TRACE_CURSOR_INVALID", + "Last-Event-ID must be greater than or equal to -1.", + ) + cursor = cursor if cursor is not None else -1 async def stream() -> AsyncIterator[str]: - async for event in container.agent.events(run_id): - yield as_sse(event.event.value, event.model_dump_json()) + async for event in container.agent.events(run_id, after_sequence=cursor): + yield as_sse( + event.event.value, + event.model_dump_json(), + event_id=event.sequence, + ) return StreamingResponse(stream(), media_type="text/event-stream") +@router.get( + "/agent/runs/{run_id}/trace", + response_model=AgentTraceResponse, + tags=["Agent"], +) +async def get_agent_trace( + run_id: str, + after_sequence: int = Query(default=-1, ge=-1), + limit: int = Query(default=200, ge=1, le=500), +) -> AgentTraceResponse: + try: + return container.agent.get_trace( + run_id, after_sequence=after_sequence, limit=limit + ) + except AgentRunNotFoundError as exc: + raise ApiError( + 404, + "AGENT_RUN_NOT_FOUND", + f"Agent run does not exist: {run_id}", + {"run_id": run_id}, + ) from exc + + @router.post( "/agent/runs/{run_id}/permissions/{request_id}", response_model=OperationResponse, diff --git a/backend/tests/test_agent_core.py b/backend/tests/test_agent_core.py index 0aa843b..4901bb4 100644 --- a/backend/tests/test_agent_core.py +++ b/backend/tests/test_agent_core.py @@ -1,10 +1,18 @@ import asyncio +from datetime import datetime, timezone +import pytest + +from app.agent.trace_repository import AgentTraceRepository from app.agent.permissions import PermissionMode from app.agent.tools import ToolExecutionContext from app.container import build_container +from app.database.db import connect +from app.errors import ApiError +from app.routes import agent_events from app.contracts import ( AgentEventType, + AgentRun, AgentRunCreateRequest, AgentRunStatus, ToolCall, @@ -108,8 +116,168 @@ def test_permission_confirmation_resumes_agent() -> None: created.run_id, request_id, "allow_once" ) completed = await container.agent.wait(created.run_id) + events = [event async for event in container.agent.events(created.run_id)] assert completed.status == AgentRunStatus.completed assert completed.tool_results[0].success is True + assert AgentEventType.permission_resolved in {event.event for event in events} + + run(scenario()) + + +def test_agent_trace_persists_and_replays_from_sequence() -> None: + async def scenario() -> None: + first = build_container() + created = await first.agent.create_run( + AgentRunCreateRequest( + input="persistent trace", + provider_id="mock", + model="mock-1", + metadata={"suite": "agent-benchmark-v1"}, + ) + ) + completed = await first.agent.wait(created.run_id) + + restarted = build_container() + restored = restarted.agent.get_run(created.run_id) + first_page = restarted.agent.get_trace( + created.run_id, after_sequence=-1, limit=2 + ) + second_page = restarted.agent.get_trace( + created.run_id, + after_sequence=first_page.next_sequence, + limit=100, + ) + replay = [ + event + async for event in restarted.agent.events( + created.run_id, after_sequence=first_page.next_sequence + ) + ] + + assert completed.status == restored.status == AgentRunStatus.completed + assert first_page.has_more is True + assert [item.sequence for item in first_page.items] == [0, 1] + assert second_page.items[0].sequence == 2 + assert replay == second_page.items + assert first_page.summary.model_calls == 1 + assert first_page.summary.token_usage == completed.token_usage + assert first_page.config_snapshot["metadata"] == { + "suite": "agent-benchmark-v1" + } + assert second_page.items[-1].event == AgentEventType.run_completed + + run(scenario()) + + +def test_interrupted_persisted_run_is_closed_after_restart() -> None: + now = datetime.now(timezone.utc) + request = AgentRunCreateRequest( + input="interrupted", + provider_id="mock", + model="mock-1", + ) + persisted = AgentRun( + run_id="run_interrupted", + status=AgentRunStatus.running, + input=request.input, + provider_id=request.provider_id, + model=request.model, + max_steps=request.max_steps, + created_at=now, + updated_at=now, + ) + AgentTraceRepository().create_run(persisted, request, {"model": "mock-1"}) + + restarted = build_container() + recovered = restarted.agent.get_run(persisted.run_id) + events = run( + _collect_events(restarted.agent.events(persisted.run_id, after_sequence=-1)) + ) + + assert recovered.status == AgentRunStatus.failed + assert recovered.error_code == "AGENT_PROCESS_RESTARTED" + assert events[-1].event == AgentEventType.run_failed + assert events[-1].sequence == 0 + + +def test_trace_redacts_secrets_and_truncates_large_values() -> None: + async def scenario() -> None: + container = build_container() + secret = "sk-should-not-be-stored" + created = await container.agent.create_run( + AgentRunCreateRequest( + input=f'/tool system.echo {{"text":"{"x" * 4200}","api_key":"{secret}"}}', + provider_id="mock", + model="mock-1", + allowed_tools=["system.echo"], + metadata={"authorization": secret}, + ) + ) + await container.agent.wait(created.run_id) + trace = container.agent.get_trace( + created.run_id, after_sequence=-1, limit=100 + ) + tool_call = next( + item for item in trace.items if item.event == AgentEventType.tool_call + ) + + assert tool_call.data["arguments"]["api_key"] == "[REDACTED]" + assert str(tool_call.data["arguments"]["text"]).endswith("...[TRUNCATED]") + assert trace.config_snapshot["metadata"]["authorization"] == "[REDACTED]" + assert secret not in trace.model_dump_json() + conn = connect() + try: + stored_row = conn.execute( + """ + SELECT run_json, request_json, config_snapshot_json + FROM agent_runs WHERE run_id = ? + """, + (created.run_id,), + ).fetchone() + stored = "\n".join(str(value) for value in stored_row) + finally: + conn.close() + assert secret not in stored + + run(scenario()) + + +async def _collect_events(iterator): + return [event async for event in iterator] + + +def test_agent_sse_uses_last_event_id_and_emits_event_ids(monkeypatch) -> None: + async def scenario() -> None: + test_container = build_container() + monkeypatch.setattr("app.routes.container", test_container) + created = await test_container.agent.create_run( + AgentRunCreateRequest( + input="resume sse", + provider_id="mock", + model="mock-1", + ) + ) + await test_container.agent.wait(created.run_id) + + response = await agent_events( + created.run_id, after_sequence=None, last_event_id="1" + ) + chunks = [chunk async for chunk in response.body_iterator] + body = "".join( + chunk.decode("utf-8") if isinstance(chunk, bytes) else chunk + for chunk in chunks + ) + + assert "id: 0\n" not in body + assert "id: 1\n" not in body + assert "id: 2\n" in body + assert "event: RunCompleted" in body + + with pytest.raises(ApiError) as error: + await agent_events( + created.run_id, after_sequence=None, last_event_id="invalid" + ) + assert error.value.code == "TRACE_CURSOR_INVALID" run(scenario()) diff --git a/backend/tests/test_api.py b/backend/tests/test_api.py index d7d76c4..f62496e 100644 --- a/backend/tests/test_api.py +++ b/backend/tests/test_api.py @@ -89,6 +89,7 @@ def test_openapi_contains_documented_frontend_interfaces() -> None: "/api/agent/runs", "/api/agent/runs/{run_id}/cancel", "/api/agent/runs/{run_id}/events", + "/api/agent/runs/{run_id}/trace", "/api/skills", "/api/plugins", "/api/plugins/install", diff --git a/docs/AI-Core与Agent-Core开发说明.md b/docs/AI-Core与Agent-Core开发说明.md index 34f042c..077d066 100644 --- a/docs/AI-Core与Agent-Core开发说明.md +++ b/docs/AI-Core与Agent-Core开发说明.md @@ -2,7 +2,7 @@ > 本文档用于团队开发和模块联调，记录当前已经落地的核心边界与使用方式。 -> 更新日期：2026-08-30。第一阶段 AI Core、Agent Core、Extension Core 和 Model Core 主链路已经完成，后端当前回归基线为 71 项测试通过。 +> 更新日期：2026-09-01。第一阶段 AI Core、Agent Core、Extension Core 和 Model Core 主链路已经完成；第二阶段 Agent Trace 持久化和可恢复 SSE 已落地，后端当前回归基线为 80 项测试通过。 ## 当前实现 @@ -28,6 +28,7 @@ backend/app/ │ └── mock.py 离线开发 Provider ├── agent/ │ ├── runtime.py Agent Loop、限制、取消、Trace 和 SSE +│ ├── trace_repository.py Run/Event SQLite 持久化、分页、摘要与脱敏 │ ├── tools.py Tool 注册、参数校验、隔离执行和结果转换 │ ├── permissions.py 权限策略、确认请求和会话授权 │ └── builtin_tools.py 无副作用的内置开发 Tool @@ -53,7 +54,7 @@ Router 只负责 HTTP/SSE 与错误转换，不实现 Agent、Tool 或 Provider - Permission； - Step、Timeout、Token Budget、取消； - Tool 并发上限与 run 级网络权限； -- 内存 Trace 与 SSE； +- SQLite Trace、分页快照与可恢复 SSE； - Skill Manifest、Prompt、Tool/Permission/模型能力解析； - Plugin Manifest、生命周期和 Tool Contribution； - Skill 调用内置 Tool 与 Plugin Tool； @@ -171,10 +172,13 @@ POST /api/agent/runs ```text GET /api/agent/runs/{run_id} GET /api/agent/runs/{run_id}/events +GET /api/agent/runs/{run_id}/trace?after_sequence=-1&limit=200 POST /api/agent/runs/{run_id}/cancel ``` -当前 Run 与 Trace 保存在内存中，AI Core 重启后清空。Runtime 最多保留 200 个 Run，每个 Run 最多保留 2000 个事件，并限制单轮 Tool Call 数量，避免长时间运行时无界增长。后续数据库层接入时替换 Repository，不改变 API Contract。 +Run 与 AgentEvent 已写入 SQLite，`run_id + sequence` 是幂等键。SSE 每帧包含 `id: sequence`；客户端可以通过 `Last-Event-ID` 请求头或 `after_sequence` 查询参数恢复缺失事件。Trace API 返回平铺事件、下一游标、分页状态、模型/工具调用统计、耗时、Token Usage 和创建 Run 时的配置快照，不负责生成前端树形布局。 + +运行时内存仍只保留最近 2000 个事件用于实时订阅，完整 Trace 以 SQLite 为准。AI Core 重启后，已经终止的 Run 可以继续查询和回放；重启前未终止的 Run 会收束为 `AGENT_PROCESS_RESTARTED`，避免永久停在 `running`。API Key、Authorization、Password、Secret、常见 `sk-`/Bearer 值在入库前脱敏，超长字符串和集合会截断。 ## Tool Calling @@ -333,7 +337,7 @@ Skill Manifest - 已实现 Mock、OpenAI-Compatible Chat Completions 与 Ollama Adapter；OpenAI Responses 和 Anthropic Messages 尚未实现。 - Provider 配置暂存内存，后续通过 Repository 接入 SQLite；PATCH 已支持用显式 `null` 清空 base URL、默认模型和凭据引用。 -- Run/Trace 暂存内存；下一步抽象 Repository 并接入 SQLite。 +- Run/Trace 已通过 Repository 接入 SQLite；后续增加按保留策略归档和 Benchmark 引用保护。 - Permission 已有核心等待/恢复机制，前端确认 UI 已完成联调和中文展示。 - Task 已持久化到 SQLite；Attachment Tool 读取 Host 管理目录中的 UTF-8 文件。 - `audio.transcribe` 当前消费 Host 预生成的 transcript；faster-whisper 与说话人分离仍按技术基线在第二阶段接入。 diff --git a/docs/AI笔记软件技术栈说明-团队版-v2.3.md b/docs/AI笔记软件技术栈说明-团队版-v2.3.md index b0a2c9d..5cade4b 100644 --- a/docs/AI笔记软件技术栈说明-团队版-v2.3.md +++ b/docs/AI笔记软件技术栈说明-团队版-v2.3.md @@ -5,7 +5,7 @@ > 适用范围：桌面客户端、本地知识库、RAG、Agent、Skill、多模型接入、多模态处理与可选云同步 > 目标读者：前端、Rust 桌面端、Python AI Core、算法、测试与后续接手项目的开发成员 -> 实施状态更新：2026-08-31。本文同时包含目标架构、当前实现和第二阶段接口基线。第一阶段已完成 Vue Web 联调前端、FastAPI、Knowledge/Retrieval、Agent/Tool/Permission、Skill/Plugin 声明式运行时、Mock/OpenAI-Compatible/Ollama Provider、DeepSeek/OpenAI 预设、模型发现及开发阶段 Fernet 凭据存储。Web Workspace 已通过 FastAPI 接入后端配置的真实单 Vault。第二阶段在现有边界上接入真实音频处理、MCP、Plugin Command/Settings、Provider 协议增强、Benchmark、文档导出、主题包、Agent Trace、Mermaid 和函数图像。Tauri/Rust Host、Stronghold、原生多 Vault 文件系统和 Sync Server 仍未实现。 +> 实施状态更新：2026-09-01。本文同时包含目标架构、当前实现和第二阶段接口基线。第一阶段已完成 Vue Web 联调前端、FastAPI、Knowledge/Retrieval、Agent/Tool/Permission、Skill/Plugin 声明式运行时、Mock/OpenAI-Compatible/Ollama Provider、DeepSeek/OpenAI 预设、模型发现及开发阶段 Fernet 凭据存储。Web Workspace 已通过 FastAPI 接入后端配置的真实单 Vault，第二阶段 Agent Trace 持久化、分页快照和可恢复 SSE 已完成。后续继续接入真实音频处理、MCP、Plugin Command/Settings、Provider 协议增强、Benchmark、文档导出、主题包、Trace 可视化、Mermaid 和函数图像。Tauri/Rust Host、Stronghold、原生多 Vault 文件系统和 Sync Server 仍未实现。 --- @@ -545,7 +545,7 @@ index_jobs sync_state ``` -其中 `notes` 和 `blocks` 保存 Markdown 的结构化投影；FTS5 建立全文索引；sqlite-vec 保存 Block 向量；`agent_runs` 和 `tool_calls` 保存 Agent Trace；Provider 表保存非敏感模型配置。 +其中 `notes` 和 `blocks` 保存 Markdown 的结构化投影；FTS5 建立全文索引；sqlite-vec 保存 Block 向量；当前实现以 `agent_runs` 和 `agent_events` 保存 Agent Trace；Provider 表保存非敏感模型配置。 API Key、同步 Token 等机密数据不进入 SQLite，通过 `credential_id` 与 Stronghold 中的实际密钥关联。 @@ -773,7 +773,7 @@ Agent Run 至少提供以下限制： - 网络访问权限； - 并发 Tool 数量。 -Agent 运行过程中产生的每一步写入 `agent_runs` 和 `tool_calls`，用户可以在 Agent Trace 中查看工具名称、参数摘要、耗时、执行结果和权限状态。 +Agent 运行过程中产生的每一步写入 `agent_runs` 和 `agent_events`，用户可以在 Agent Trace 中查看工具名称、参数摘要、耗时、执行结果和权限状态。SSE 与 Benchmark 均从同一事件事实读取，不维护旁路数据。 ### 10.3 Tool Registry @@ -2321,7 +2321,7 @@ Markdown Workspace 第一阶段 Plugin Runtime 已完成安装、启用、停用、权限和声明式 Tool 注册，建立 Skill 调用 Plugin Tool 的基础链路。Command、Settings 和 MCP 执行不计入第一阶段完成项。 -截至 2026-08-31，上述第一阶段后端链路和 Web 联调前端均已完成，第二阶段前置的 Workspace 去 Mock 联调也已完成。当前验证基线为后端 76 项测试、前端 26 项测试及生产构建通过。向量链路当前使用 `HashEmbeddingProvider` 验证工程正确性，真实 Embedding 召回质量不属于该测试结论。 +截至 2026-09-01，上述第一阶段后端链路和 Web 联调前端均已完成，第二阶段前置的 Workspace 去 Mock 联调及 Agent Trace 持久化/恢复接口也已完成。当前验证基线为后端 80 项测试、前端 27 项测试及生产构建通过。向量链路当前使用 `HashEmbeddingProvider` 验证工程正确性，真实 Embedding 召回质量不属于该测试结论。 第二阶段在既有 Contract 上接入： @@ -2405,7 +2405,7 @@ Sync Server 按独立服务开发和部署，不进入桌面客户端核心启 ## 25. 当前技术基线摘要 -目标桌面端采用 Tauri 2、Rust、Vue 3 和 TypeScript；当前可运行形态是 Vue/Vite Web 前端加 FastAPI。用户笔记以 Markdown 和 Assets 保存在本地 Vault，SQLite 已管理笔记元数据、全文索引、向量索引和任务；Agent Trace 与 Provider/Extension Registry 当前仍为内存实现。 +目标桌面端采用 Tauri 2、Rust、Vue 3 和 TypeScript；当前可运行形态是 Vue/Vite Web 前端加 FastAPI。用户笔记以 Markdown 和 Assets 保存在本地 Vault，SQLite 已管理笔记元数据、全文索引、向量索引、任务及 Agent Trace；Provider/Extension Registry 当前仍为内存实现。 Python AI Core 未来作为 Tauri Sidecar 运行，当前由开发命令独立启动，FastAPI 提供本地接口。Knowledge Core 管理笔记结构；Retrieval Core 当前通过 FTS5、`HashEmbeddingProvider`、sqlite-vec、RRF 和轻量 Reranker 跑通混合检索，真实 Embedding 与正式 Benchmark 在第二阶段接入；Agent Runtime 使用 Tool Registry 操作知识库和任务，并将扩展 Agent Trace Contract 供可视化和 Benchmark 共用；Skill Runtime 将提示词、工具、权限和检索参数组装为可复用 Agent 配置。 diff --git a/docs/前端壳子与接口层开发说明.md b/docs/前端壳子与接口层开发说明.md index c1be39c..4ca1eb9 100644 --- a/docs/前端壳子与接口层开发说明.md +++ b/docs/前端壳子与接口层开发说明.md @@ -186,13 +186,13 @@ pnpm build ```text pnpm build passed -pnpm test 26 passed -uv run pytest 76 passed +pnpm test 27 passed +uv run pytest 80 passed preview smoke HTTP 200 git diff --check passed ``` -当前前端使用 Vitest 执行 Store、Workspace API Adapter、文件树、编辑器组件、智能体标签、轻量动效约束、Markdown 对比度 Token、scoped CSS 选择器约束和 Shiki GitHub 双主题测试；`pnpm build` 同时执行 `vue-tsc -b` 与 Vite 生产构建。后端测试出现过 `.pytest_cache` 无法写入的 Windows 权限警告，不影响 76 项测试结果，也不涉及产品代码。 +当前前端使用 Vitest 执行 Store、Workspace API Adapter、SSE 恢复游标、文件树、编辑器组件、智能体标签、轻量动效约束、Markdown 对比度 Token、scoped CSS 选择器约束和 Shiki GitHub 双主题测试；`pnpm build` 同时执行 `vue-tsc -b` 与 Vite 生产构建。后端测试出现过 `.pytest_cache` 无法写入的 Windows 权限警告，不影响 80 项测试结果，也不涉及产品代码。 Vite 当前会提示 Chat 与 Workspace 的部分异步 Chunk 超过 500 kB，这是 Milkdown、CodeMirror、KaTeX 和 Shiki 等编辑/渲染依赖带来的性能优化项，不影响构建成功或功能正确性；进入桌面打包前应通过手动分包或更细粒度动态加载继续优化。 diff --git a/docs/后端接口契约-开发版.md b/docs/后端接口契约-开发版.md index a12d81a..1ed8943 100644 --- a/docs/后端接口契约-开发版.md +++ b/docs/后端接口契约-开发版.md @@ -55,7 +55,8 @@ Web 联调阶段只暴露后端通过 `APP_VAULT_PATH` 配置的单一 Vault， | POST | `/api/agent/runs` | 创建 Agent Run | | GET | `/api/agent/runs/{run_id}` | 获取 Agent Run 状态与 Trace 摘要 | | POST | `/api/agent/runs/{run_id}/cancel` | 取消 Agent Run | -| GET | `/api/agent/runs/{run_id}/events` | 订阅 AgentEvent SSE | +| GET | `/api/agent/runs/{run_id}/events` | 订阅 AgentEvent SSE，支持 `Last-Event-ID` / `after_sequence` 恢复 | +| GET | `/api/agent/runs/{run_id}/trace` | 分页读取持久化 Trace、摘要和运行配置快照 | | POST | `/api/agent/runs/{run_id}/permissions/{request_id}` | 响应 Tool 权限确认 | | GET | `/api/tools` | 获取已注册 Tool Definition | @@ -173,9 +174,10 @@ RunCancelled ## 当前实现状态 -更新至 2026-08-31：后端 76 项回归测试通过。 +更新至 2026-09-01：后端 80 项回归测试通过。 - Chat、Agent Run、Agent Events、Tool 列表、Provider 配置生命周期、模型列表和连接测试已经接入 AI Core。 +- Agent Run/Event 已持久化到 SQLite；SSE 帧携带 sequence `id`，断线后可以回放缺失事件。Trace API 与 Benchmark 共用同一事件事实，并在入库前执行 Secret 脱敏和结果限长。 - Provider Adapter 当前包含 Mock、真正增量 SSE 的 OpenAI-Compatible Chat Completions，以及 Ollama JSONL Streaming。 - Notes、Search、Index、Skills、Plugins、Tasks 和 Provider 生命周期均已接入业务服务。 - Workspace 已接入后端配置的真实 Vault；文件树、笔记读写、文件/目录新建、重命名和删除不再使用前端 Mock Fallback。 diff --git a/docs/第一阶段测试验证操作手册.md b/docs/第一阶段测试验证操作手册.md index 684cde9..a18ec82 100644 --- a/docs/第一阶段测试验证操作手册.md +++ b/docs/第一阶段测试验证操作手册.md @@ -68,7 +68,7 @@ uv run pytest -q -p no:cacheprovider 当前基线： ```text -76 passed +80 passed ``` 通过标准：退出码为 0、失败数为 0。用例数可以随功能增加，但不得低于当前基线。 @@ -83,8 +83,8 @@ pnpm test 当前基线： ```text -10 test files passed -26 tests passed +11 test files passed +27 tests passed ``` 通过标准：退出码为 0、失败数为 0。测试覆盖 Provider Store、主题偏好、Workspace、文件树、文件切换、可视化编辑器、智能体中文标签、轻量动效性能约束、Markdown 对比度 Token、scoped CSS 选择器约束和 Shiki GitHub 双主题输出。 diff --git a/docs/第二阶段接口契约-开发版.md b/docs/第二阶段接口契约-开发版.md index 21168b2..4da9867 100644 --- a/docs/第二阶段接口契约-开发版.md +++ b/docs/第二阶段接口契约-开发版.md @@ -43,8 +43,8 @@ | Transcription | GET | `/api/media/transcriptions/{job_id}/events` | 计划新增 | 订阅模型加载、分离和转写进度 | | Transcription | POST | `/api/media/transcriptions/{job_id}/cancel` | 计划新增 | 取消音频任务 | | Transcription | POST | `/api/media/transcriptions/{job_id}/notes` | 计划新增 | 将 Transcript 写入 Knowledge Core | -| Agent Trace | GET | `/api/agent/runs/{run_id}/events` | 扩展 | 支持游标恢复并增加模型与权限事件 | -| Agent Trace | GET | `/api/agent/runs/{run_id}/trace` | 计划新增 | 分页读取可回放 Trace 快照 | +| Agent Trace | GET | `/api/agent/runs/{run_id}/events` | 已实现 | 支持游标恢复并增加模型与权限事件 | +| Agent Trace | GET | `/api/agent/runs/{run_id}/trace` | 已实现 | 分页读取可回放 Trace 快照 | | Plugin Host | GET | `/api/plugins/{plugin_id}/host` | 计划新增 | 获取 MCP Host 健康状态 | | Plugin Host | POST | `/api/plugins/{plugin_id}/host/restart` | 计划新增 | 重启异常 Host 并重新发现 Tool | | Plugin Command | GET | `/api/plugin-contributions/commands` | 计划新增 | 获取前端可展示的 Command | @@ -310,6 +310,7 @@ RunCancelled ```text ModelCallStarted ModelCallCompleted +ModelCallFailed PermissionResolved ``` @@ -330,8 +331,10 @@ PermissionResolved "model_calls": 2, "tool_calls": 3, "duration_ms": 1530, - "token_usage": 2048 - } + "token_usage": 2048, + "errors": 0 + }, + "config_snapshot": {} } ``` @@ -388,6 +391,7 @@ ToolCall 和 ToolResult 增加可选 `parent_model_call_id`、`duration_ms` 和 AGENT_RUN_NOT_FOUND TRACE_NOT_AVAILABLE TRACE_CURSOR_EXPIRED +TRACE_CURSOR_INVALID ``` --- diff --git a/frontend/src/contracts/index.ts b/frontend/src/contracts/index.ts index afcdcde..0fcfcfd 100644 --- a/frontend/src/contracts/index.ts +++ b/frontend/src/contracts/index.ts @@ -143,6 +143,10 @@ export type AgentEventType = | 'PermissionRequired' | 'Usage' | 'Citation' + | 'ModelCallStarted' + | 'ModelCallCompleted' + | 'ModelCallFailed' + | 'PermissionResolved' | 'RunCompleted' | 'RunFailed' | 'RunCancelled' @@ -155,6 +159,24 @@ export interface AgentEvent { timestamp: string } +export interface AgentTraceSummary { + model_calls: number + tool_calls: number + duration_ms: number + token_usage: number + errors: number +} + +export interface AgentTraceResponse { + run_id: string + status: AgentRunStatus + items: AgentEvent[] + next_sequence: number + has_more: boolean + summary: AgentTraceSummary + config_snapshot: Record +} + export interface ToolCall { tool_call_id: string name: string diff --git a/frontend/src/features/agent/labels.ts b/frontend/src/features/agent/labels.ts index 871e8d4..c6646d6 100644 --- a/frontend/src/features/agent/labels.ts +++ b/frontend/src/features/agent/labels.ts @@ -18,6 +18,10 @@ const eventLabels: Record = { PermissionRequired: '请求权限', Usage: '用量统计', Citation: '引用来源', + ModelCallStarted: '模型调用开始', + ModelCallCompleted: '模型调用完成', + ModelCallFailed: '模型调用失败', + PermissionResolved: '权限已处理', RunCompleted: '运行完成', RunFailed: '运行失败', RunCancelled: '运行取消', @@ -86,6 +90,10 @@ const detailLabels: Record = { total_tokens: '令牌总数', status: '状态', duration_ms: '耗时（毫秒）', + model_call_id: '模型调用 ID', + parent_model_call_id: '上级模型调用 ID', + finish_reason: '结束原因', + decision: '授权决定', } export function runStatusLabel(status?: AgentRunStatus): string { diff --git a/frontend/src/services/agentService.ts b/frontend/src/services/agentService.ts index d6a5f41..5c95728 100644 --- a/frontend/src/services/agentService.ts +++ b/frontend/src/services/agentService.ts @@ -1,6 +1,6 @@ import apiClient from './apiClient' import { SseClient } from './sseClient' -import type { AgentRun, AgentEvent, ApiAgentRun, OperationResponse, PageMeta, ToolDefinition, PermissionRequest } from '@/contracts' +import type { AgentRun, AgentEvent, AgentTraceResponse, ApiAgentRun, OperationResponse, PageMeta, ToolDefinition, PermissionRequest } from '@/contracts' function toAgentRun(run: ApiAgentRun): AgentRun { // API 的 token_usage 是累计值，UI 模型预留了输入/输出拆分字段。 @@ -54,6 +54,13 @@ export async function cancelAgentRun(runId: string): Promise return apiClient.post(`/api/agent/runs/${runId}/cancel`) } +export async function getAgentTrace( + runId: string, + params?: { after_sequence?: number; limit?: number }, +): Promise { + return apiClient.get(`/api/agent/runs/${runId}/trace`, { params }) +} + export async function listTools(): Promise { const response = await apiClient.get<{ items: ToolDefinition[] }>('/api/tools') return response.items @@ -66,12 +73,14 @@ export function streamAgentEvents( onError?: (error: Error) => void onDone?: () => void onOpen?: () => void - } + }, + afterSequence = -1, ): SseClient { // 将通用 SSE 包装成领域事件，Store 无需了解传输层 envelope。 const client = new SseClient({ - url: `/api/agent/runs/${runId}/events`, + url: `/api/agent/runs/${runId}/events?after_sequence=${afterSequence}`, method: 'GET', + lastEventId: afterSequence >= 0 ? String(afterSequence) : undefined, onEvent: (eventName, data) => { handlers.onEvent?.({ event: eventName as AgentEvent['event'], diff --git a/frontend/src/services/sseClient.spec.ts b/frontend/src/services/sseClient.spec.ts new file mode 100644 index 0000000..ba4fa7a --- /dev/null +++ b/frontend/src/services/sseClient.spec.ts @@ -0,0 +1,42 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' + +import { SseClient } from './sseClient' + +afterEach(() => { + vi.unstubAllGlobals() + vi.restoreAllMocks() +}) + +describe('SseClient resumable event transport', () => { + it('sends Last-Event-ID and exposes the returned SSE id', async () => { + const fetchMock = vi.fn().mockResolvedValue( + new Response( + 'id: 3\nevent: ModelCallCompleted\ndata: {"sequence":3,"data":{"duration_ms":12}}\n\n', + { status: 200, headers: { 'Content-Type': 'text/event-stream' } }, + ), + ) + vi.stubGlobal('fetch', fetchMock) + const received = vi.fn() + const client = new SseClient({ + url: '/api/agent/runs/run-1/events?after_sequence=2', + method: 'GET', + lastEventId: '2', + onEvent: received, + }) + + await client.connect() + + expect(fetchMock).toHaveBeenCalledWith( + '/api/agent/runs/run-1/events?after_sequence=2', + expect.objectContaining({ + method: 'GET', + headers: expect.objectContaining({ 'Last-Event-ID': '2' }), + }), + ) + expect(received).toHaveBeenCalledWith( + 'ModelCallCompleted', + { sequence: 3, data: { duration_ms: 12 } }, + '3', + ) + }) +}) diff --git a/frontend/src/services/sseClient.ts b/frontend/src/services/sseClient.ts index 30389b1..34db218 100644 --- a/frontend/src/services/sseClient.ts +++ b/frontend/src/services/sseClient.ts @@ -1,12 +1,17 @@ import { resolveApiUrl } from './apiClient' -export type SseEventHandler = (event: string, data: Record) => void +export type SseEventHandler = ( + event: string, + data: Record, + eventId?: string, +) => void export interface SseClientOptions { url: string method?: string body?: unknown token?: string + lastEventId?: string onEvent?: SseEventHandler onError?: (error: Error) => void onOpen?: () => void @@ -26,7 +31,7 @@ export class SseClient { } async connect() { - const { url, method = 'POST', body, token, onEvent, onError, onOpen, onDone } = this.options + const { url, method = 'POST', body, token, lastEventId, onEvent, onError, onOpen, onDone } = this.options try { const headers: Record = { @@ -38,6 +43,9 @@ export class SseClient { if (token) { headers['Authorization'] = `Bearer ${token}` } + if (lastEventId !== undefined) { + headers['Last-Event-ID'] = lastEventId + } const resp = await fetch(resolveApiUrl(url), { method, @@ -57,17 +65,19 @@ export class SseClient { // 一个 UTF-8 字符或 SSE 行可能横跨多个网络分片，必须累积后再按空行派发。 const decoder = new TextDecoder('utf-8') let eventName = 'message' + let eventId: string | undefined let dataLines: string[] = [] let doneNotified = false const dispatchEvent = () => { if (!dataLines.length) { eventName = 'message' + eventId = undefined return } try { const data = JSON.parse(dataLines.join('\n')) as Record - onEvent?.(eventName, data) + onEvent?.(eventName, data, eventId) if (!doneNotified && ['Done', 'RunCompleted', 'RunFailed', 'RunCancelled'].includes(eventName)) { doneNotified = true onDone?.() @@ -76,6 +86,7 @@ export class SseClient { onError?.(error instanceof Error ? error : new Error('Malformed SSE data')) } eventName = 'message' + eventId = undefined dataLines = [] } @@ -87,6 +98,7 @@ export class SseClient { let fieldValue = separator === -1 ? '' : line.slice(separator + 1) if (fieldValue.startsWith(' ')) fieldValue = fieldValue.slice(1) if (field === 'event') eventName = fieldValue + if (field === 'id') eventId = fieldValue if (field === 'data') dataLines.push(fieldValue) } @@ -118,7 +130,7 @@ export class SseClient { this.controller.abort() } - // TODO(streaming): Agent 事件持久化后，增加 Last-Event-ID 与指数退避重连。 + // TODO(streaming): 桌面网络策略确定后，在 Store 层增加有上限的指数退避重连。 isConnected() { return this.connected