feat(agent): 持久化 Agent Trace 并支持 SSE 断点恢复 #7
@@ -10,7 +10,7 @@
|
||||
NotesAgent/
|
||||
├── frontend/ Vue 3 + TypeScript + Vite 前端
|
||||
├── backend/ FastAPI + Pydantic 后端
|
||||
├── docs/ 分工与技术栈说明
|
||||
├── docs/ 架构、契约、开发说明、协作规范与问题复盘
|
||||
└── server sync/ 云同步服务预留目录，当前未实现
|
||||
```
|
||||
|
||||
@@ -36,7 +36,7 @@ python --version
|
||||
uv --version
|
||||
```
|
||||
|
||||
当前 Web 联调不需要 Rust 和 Tauri。开始桌面端集成后，再按照 `docs/AI笔记软件技术栈说明-团队版-v2.3.md` 安装 Rust Toolchain 与 Tauri CLI。
|
||||
当前 Web 联调不需要 Rust 和 Tauri。开始桌面端集成后，再按照 `docs/architecture/AI笔记软件技术栈说明-团队版-v2.3.md` 安装 Rust Toolchain 与 Tauri CLI。
|
||||
|
||||
## 首次初始化
|
||||
|
||||
@@ -118,7 +118,7 @@ cd frontend
|
||||
pnpm test
|
||||
```
|
||||
|
||||
当前回归基线为后端 76 项测试、前端 26 项测试，且生产构建通过。测试数量会随功能增长，以本地实际输出和 CI 为准。
|
||||
当前回归基线为后端 81 项测试、前端 27 项测试，且生产构建通过。测试数量会随功能增长，以本地实际输出和 CI 为准。
|
||||
|
||||
构建产物位于 `frontend/dist`，该目录不提交到 Git。
|
||||
|
||||
@@ -126,24 +126,15 @@ pnpm test
|
||||
|
||||
| 文档 | 用途 |
|
||||
| --- | --- |
|
||||
| [技术栈说明](docs/AI笔记软件技术栈说明-团队版-v2.3.md) | 目标架构、第二阶段技术边界与模块依赖 |
|
||||
| [第一阶段分工表](docs/第一阶段分工表.md) | 成员职责、协作关系与当前交付状态 |
|
||||
| [第二阶段分工表](docs/第二阶段团队分工表.md) | 第二阶段人员职责、任务顺序、协作关系与验收项 |
|
||||
| [第一阶段测试验证操作手册](docs/第一阶段测试验证操作手册.md) | 自动化测试、接口主链路、前端人工验收与记录模板 |
|
||||
| [后端接口契约](docs/后端接口契约-开发版.md) | HTTP/SSE 接口、错误和当前实现状态 |
|
||||
| [第二阶段接口契约](docs/第二阶段接口契约-开发版.md) | 第二阶段公共 DTO、计划接口、SSE、错误码与联调顺序 |
|
||||
| [AI Core 与 Agent Core](docs/AI-Core与Agent-Core开发说明.md) | Provider、Agent、Tool、Permission 与 Extension Core |
|
||||
| [Knowledge 与 Retrieval Core](docs/Knowledge与Retrieval-Core开发说明.md) | Block、索引、混合检索和 Citation |
|
||||
| [模型提供商与模型发现](docs/模型提供商与模型发现开发说明.md) | Provider 预设、模型发现和凭据边界 |
|
||||
| [前端页面需求](docs/前端页面需求说明-开发版.md) | 页面、交互、状态与验收基线 |
|
||||
| [前端实现说明](docs/前端壳子与接口层开发说明.md) | 当前前端目录、Service、SSE 和运行边界 |
|
||||
| [前端写作体验](docs/前端写作体验优化开发说明.md) | Milkdown、CodeMirror、格式栏和 Shiki |
|
||||
| [前端视觉与轻量动效](docs/前端视觉与轻量动效优化开发说明.md) | Design Token、页面美化、性能边界与主题注入约定 |
|
||||
| [Git 使用细则](docs/Git使用细则-团队开发版.md) | 分支、提交、PR、Review 与合并流程 |
|
||||
| [代码注释与 TODO 约定](docs/代码注释与TODO约定.md) | 注释原则、TODO 格式、领域标签与当前待办索引 |
|
||||
| [后端审阅复盘](docs/后端全面审阅问题与修复复盘.md) | 后端问题原因、后果与修复方案 |
|
||||
| [Knowledge/Retrieval 复盘](docs/Knowledge与Retrieval-Core问题与修复复盘.md) | 检索与事务问题复盘 |
|
||||
| [前端审阅复盘](docs/前端合并审阅问题与修复复盘.md) | 前端工程、契约和交互问题复盘 |
|
||||
| [文档总索引](docs/README.md) | 文档分类、阅读顺序和维护规则 |
|
||||
| [技术栈说明](docs/architecture/AI笔记软件技术栈说明-团队版-v2.3.md) | 目标架构、第二阶段技术边界与模块依赖 |
|
||||
| [第二阶段分工表](docs/architecture/第二阶段团队分工表.md) | 第二阶段人员职责、任务顺序、协作关系与验收项 |
|
||||
| [后端接口契约](docs/contracts/后端接口契约-开发版.md) | HTTP/SSE 接口、错误和当前实现状态 |
|
||||
| [第二阶段接口契约](docs/contracts/第二阶段接口契约-开发版.md) | 第二阶段公共 DTO、计划接口、SSE、错误码与联调顺序 |
|
||||
| [AI Core 与 Agent Core](docs/development/AI-Core与Agent-Core开发说明.md) | Provider、Agent、Tool、Permission 与 Extension Core |
|
||||
| [Git 使用细则](docs/guides/Git使用细则-团队开发版.md) | 分支、提交、PR、Review 与合并流程 |
|
||||
| [CI/CD 细则](docs/guides/CI-CD细则-团队开发版.md) | Gitea 流水线、质量门禁、产物、发布与回滚规则 |
|
||||
| [Agent Trace 复盘](docs/retrospectives/Agent-Core第二阶段问题与修复复盘.md) | Agent 持久化、SSE 恢复、事件契约与脱敏问题复盘 |
|
||||
|
||||
## 日常开发注意事项
|
||||
|
||||
@@ -153,6 +144,7 @@ pnpm test
|
||||
- API 默认监听 `127.0.0.1:8000`，前端默认监听 `127.0.0.1:5173`。
|
||||
- 后端附件目录默认是 `backend/data/attachments`，可通过 `APP_ATTACHMENTS_PATH` 覆盖；该目录由桌面 Host 管理。
|
||||
- 跨模块接口发生变化时，需要同步更新前后端类型和 `docs` 中的接口说明。
|
||||
- 当前已实现接口见 `docs/后端接口契约-开发版.md`，第二阶段规划接口见 `docs/第二阶段接口契约-开发版.md`；已实现能力以 `/openapi.json` 为准。
|
||||
- 前端页面、交互、状态管理和第一阶段验收要求见 `docs/前端页面需求说明-开发版.md`。
|
||||
- 分支、提交、Pull Request、Review 和冲突处理规范见 `docs/Git使用细则-团队开发版.md`。
|
||||
- 当前已实现接口见 `docs/contracts/后端接口契约-开发版.md`，第二阶段规划接口见 `docs/contracts/第二阶段接口契约-开发版.md`；已实现能力以 `/openapi.json` 为准。
|
||||
- 前端页面、交互、状态管理和第一阶段验收要求见 `docs/contracts/前端页面需求说明-开发版.md`。
|
||||
- 分支、提交、Pull Request、Review 和冲突处理规范见 `docs/guides/Git使用细则-团队开发版.md`。
|
||||
- CI 检查、产物、发布和回滚规范见 `docs/guides/CI-CD细则-团队开发版.md`。
|
||||
|
||||
+4
-4
@@ -23,10 +23,10 @@ uv run uvicorn app.main:app --reload --host 127.0.0.1 --port 8000
|
||||
uv run pytest
|
||||
```
|
||||
|
||||
当前基线为 71 项测试通过。Provider API Key 可通过前端设置页写入，也可用 `OPENAI_API_KEY`、`DEEPSEEK_API_KEY` 或 `AINOTE_CREDENTIAL_<ID>` 注入；不要把真实密钥写入仓库。
|
||||
当前基线为 81 项测试通过。Provider API Key 可通过前端设置页写入，也可用 `OPENAI_API_KEY`、`DEEPSEEK_API_KEY` 或 `AINOTE_CREDENTIAL_<ID>` 注入；不要把真实密钥写入仓库。
|
||||
|
||||
团队接口清单见 `../docs/后端接口契约-开发版.md`，机器可读契约以运行时的 `/openapi.json` 为准。
|
||||
团队接口清单见 `../docs/contracts/后端接口契约-开发版.md`，机器可读契约以运行时的 `/openapi.json` 为准。
|
||||
|
||||
AI Core 与 Agent Core 的模块边界、Mock Provider 和 Tool Calling 调试方式见 `../docs/AI-Core与Agent-Core开发说明.md`。
|
||||
AI Core 与 Agent Core 的模块边界、Mock Provider 和 Tool Calling 调试方式见 `../docs/development/AI-Core与Agent-Core开发说明.md`。
|
||||
|
||||
Knowledge Core 与 Retrieval Core 的模块边界、数据模型、接口与检索流程见 `../docs/Knowledge与Retrieval-Core开发说明.md`。
|
||||
Knowledge Core 与 Retrieval Core 的模块边界、数据模型、接口与检索流程见 `../docs/development/Knowledge与Retrieval-Core开发说明.md`。
|
||||
|
||||
@@ -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:
|
||||
|
||||
+170
-27
@@ -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,6 +275,19 @@ 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)
|
||||
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,
|
||||
@@ -220,6 +298,29 @@ class AgentRuntime:
|
||||
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(
|
||||
record,
|
||||
@@ -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
|
||||
|
||||
@@ -0,0 +1,372 @@
|
||||
"""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, apply_limits: bool = True
|
||||
) -> Any:
|
||||
"""递归净化持久化数据；可按审计用途限制体积，Secret 始终脱敏。"""
|
||||
|
||||
if apply_limits and depth >= MAX_TRACE_DEPTH:
|
||||
return "[MAX_DEPTH]"
|
||||
if isinstance(value, dict):
|
||||
sanitized: dict[str, Any] = {}
|
||||
for index, (key, item) in enumerate(value.items()):
|
||||
if apply_limits and 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, apply_limits=apply_limits
|
||||
)
|
||||
)
|
||||
return sanitized
|
||||
if isinstance(value, (list, tuple)):
|
||||
source_items = value[:MAX_TRACE_COLLECTION] if apply_limits else value
|
||||
items = [
|
||||
sanitize_trace_value(
|
||||
item, depth=depth + 1, apply_limits=apply_limits
|
||||
)
|
||||
for item in source_items
|
||||
]
|
||||
if apply_limits and 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 apply_limits and 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, apply_limits=apply_limits
|
||||
)
|
||||
|
||||
|
||||
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:
|
||||
# Run 是重启后 GET/list 的完整事实；只做 Secret 脱敏，不套用 Trace 摘要限长。
|
||||
return json.dumps(
|
||||
sanitize_trace_value(
|
||||
run.model_dump(mode="json"), apply_limits=False
|
||||
),
|
||||
ensure_ascii=False,
|
||||
)
|
||||
@@ -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"]
|
||||
|
||||
|
||||
@@ -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);
|
||||
""",
|
||||
]
|
||||
|
||||
|
||||
|
||||
+57
-6
@@ -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,
|
||||
|
||||
@@ -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,200 @@ 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())
|
||||
|
||||
|
||||
def test_persisted_agent_run_preserves_long_input_and_output() -> None:
|
||||
"""审计事件可以限长，但重启后读取的 AgentRun 不能丢失正文。"""
|
||||
|
||||
now = datetime.now(timezone.utc)
|
||||
long_input = "输入" * 2_500
|
||||
long_output = "输出" * 2_500
|
||||
request = AgentRunCreateRequest(
|
||||
input=long_input,
|
||||
provider_id="mock",
|
||||
model="mock-1",
|
||||
)
|
||||
persisted = AgentRun(
|
||||
run_id="run_long_content",
|
||||
status=AgentRunStatus.completed,
|
||||
input=long_input,
|
||||
output=long_output,
|
||||
provider_id=request.provider_id,
|
||||
model=request.model,
|
||||
max_steps=request.max_steps,
|
||||
created_at=now,
|
||||
updated_at=now,
|
||||
)
|
||||
repository = AgentTraceRepository()
|
||||
repository.create_run(persisted, request, {"model": request.model})
|
||||
|
||||
restored = repository.get_run(persisted.run_id)
|
||||
|
||||
assert restored is not None
|
||||
assert restored.input == long_input
|
||||
assert restored.output == long_output
|
||||
|
||||
|
||||
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())
|
||||
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -0,0 +1,69 @@
|
||||
# NotesAgent 文档索引
|
||||
|
||||
本目录集中保存团队开发期间需要长期维护的架构、接口、实现、协作和问题复盘文档。文档按用途分类，避免设计约束、开发记录与故障复盘混放。
|
||||
|
||||
## 目录分类
|
||||
|
||||
| 目录 | 内容 | 适用场景 |
|
||||
| --- | --- | --- |
|
||||
| [`architecture/`](architecture/) | 技术栈、阶段目标与团队分工 | 确认整体边界、模块依赖和阶段范围 |
|
||||
| [`contracts/`](contracts/) | 前后端接口契约与页面需求 | 开发前对齐 DTO、路由、事件和交互 |
|
||||
| [`development/`](development/) | 各模块的实现说明 | 阅读现有代码、联调和扩展功能 |
|
||||
| [`guides/`](guides/) | Git、测试、注释和 CI/CD 规范 | 日常开发、提交、审阅和发布 |
|
||||
| [`retrospectives/`](retrospectives/) | 审阅发现的问题与修复复盘 | 排查同类问题、撰写总结或博客 |
|
||||
|
||||
## architecture：架构与分工
|
||||
|
||||
- [AI 笔记软件技术栈说明](architecture/AI笔记软件技术栈说明-团队版-v2.3.md)
|
||||
- [第一阶段分工表](architecture/第一阶段分工表.md)
|
||||
- [第二阶段团队分工表](architecture/第二阶段团队分工表.md)
|
||||
|
||||
## contracts：契约与需求
|
||||
|
||||
- [后端接口契约](contracts/后端接口契约-开发版.md)
|
||||
- [第二阶段接口契约](contracts/第二阶段接口契约-开发版.md)
|
||||
- [前端页面需求说明](contracts/前端页面需求说明-开发版.md)
|
||||
|
||||
运行中的后端以 `/openapi.json` 为机器可读事实来源。接口契约用于描述设计意图、联调约束和实现状态；两者不一致时，应先确认代码行为，再在同一个 PR 中同步修正文档或实现。
|
||||
|
||||
## development：开发说明
|
||||
|
||||
- [AI Core 与 Agent Core 开发说明](development/AI-Core与Agent-Core开发说明.md)
|
||||
- [Knowledge 与 Retrieval Core 开发说明](development/Knowledge与Retrieval-Core开发说明.md)
|
||||
- [模型提供商与模型发现开发说明](development/模型提供商与模型发现开发说明.md)
|
||||
- [前端壳子与接口层开发说明](development/前端壳子与接口层开发说明.md)
|
||||
- [前端写作体验优化开发说明](development/前端写作体验优化开发说明.md)
|
||||
- [前端视觉与轻量动效优化开发说明](development/前端视觉与轻量动效优化开发说明.md)
|
||||
|
||||
## guides：团队协作规范
|
||||
|
||||
- [Git 使用细则](guides/Git使用细则-团队开发版.md)
|
||||
- [CI/CD 细则](guides/CI-CD细则-团队开发版.md)
|
||||
- [代码注释与 TODO 约定](guides/代码注释与TODO约定.md)
|
||||
- [第一阶段测试验证操作手册](guides/第一阶段测试验证操作手册.md)
|
||||
|
||||
## retrospectives：问题与修复复盘
|
||||
|
||||
- [后端全面审阅问题与修复复盘](retrospectives/后端全面审阅问题与修复复盘.md)
|
||||
- [Agent Core 第二阶段问题与修复复盘](retrospectives/Agent-Core第二阶段问题与修复复盘.md)
|
||||
- [Knowledge 与 Retrieval Core 问题与修复复盘](retrospectives/Knowledge与Retrieval-Core问题与修复复盘.md)
|
||||
- [前端合并审阅问题与修复复盘](retrospectives/前端合并审阅问题与修复复盘.md)
|
||||
|
||||
## 推荐阅读顺序
|
||||
|
||||
新成员或新阶段开始时，建议按以下顺序阅读：
|
||||
|
||||
1. 技术栈说明和当前阶段分工表；
|
||||
2. 所负责功能对应的接口契约；
|
||||
3. 对应模块的开发说明；
|
||||
4. Git、CI/CD、测试及注释规范；
|
||||
5. 与当前任务相关的问题复盘。
|
||||
|
||||
## 维护规则
|
||||
|
||||
- 新文档先判断用途，再放入对应分类目录，不在 `docs/` 根目录继续堆放业务文档。
|
||||
- 移动或重命名文档时，同步修正仓库内全部链接，并执行本地链接检查。
|
||||
- 接口、数据结构或事件格式发生变化时，同一个 PR 内同步更新契约和相关开发说明。
|
||||
- 问题复盘至少写清原因、后果、解决思路、实际方案和验证结果。
|
||||
- `.local-plans/` 只保存个人或阶段性的本地计划，不属于正式团队文档，不应提交到远程仓库。
|
||||
- 文档中的“计划实现”和“已经实现”必须明确区分；实现状态以代码、测试和运行时契约为准。
|
||||
@@ -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
|
||||
|
||||
@@ -1859,7 +1859,7 @@ POST /api/index/rebuild
|
||||
GET /health
|
||||
```
|
||||
|
||||
第一阶段已实现路径和第二阶段冻结草案分别见 `后端接口契约-开发版.md` 与 `第二阶段接口契约-开发版.md`。规划接口完成前不得在前端假定其可用；完成后以 OpenAPI、Pydantic Contract 和 TypeScript Wire DTO 的一致结果为准。
|
||||
第一阶段已实现路径和第二阶段冻结草案分别见 `../contracts/后端接口契约-开发版.md` 与 `../contracts/第二阶段接口契约-开发版.md`。规划接口完成前不得在前端假定其可用；完成后以 OpenAPI、Pydantic Contract 和 TypeScript Wire DTO 的一致结果为准。
|
||||
|
||||
HTTP 返回统一错误结构：
|
||||
|
||||
@@ -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 持久化/恢复接口也已完成。当前验证基线为后端 81 项测试、前端 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 配置。
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
> 文档用途：供团队在第一阶段进行页面设计、Vue 开发、前后端联调和验收。
|
||||
> 文档性质：开发需求基线，不是最终视觉规范或产品宣传文档。
|
||||
> 依据：`第一阶段分工表.md`、`AI笔记软件技术栈说明-团队版-v2.3.md`、`后端接口契约-开发版.md`。
|
||||
> 依据：`../architecture/第一阶段分工表.md`、`../architecture/AI笔记软件技术栈说明-团队版-v2.3.md`、`后端接口契约-开发版.md`。
|
||||
|
||||
> 实现状态：更新至 2026-08-31。全部已注册业务路由均已有真实页面；Markdown 写作/源码模式、Search、Chat、智能体执行轨迹、扩展管理、设置、Provider 预设、模型发现和开发阶段加密凭据输入均已落地。Web Workspace 已连接 FastAPI 管理的真实单 Vault；Tauri 原生目录选择和多 Vault 尚未接入。
|
||||
|
||||
@@ -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：后端 81 项回归测试通过。
|
||||
|
||||
- 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。
|
||||
@@ -4,7 +4,7 @@
|
||||
>
|
||||
> 更新日期：2026-08-31
|
||||
>
|
||||
> 依据：`第二阶段团队分工表.md`、`AI笔记软件技术栈说明-团队版-v2.3.md`、`后端接口契约-开发版.md`
|
||||
> 依据：`../architecture/第二阶段团队分工表.md`、`../architecture/AI笔记软件技术栈说明-团队版-v2.3.md`、`后端接口契约-开发版.md`
|
||||
|
||||
本文统一第二阶段新增能力的 HTTP、SSE、前端 Service、桌面 Host 和内部模块接口。文中标记为“计划新增”的路径尚未实现，不能据此判断当前服务已经支持；实现完成后以 FastAPI `/openapi.json`、TypeScript Wire DTO 和自动化测试共同作为最终依据。
|
||||
|
||||
@@ -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
|
||||
```
|
||||
|
||||
---
|
||||
@@ -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 已落地，后端当前回归基线为 81 项测试通过。
|
||||
|
||||
## 当前实现
|
||||
|
||||
@@ -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 值在入库前脱敏。Event、Request 和配置快照中的超长字符串与集合会截断；作为查询事实来源的 `AgentRun` 只脱敏、不限长，保证重启前后 input/output 内容一致。
|
||||
|
||||
## 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 与说话人分离仍按技术基线在第二阶段接入。
|
||||
+1
-1
@@ -3,7 +3,7 @@
|
||||
> 本文档用于团队开发和模块联调，记录 Knowledge Core / Retrieval Core 已经落地的
|
||||
> 模块边界、数据模型、接口与使用方式，对应分工表中的杨星萱。
|
||||
|
||||
> 更新日期：2026-08-30。第一阶段 Knowledge/Retrieval 主链路已经完成，并已接入 Agent Tool Registry；完整后端回归基线为 71 项测试通过。
|
||||
> 更新日期：2026-09-01。第一阶段 Knowledge/Retrieval 主链路已经完成，并已接入 Agent Tool Registry；完整后端回归基线为 81 项测试通过。
|
||||
|
||||
## 当前实现
|
||||
|
||||
@@ -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 81 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 权限警告，不影响 81 项测试结果，也不涉及产品代码。
|
||||
|
||||
Vite 当前会提示 Chat 与 Workspace 的部分异步 Chunk 超过 500 kB，这是 Milkdown、CodeMirror、KaTeX 和 Shiki 等编辑/渲染依赖带来的性能优化项，不影响构建成功或功能正确性；进入桌面打包前应通过手动分包或更细粒度动态加载继续优化。
|
||||
|
||||
@@ -104,4 +104,4 @@ pnpm build
|
||||
|
||||
自动化验证覆盖 Provider 预设、OpenAI-Compatible `/models` 请求与鉴权头、模型映射、前端自动刷新、排序去重及按 Provider 隔离错误。生产构建同时执行 Vue 和 TypeScript 类型检查。
|
||||
|
||||
当前完整回归基线：后端 71 项测试、前端 14 项测试通过，前端生产构建通过。Provider 配置目前仍保存在内存 Registry，AI Core 重启后需要重新创建；凭据密文会保留。OpenAI Responses 与 Anthropic Messages Adapter 尚未实现，设置页正式预设不会使用这两种协议。
|
||||
当前完整回归基线：后端 81 项测试、前端 27 项测试通过，前端类型检查和生产构建通过。Provider 配置目前仍保存在内存 Registry，AI Core 重启后需要重新创建；凭据密文会保留。OpenAI Responses 与 Anthropic Messages Adapter 尚未实现，设置页正式预设不会使用这两种协议。
|
||||
@@ -0,0 +1,166 @@
|
||||
# CI/CD 细则（团队开发版）
|
||||
|
||||
> 本文档规定 NotesAgent 在 Gitea 上的持续集成、构建产物、发布和回滚要求。当前仓库尚未提交 Gitea Actions 工作流，因此本文首先作为落地流水线时的统一规范；流水线启用前，Pull Request 仍须人工执行同等检查。
|
||||
|
||||
> 更新日期：2026-09-01。当前阶段的 CD 指“生成可验证的候选构建与发布产物”，不包含把后端自动部署到公网环境。
|
||||
|
||||
## 1. 目标与原则
|
||||
|
||||
CI/CD 用于尽早发现依赖锁文件失效、类型错误、测试回归、前后端契约不一致和生产构建失败。流水线应遵守以下原则：
|
||||
|
||||
- 以 Gitea 为唯一远程和流水线入口；
|
||||
- `main` 始终保持可安装、可测试、可构建；
|
||||
- 安装依赖时使用锁文件，避免流水线与开发机解析出不同版本；
|
||||
- 未通过必需检查的提交不得合入 `main`；
|
||||
- 外部模型、真实 API Key 和用户本地数据不得成为基础 CI 的前置条件；
|
||||
- 缓存只用于加速，不得影响构建结果；删除缓存后流水线仍应成功；
|
||||
- 测试、构建和发布步骤使用最小权限，敏感信息不得写入日志或产物。
|
||||
|
||||
## 2. 运行环境基线
|
||||
|
||||
| 组件 | CI 要求 | 说明 |
|
||||
| --- | --- | --- |
|
||||
| Python | 3.12 | 项目最低支持 3.11，CI 使用团队推荐版本 |
|
||||
| uv | 当前稳定版，并在日志中输出版本 | 按 `backend/uv.lock` 安装后端依赖 |
|
||||
| Node.js | 22 LTS | 满足前端环境要求并保持 Runner 兼容性 |
|
||||
| pnpm | 10 | 按 `frontend/pnpm-lock.yaml` 安装前端依赖 |
|
||||
| 操作系统 | Linux Runner 为基础门禁 | 桌面端启用后再增加 Windows、macOS 构建矩阵 |
|
||||
|
||||
Runner 镜像或 Action 的大版本必须固定。升级 Python、Node.js、uv、pnpm 或基础 Action 时，应使用独立的 `chore/` 分支，并完整运行前后端检查。
|
||||
|
||||
## 3. 触发规则
|
||||
|
||||
| 事件 | 必须执行 | 用途 |
|
||||
| --- | --- | --- |
|
||||
| Pull Request 指向 `main` | 文档检查、后端测试、前端测试、类型检查、生产构建 | 合并门禁 |
|
||||
| 推送到 `main` | 全量检查、集成冒烟、保存候选构建 | 验证合并结果 |
|
||||
| 推送功能分支 | 至少执行受影响模块的检查 | 尽早反馈；不得替代 PR 全量门禁 |
|
||||
| 推送 `v*` 标签 | 全量检查、构建、校验和、发布候选产物 | 正式发布入口 |
|
||||
| 手动触发 | 可选择全量回归或重新生成候选产物 | 发布前复核和故障恢复 |
|
||||
|
||||
纯文档变更可以跳过前后端耗时任务，但必须执行文档链接检查和 `git diff --check`。只有可靠的路径检测结果才能判定为纯文档变更；锁文件、工作流、构建配置和接口契约变更一律按代码变更处理。
|
||||
|
||||
## 4. Pull Request 必需检查
|
||||
|
||||
建议将以下 Job 名称固定为 Gitea 分支保护所要求的状态检查：
|
||||
|
||||
| Job | 必需命令或行为 | 通过标准 |
|
||||
| --- | --- | --- |
|
||||
| `docs-check` | `git diff --check`，检查仓库内 Markdown 相对链接 | 无空白错误、无失效本地链接 |
|
||||
| `backend-test` | `uv sync --frozen`、编译检查、`uv run pytest` | 依赖锁有效且测试全部通过 |
|
||||
| `frontend-test` | `pnpm install --frozen-lockfile`、`pnpm test` | 依赖锁有效且测试全部通过 |
|
||||
| `frontend-typecheck` | `pnpm type-check` | 无 TypeScript/Vue 类型错误 |
|
||||
| `frontend-build` | `pnpm build` | Vite 生产构建成功 |
|
||||
| `integration-smoke` | 启动 FastAPI，验证健康检查和关键本地链路 | 服务可启动，响应与契约符合预期 |
|
||||
|
||||
后端 Job 的基准命令：
|
||||
|
||||
```bash
|
||||
cd backend
|
||||
uv sync --frozen
|
||||
uv run python -m compileall -q app
|
||||
uv run pytest
|
||||
```
|
||||
|
||||
前端 Job 的基准命令：
|
||||
|
||||
```bash
|
||||
cd frontend
|
||||
pnpm install --frozen-lockfile
|
||||
pnpm test
|
||||
pnpm type-check
|
||||
pnpm build
|
||||
```
|
||||
|
||||
`integration-smoke` 应使用 Mock Provider、临时数据库和临时附件目录，不访问 OpenAI、DeepSeek 或其他外部服务。测试结束后必须关闭服务并清理临时数据。
|
||||
|
||||
## 5. 路径与模块检查规则
|
||||
|
||||
- 修改 `backend/**`、`backend/uv.lock` 或后端配置时，必须运行 `backend-test` 和 `integration-smoke`。
|
||||
- 修改 `frontend/**`、`frontend/pnpm-lock.yaml` 或前端配置时，必须运行全部前端 Job。
|
||||
- 修改 `docs/contracts/**`、FastAPI 路由、DTO、SSE 事件或前端 Service 类型时，必须同时运行前后端全量检查。
|
||||
- 修改 `.gitea/**`、根目录工程配置或依赖版本时，必须运行所有 Job。
|
||||
- 修改 `docs/**` 以外且无法明确归类的文件时，默认运行所有 Job。
|
||||
|
||||
路径过滤只用于减少无关重复任务，不得造成关键检查缺失。若无法可靠判断影响范围，应执行全量流水线。
|
||||
|
||||
## 6. 凭据与敏感信息
|
||||
|
||||
- 基础 CI 不配置真实模型 API Key，Provider 相关测试统一使用 Mock 或请求桩。
|
||||
- 确需发布签名或访问受保护服务时，凭据只保存在 Gitea Actions Secrets 中，不写入仓库、工作流参数、缓存或构建产物。
|
||||
- 来自外部分支或不受信任 Pull Request 的任务不得读取发布凭据。
|
||||
- Secret 名称表达用途和环境，例如 `RELEASE_SIGNING_KEY`；禁止使用含义模糊的 `KEY1`、`TOKEN2`。
|
||||
- 日志中禁止输出请求头、完整 Token、API Key、用户笔记内容和本地凭据存储内容。
|
||||
- 生产凭据与测试凭据分离，并遵循最小权限、定期轮换和可撤销原则。
|
||||
|
||||
前端构建时注入的变量会进入静态资源，不能用于保存秘密。只有明确可公开的配置才允许使用 Vite 客户端环境变量。
|
||||
|
||||
## 7. 缓存与产物
|
||||
|
||||
可以缓存 uv 下载缓存和 pnpm Store，缓存键至少包含操作系统、运行时版本和对应锁文件哈希。不得缓存：
|
||||
|
||||
- `backend/.venv/`；
|
||||
- `frontend/node_modules/`；
|
||||
- `backend/data/`、测试数据库和用户附件；
|
||||
- `.env`、API Key、本地凭据库或签名材料。
|
||||
|
||||
普通 PR 不上传可执行发布包，只保留必要的测试报告和前端构建日志。`main` 或版本标签的候选产物应记录提交 SHA，生成 SHA-256 校验和，并设置明确的保留期限；非正式候选产物建议保留 14 天。
|
||||
|
||||
## 8. 分支保护与合并门禁
|
||||
|
||||
Gitea 中的 `main` 应启用以下保护：
|
||||
|
||||
- 禁止普通成员直接推送和强制推送；
|
||||
- 要求 Pull Request 审阅通过；
|
||||
- 要求第 4 节列出的适用状态检查成功；
|
||||
- Head 更新后使旧审阅和旧检查失效，必须针对最新提交重新检查；
|
||||
- 对话和审阅意见处理完成后才允许合并；
|
||||
- 优先使用 squash 或 rebase 保持主线清晰，具体方式遵循 [Git 使用细则](Git使用细则-团队开发版.md)。
|
||||
|
||||
临时绕过门禁只允许用于明确的仓库级故障。绕过者需要记录原因、影响、补验计划，并在恢复后立即补跑全部检查。
|
||||
|
||||
## 9. 发布流程
|
||||
|
||||
当前阶段按以下顺序生成发布候选：
|
||||
|
||||
1. 从已通过全部检查的 `main` 提交确定发布 SHA；
|
||||
2. 更新版本号、变更说明和必要文档；
|
||||
3. 创建形如 `v0.2.0` 的语义化版本标签；
|
||||
4. 标签流水线重新执行全部测试和生产构建；
|
||||
5. 对产物执行本地启动或安装冒烟测试；
|
||||
6. 生成校验和，并把版本、提交 SHA、构建环境和已知限制写入发布说明；
|
||||
7. 人工确认后在 Gitea 发布页面公开产物。
|
||||
|
||||
Tauri 桌面端接入后，发布流水线再增加 Windows、macOS 和 Linux 构建矩阵、平台签名及安装包验证。在签名、更新通道和回滚方案准备完成前，不启用面向用户的自动更新。
|
||||
|
||||
## 10. 回滚与热修复
|
||||
|
||||
- 尚未公开的候选产物直接标记为失败，不覆盖同一版本的已有产物；修复后递增预发布编号或版本号。
|
||||
- 已发布版本出现问题时，优先停止分发并回退到最近一个已验证版本。
|
||||
- 代码修复从 `main` 创建 `hotfix/<模块>-<问题>` 分支，通过完整门禁后合并并发布补丁版本。
|
||||
- 禁止重写已公开版本标签或用新文件替换旧版本同名产物。
|
||||
- 回滚或热修复完成后，在 `docs/retrospectives/` 记录原因、影响、处置过程和防复发措施。
|
||||
|
||||
## 11. 流水线失败处理
|
||||
|
||||
1. 先确认失败是否可在本地使用相同锁文件和命令复现；
|
||||
2. 判断是代码、测试、依赖、Runner 还是外部基础设施问题；
|
||||
3. 代码或测试问题由当前 PR 修复，不通过重跑掩盖不稳定测试；
|
||||
4. Runner 或 Gitea 故障应记录日志和时间，恢复后针对同一 Head 重新执行；
|
||||
5. 连续出现的偶发失败必须作为缺陷处理，明确负责人并增加稳定性修复；
|
||||
6. 修复流水线本身时，不得顺便降低测试范围或绕过既有门禁。
|
||||
|
||||
## 12. 落地清单
|
||||
|
||||
首次创建 `.gitea/workflows/` 时，应逐项确认：
|
||||
|
||||
- [ ] 工作流只使用 Gitea Runner 支持且来源可信的 Action；
|
||||
- [ ] Python、Node.js、uv 和 pnpm 版本符合本规范；
|
||||
- [ ] 后端和前端依赖均以 frozen 模式安装；
|
||||
- [ ] 必需 Job 名称与 `main` 分支保护一致；
|
||||
- [ ] Mock 测试不依赖外部模型服务和真实凭据；
|
||||
- [ ] 缓存键包含锁文件哈希，缓存内容不含用户数据或秘密；
|
||||
- [ ] PR、`main`、版本标签和手动触发行为分别验证；
|
||||
- [ ] 失败任务能返回非零退出码，后续发布步骤不会继续；
|
||||
- [ ] 候选产物包含提交 SHA、校验和和保留期限；
|
||||
- [ ] 团队成员能够按本文档在本地复现全部门禁。
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
> 本文档用于 Notes Agent 团队日常开发。目标是让三名成员可以并行开发、稳定联调，并确保 `main` 始终处于可运行状态。
|
||||
|
||||
> 更新日期：2026-08-30。当前远程只使用 `gitea`，功能分支不添加个人或工具名称前缀；合并门槛为相关测试、前端生产构建、文档同步和 `git diff --check` 全部通过。
|
||||
> 更新日期：2026-09-01。当前远程只使用 `gitea`，功能分支不添加个人或工具名称前缀；合并门槛为相关测试、前端生产构建、文档同步和 `git diff --check` 全部通过。自动化门禁、产物和发布规则见 [CI/CD 细则](CI-CD细则-团队开发版.md)。
|
||||
|
||||
## 1. 仓库与远程
|
||||
|
||||
@@ -622,4 +622,4 @@ git push -u gitea <分支名>
|
||||
git log --oneline --decorate -10
|
||||
```
|
||||
|
||||
本细则的核心要求是：`main` 可运行、改动可 Review、问题可追踪、敏感信息不入库、跨模块变化同步 Contract 与文档。
|
||||
本细则的核心要求是：`main` 可运行、改动可 Review、问题可追踪、敏感信息不入库、跨模块变化同步 Contract 与文档。流水线启用后，合并和发布还必须满足 [CI/CD 细则](CI-CD细则-团队开发版.md) 中的状态检查与产物要求。
|
||||
@@ -68,7 +68,7 @@ uv run pytest -q -p no:cacheprovider
|
||||
当前基线：
|
||||
|
||||
```text
|
||||
76 passed
|
||||
81 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 双主题输出。
|
||||
@@ -0,0 +1,491 @@
|
||||
# Agent Core 第二阶段：Trace 持久化与 SSE 恢复问题复盘
|
||||
|
||||
> 审阅与修复日期：2026-09-01
|
||||
> 涉及分支：`feat/agent-trace-persistence`
|
||||
> 功能提交：`3cb197a feat(agent): 持久化Trace并支持SSE恢复`
|
||||
> 文档用途：记录 Agent Run、Trace、SSE 恢复和审计数据安全问题的形成原因、实际后果、解决思路与落地方案，供后续开发文档、比赛材料和技术博客写作使用。
|
||||
|
||||
## 1. 背景与结论
|
||||
|
||||
第一阶段 Agent Runtime 已经能够完成模型调用、Tool Calling、权限确认、取消、Usage 和 Citation，但 Run 与 Event 仍以进程内字典和列表为事实来源。第一阶段审阅加入的 Run/Event 数量上限解决了内存无界增长，却没有解决重启丢失、断线续传、Benchmark 复用和敏感数据审计等第二阶段问题。
|
||||
|
||||
本轮处理了 9 类问题：
|
||||
|
||||
| 编号 | 问题 | 级别 | 处理结果 |
|
||||
| --- | --- | --- | --- |
|
||||
| A-01 | Agent Run 与 Event 只存在于进程内存 | P0 | 增加 SQLite v3 Schema 和 Trace Repository |
|
||||
| A-02 | Event 裁剪后 sequence 可能重复 | P0 | 改为独立单调序号并建立数据库幂等键 |
|
||||
| A-03 | SSE 断线后无法从指定事件恢复 | P1 | 支持 `Last-Event-ID`、`after_sequence` 和 SSE `id` |
|
||||
| A-04 | 缺少可分页 Trace 与 Benchmark 配置快照 | P1 | 增加 Trace API、统计摘要与配置快照 |
|
||||
| A-05 | Trace 缺少模型调用、父子关系和权限结果 | P1 | 增加第二阶段事件及耗时/parent 字段 |
|
||||
| A-06 | Trace 可能保存 Secret 和超大 Tool Result | P0 | 全部持久化副本统一脱敏，审计摘要限长 |
|
||||
| A-07 | 进程重启后未终止 Run 永久显示运行中 | P1 | 自动收束为 `AGENT_PROCESS_RESTARTED` |
|
||||
| A-08 | 前端 DTO 与 SSE Client 无法消费恢复协议 | P1 | 同步 TypeScript Contract、Service、标签和 SSE id |
|
||||
| A-09 | AgentRun 与审计摘要共用限长规则 | P1 | Run 只脱敏不限长，保证重启前后内容一致 |
|
||||
|
||||
修复后的验证基线为：后端 81 项测试、前端 27 项测试、TypeScript 类型检查和生产构建通过。
|
||||
|
||||
## 2. A-01：Agent Run 与 Event 只存在于进程内存
|
||||
|
||||
### 原因
|
||||
|
||||
旧 `AgentRuntime` 使用 `_records: dict[str, RunRecord]` 保存全部运行状态。每个 `RunRecord` 内部再保存 `events`、实时订阅队列和异步任务。创建、查询、列表和 SSE 回放都只读取这个字典，没有 Repository 边界。
|
||||
|
||||
第一阶段为控制内存加入了最多 200 个 Run、每个 Run 2000 个 Event 的限制。这是必要的资源保护，但只是减少进程内数据量，不能替代持久化。
|
||||
|
||||
### 后果
|
||||
|
||||
- AI Core 重启后，历史 Run、Tool Result、Citation 和错误全部丢失；
|
||||
- 前端刷新或重新连接后，只能读取当前进程尚未淘汰的数据；
|
||||
- Agent Benchmark 无法使用稳定的历史事实计算指标；
|
||||
- Run 列表随着进程重启清空，界面记录和实际操作脱节；
|
||||
- 内存裁剪后的旧事件无法再恢复。
|
||||
|
||||
### 解决思路
|
||||
|
||||
将 SQLite 中的 Run/Event 作为唯一持久化事实，把内存降级为执行上下文和实时订阅窗口。Runtime 继续负责状态机，Repository 负责落库、分页、恢复和统计，Router 不直接访问数据库。
|
||||
|
||||
### 解决方案
|
||||
|
||||
新增 SQLite v3 migration：
|
||||
|
||||
```text
|
||||
agent_runs
|
||||
├── run_id
|
||||
├── status
|
||||
├── run_json
|
||||
├── request_json
|
||||
├── config_snapshot_json
|
||||
├── created_at
|
||||
└── updated_at
|
||||
|
||||
agent_events
|
||||
├── run_id
|
||||
├── sequence
|
||||
├── event
|
||||
├── data_json
|
||||
└── timestamp
|
||||
```
|
||||
|
||||
新增 `AgentTraceRepository`，负责：
|
||||
|
||||
- 创建 Run 及请求、配置快照；
|
||||
- 在同一事务中更新 Run 并追加 Event；
|
||||
- 按创建时间分页列出 Run；
|
||||
- 按 sequence 读取 Event；
|
||||
- 生成 Trace 分页响应和统计摘要；
|
||||
- 恢复进程中断的非终态 Run。
|
||||
|
||||
内存中的 2000 条 Event 上限继续保留，但只用于实时订阅窗口；完整记录由 SQLite 管理。
|
||||
|
||||
## 3. A-02：Event 裁剪后 sequence 可能重复
|
||||
|
||||
### 原因
|
||||
|
||||
旧 `_publish()` 使用下面的方式生成序号：
|
||||
|
||||
```python
|
||||
sequence = len(record.events)
|
||||
```
|
||||
|
||||
当事件数量超过 2000 后，Runtime 会删除列表头部。列表长度重新回到 2000，后续事件仍会得到 2000，形成重复 sequence。
|
||||
|
||||
### 后果
|
||||
|
||||
- `run_id + sequence` 无法作为幂等键；
|
||||
- 前端按 sequence 去重时会错误丢弃新事件；
|
||||
- 时间线排序出现同序号节点；
|
||||
- 持久化后会触发主键冲突，或者在错误的覆盖策略下破坏旧事件；
|
||||
- Benchmark 无法可靠还原 Tool Call 顺序。
|
||||
|
||||
### 解决思路
|
||||
|
||||
sequence 应属于 Run 的逻辑时钟，不能从当前缓存长度推导。内存是否裁剪不得影响序号。
|
||||
|
||||
### 解决方案
|
||||
|
||||
- `RunRecord` 增加独立的 `next_sequence`；
|
||||
- 每次发布读取当前值，再原子递增；
|
||||
- `agent_events` 使用 `(run_id, sequence)` 复合主键；
|
||||
- Repository 对重复写入使用幂等插入，不覆盖已经存在的事件事实；
|
||||
- Trace 和 SSE 均严格按 sequence 升序返回。
|
||||
|
||||
## 4. A-03：SSE 断线后无法恢复
|
||||
|
||||
### 原因
|
||||
|
||||
旧 SSE 接口只能从内存列表头部重新回放全部历史，再切换到实时队列。协议帧只有 `event:` 和 `data:`，没有 SSE 标准的 `id:`。接口也不读取 `Last-Event-ID` 或查询游标。
|
||||
|
||||
前端即使知道自己最后处理到哪个 sequence，也无法把该位置传回服务端。
|
||||
|
||||
### 后果
|
||||
|
||||
- 短暂断网或页面切换后只能从头回放；
|
||||
- 长 Run 重连会重复传输大量事件；
|
||||
- 前端需要依赖本地去重掩盖服务端缺少恢复能力；
|
||||
- AI Core 重启后无法续传，因为历史事件本身也不存在；
|
||||
- 实时与历史交界处容易漏事件或重复事件。
|
||||
|
||||
### 解决思路
|
||||
|
||||
恢复协议以 sequence 为游标。服务端先注册实时订阅，再读取 `sequence > cursor` 的持久化历史，随后消费实时队列；交界处允许重复，但 Runtime 和前端都按 sequence 去重。
|
||||
|
||||
### 解决方案
|
||||
|
||||
接口支持两种游标输入：
|
||||
|
||||
```http
|
||||
GET /api/agent/runs/{run_id}/events?after_sequence=42
|
||||
Last-Event-ID: 42
|
||||
```
|
||||
|
||||
返回帧包含：
|
||||
|
||||
```text
|
||||
id: 43
|
||||
event: ToolResult
|
||||
data: {"run_id":"...","sequence":43,"data":{...}}
|
||||
```
|
||||
|
||||
具体处理：
|
||||
|
||||
- `after_sequence` 优先于 `Last-Event-ID`；
|
||||
- 默认游标为 `-1`，表示从 sequence 0 开始；
|
||||
- 非整数或小于 `-1` 的 Header 返回 `TRACE_CURSOR_INVALID`；
|
||||
- 历史回放读取 SQLite，不依赖内存窗口；
|
||||
- 历史与实时交界处按最后已发送 sequence 跳过重复项；
|
||||
- 终态 Run 回放完终止事件后关闭连接。
|
||||
|
||||
## 5. A-04：缺少可分页 Trace 与 Benchmark 配置快照
|
||||
|
||||
### 原因
|
||||
|
||||
旧接口只有 Run 状态和 SSE。SSE 适合实时消费，不适合报告页随机访问、大 Trace 分页或 Benchmark 批量计算。Run 也没有保存创建时的 Provider、Model、Skill、允许工具和限制参数快照。
|
||||
|
||||
### 后果
|
||||
|
||||
- Trace 页面只能依赖一次长连接重建全部状态；
|
||||
- Benchmark 需要绕过正式接口读取 Runtime 内部对象；
|
||||
- Provider 或 Skill 配置变化后，旧结果失去可解释性；
|
||||
- 大 Trace 无法受控分页，接口响应体会持续增大；
|
||||
- 前端和 Benchmark 容易各自实现一套不一致的统计逻辑。
|
||||
|
||||
### 解决思路
|
||||
|
||||
提供面向读取的 Trace Snapshot API，但只返回平铺事实，不在后端生成前端树形布局。前端按 parent ID 和 sequence 构造时间线，Benchmark 从同一事件计算指标。
|
||||
|
||||
### 解决方案
|
||||
|
||||
新增接口：
|
||||
|
||||
```http
|
||||
GET /api/agent/runs/{run_id}/trace?after_sequence=-1&limit=200
|
||||
```
|
||||
|
||||
响应包含：
|
||||
|
||||
```json
|
||||
{
|
||||
"run_id": "run_123",
|
||||
"status": "completed",
|
||||
"items": [],
|
||||
"next_sequence": 199,
|
||||
"has_more": true,
|
||||
"summary": {
|
||||
"model_calls": 2,
|
||||
"tool_calls": 3,
|
||||
"duration_ms": 1530,
|
||||
"token_usage": 2048,
|
||||
"errors": 0
|
||||
},
|
||||
"config_snapshot": {}
|
||||
}
|
||||
```
|
||||
|
||||
配置快照保存：
|
||||
|
||||
- Provider ID 和类型；
|
||||
- Model；
|
||||
- Capability；
|
||||
- Skill ID；
|
||||
- 允许的 Tool；
|
||||
- `max_steps`、Token Budget 和网络权限；
|
||||
- 经过脱敏的 Metadata。
|
||||
|
||||
## 6. A-05：Trace 缺少关键执行事实
|
||||
|
||||
### 原因
|
||||
|
||||
第一阶段事件能够表达 Run、Tool、Permission Request、Usage 和 Citation，但没有明确表示一次模型调用的开始、完成或失败。Tool Call 与模型轮次之间也没有 parent ID，Tool Result 缺少统一耗时。权限接口只唤醒 Future，不记录最终决定。
|
||||
|
||||
### 后果
|
||||
|
||||
- 前端无法展示“模型调用 → 多个 Tool → 下一次模型调用”的完整树；
|
||||
- Agent Benchmark 无法计算模型调用次数和平均步骤耗时；
|
||||
- 并发 Tool Call 时难以判断属于哪个模型轮次；
|
||||
- 权限卡片消失后，Trace 中只保留“请求过权限”，不知道用户允许还是拒绝；
|
||||
- Provider 失败只能看到最终 RunFailed，缺少模型调用级上下文。
|
||||
|
||||
### 解决思路
|
||||
|
||||
保持现有 AgentEvent envelope 不变，只增加事件类型和可选数据字段。调用关系使用稳定 ID 表达，不让前端根据相邻位置猜测父子关系。
|
||||
|
||||
### 解决方案
|
||||
|
||||
新增事件：
|
||||
|
||||
```text
|
||||
ModelCallStarted
|
||||
ModelCallCompleted
|
||||
ModelCallFailed
|
||||
PermissionResolved
|
||||
```
|
||||
|
||||
补充字段：
|
||||
|
||||
- Model Call：`model_call_id`、Provider、Model、Step、Finish Reason、Token、Duration；
|
||||
- Tool Call/Result：`parent_model_call_id`、`duration_ms`；
|
||||
- Permission Resolved：`request_id`、Permission、Decision；
|
||||
- Model Call Failed：`error_code` 和耗时，不写入第三方原始敏感异常。
|
||||
|
||||
模型与 Tool 耗时使用单调时钟计算，避免系统时间调整影响 Duration。
|
||||
|
||||
## 7. A-06：Trace 可能保存 Secret 和超大结果
|
||||
|
||||
### 原因
|
||||
|
||||
Tool 参数和输出来自模型、插件或外部服务，属于不可信数据。旧事件直接保存 `ToolCall.model_dump()` 和 `ToolResult.model_dump()`。`notes.read`、附件或第三方 Tool 可以返回大段正文，参数也可能包含 `api_key`、Authorization 或 Password。
|
||||
|
||||
初版持久化修复只净化了 `agent_events.data_json`。提交前审阅发现，`agent_runs.run_json` 中的 `tool_results`、`output` 和 `input` 仍可能保存同一份敏感值，说明“只在事件层脱敏”并不完整。
|
||||
|
||||
### 后果
|
||||
|
||||
- API Key 或 Bearer Token 可能进入 SQLite、备份和测试产物；
|
||||
- Trace API 不返回 Secret，但数据库中的 Run Snapshot 仍可能泄露；
|
||||
- 单个 Tool Result 可以让 Event 和 Run JSON 快速膨胀；
|
||||
- 前端展开节点时可能因为超大 JSON 卡顿；
|
||||
- Benchmark Dataset 或报告导出可能间接携带密钥。
|
||||
|
||||
### 解决思路
|
||||
|
||||
所有进入持久化边界的数据统一经过同一个净化函数，不能分别在 Router、Runtime 和 Repository 中维护不同脱敏规则。Secret 脱敏适用于全部副本；体积限制只适用于 Trace Event、Request 和 Config 等审计数据，不能改变对外查询所依赖的 AgentRun 事实。
|
||||
|
||||
### 解决方案
|
||||
|
||||
统一处理以下对象：
|
||||
|
||||
```text
|
||||
AgentRun Snapshot
|
||||
AgentRunCreateRequest Snapshot
|
||||
Config Snapshot
|
||||
AgentEvent Data
|
||||
```
|
||||
|
||||
通用脱敏规则：
|
||||
|
||||
- `api_key`、Authorization、Access/Refresh Token、Password、Secret 等键替换为 `[REDACTED]`；
|
||||
- 常见 `sk-...` 和 `Bearer ...` 字符串模式直接替换；
|
||||
- `credential_id` 等非明文引用保留，不误判为 Secret。
|
||||
|
||||
Event、Request 和 Config 审计副本额外执行限长：
|
||||
|
||||
- 单字符串最多保留 4096 个字符；
|
||||
- 单集合最多保留 100 项；
|
||||
- 递归深度最多 8 层；
|
||||
- 超限位置使用明确的 `[TRUNCATED]` 或 `[MAX_DEPTH]` 标记。
|
||||
|
||||
`AgentRun` Snapshot 仍经过同一套 Secret 脱敏，但不执行长度、集合和深度截断。回归测试直接读取 `agent_runs` 原始 SQLite 字段，确认测试密钥没有落盘，避免只验证 API 响应造成假安全。
|
||||
|
||||
## 8. A-07：重启后未终止 Run 永久显示运行中
|
||||
|
||||
### 原因
|
||||
|
||||
Agent 的异步 Task 和 Permission Future 不能跨进程恢复。持久化 Run 后，如果直接返回数据库状态，重启前处于 `queued`、`running` 或 `waiting_permission` 的记录会一直保持非终态，但新进程中没有对应 Task 可以继续执行。
|
||||
|
||||
### 后果
|
||||
|
||||
- 前端长期显示“运行中”或“等待授权”；
|
||||
- SSE 订阅等待一个永远不会到来的终止事件；
|
||||
- Benchmark Runner 无法判断 Case 已中断；
|
||||
- 用户取消该 Run 时，新进程找不到实际 Task；
|
||||
- 统计中的成功率和耗时被悬挂 Run 污染。
|
||||
|
||||
### 解决思路
|
||||
|
||||
本阶段提供“状态恢复”，不伪装成“执行恢复”。没有可重放状态机、Provider 幂等令牌和 Tool 副作用日志之前，自动继续执行会造成重复写入或重复网络请求。
|
||||
|
||||
### 解决方案
|
||||
|
||||
新 Runtime 首次读取不属于当前进程的非终态 Run 时：
|
||||
|
||||
- 状态改为 `failed`；
|
||||
- 错误码设为 `AGENT_PROCESS_RESTARTED`；
|
||||
- 错误信息说明进程在完成前重启；
|
||||
- 使用数据库最大 sequence 加一，追加唯一 `RunFailed` 事件；
|
||||
- 后续 Run 查询、Trace 和 SSE 都返回同一终态事实。
|
||||
|
||||
已经完成、失败或取消的 Run 不修改，可以在重启后继续查询和回放。
|
||||
|
||||
## 9. A-08：前端无法消费第二阶段恢复协议
|
||||
|
||||
### 原因
|
||||
|
||||
后端增加新事件和 SSE `id` 后，前端 `AgentEventType` 仍只包含第一阶段事件。`SseClient` 只解析 `event` 和 `data`，忽略 `id`，也没有发送 `Last-Event-ID`。因此仅完成后端并不能形成可联调的 Contract。
|
||||
|
||||
### 后果
|
||||
|
||||
- `Record<AgentEventType, string>` 中文标签无法通过类型检查；
|
||||
- 前端不知道 Model Call 和 Permission Resolved 的类型；
|
||||
- 断线后无法把最后事件 ID 传回后端；
|
||||
- Trace 可视化负责人需要自行猜测 Wire DTO；
|
||||
- 后端恢复能力只能通过 Curl 使用，页面调用链没有闭环。
|
||||
|
||||
### 解决思路
|
||||
|
||||
范侧只提供稳定的前端接口适配，不越过分工实现 Trace Visualization。Pydantic Contract、TypeScript Wire DTO 和 Service 必须在同一功能提交中同步。
|
||||
|
||||
### 解决方案
|
||||
|
||||
- TypeScript 增加四类第二阶段 Agent Event；
|
||||
- 增加 `AgentTraceSummary` 和 `AgentTraceResponse`；
|
||||
- `agentService.getAgentTrace()` 封装分页查询；
|
||||
- `streamAgentEvents()` 接受 `afterSequence`；
|
||||
- `SseClient` 发送 `Last-Event-ID` 并解析返回帧的 `id`；
|
||||
- 新事件增加中文标签和详情字段名；
|
||||
- 增加 SSE 请求头和 Event ID 单元测试。
|
||||
|
||||
Trace 时间线、树形布局、筛选、节点展开和 Citation 跳转仍由前端负责人实现。
|
||||
|
||||
## 10. A-09：AgentRun 与审计摘要共用限长规则
|
||||
|
||||
### 原因
|
||||
|
||||
初版使用 `sanitize_trace_value()` 同时处理 `run_json`、Request、Config 和 Event。该函数不仅脱敏，还会截断超过 4096 字符的字符串、超过 100 项的集合和超过 8 层的结构。`run_json` 随后又被 `GET /agent/runs/{run_id}` 和 Run 列表当作重启后的事实来源，因此审计数据的防膨胀规则意外改变了业务响应。
|
||||
|
||||
### 后果
|
||||
|
||||
- 常见的长模型回答在 AI Core 重启后只剩前 4096 个字符和截断标记；
|
||||
- 长输入、Tool Result 和嵌套结果也可能丢失；
|
||||
- 同一个 Run 在进程内与重启后的接口响应不一致；
|
||||
- 前端刷新、Benchmark 复核和问题追踪无法取得原始运行结果；
|
||||
- 原有测试只验证了 Trace 参数截断，没有比较重启前后的长正文。
|
||||
|
||||
审阅时使用 5000 字符 output 复现：写入前长度为 5000，重新读取后长度变为 4110，并以 `...[TRUNCATED]` 结尾。
|
||||
|
||||
### 解决思路
|
||||
|
||||
持久化边界包含两类数据：AgentRun 是业务事实，Trace Event、Request 和 Config 是可视化与审计摘要。两类数据必须共享 Secret 脱敏规则，但不能共享有损的体积限制。
|
||||
|
||||
### 解决方案
|
||||
|
||||
- `sanitize_trace_value()` 增加明确的 `apply_limits` 策略参数；
|
||||
- 默认继续限长，保持 Event、Request 和 Config 的安全边界；
|
||||
- `_serialize_run()` 使用 `apply_limits=False`，完整保留 input、output、Tool Result 和 Citation；
|
||||
- Secret 键名及 `sk-`、Bearer 模式在两种策略下始终脱敏；
|
||||
- 增加超过 4096 字符的 input/output 持久化回归测试，直接从新 Repository 读取并逐字比较。
|
||||
|
||||
## 11. 事务、顺序与恢复不变量
|
||||
|
||||
本轮修复明确了以下不变量：
|
||||
|
||||
1. `run_id + sequence` 唯一标识一条 Agent Event。
|
||||
2. sequence 在一个 Run 内只增不减，不受内存裁剪影响。
|
||||
3. 发布事件时，在同一 SQLite 事务中更新 Run Snapshot 并追加 Event。
|
||||
4. SSE、Trace 页面和 Agent Benchmark 读取同一份 `agent_events`，不建立旁路。
|
||||
5. 终止事件为 `RunCompleted`、`RunFailed` 或 `RunCancelled`；终态 Run 不再产生业务事件。
|
||||
6. 重启后不能安全继续执行的 Run 必须明确失败，不能永久悬挂。
|
||||
7. 所有持久化 Trace 数据先脱敏、再写入。
|
||||
8. 前端按 `run_id + sequence` 去重，不能依赖一次网络读取对应一条 SSE Event。
|
||||
9. AgentRun 的持久化副本不执行审计摘要限长，重启前后业务字段必须一致。
|
||||
|
||||
## 12. 验证方法
|
||||
|
||||
后端：
|
||||
|
||||
```powershell
|
||||
cd backend
|
||||
uv run python -m compileall -q app
|
||||
uv run pytest
|
||||
```
|
||||
|
||||
前端：
|
||||
|
||||
```powershell
|
||||
cd frontend
|
||||
pnpm test
|
||||
pnpm type-check
|
||||
pnpm build
|
||||
```
|
||||
|
||||
仓库检查：
|
||||
|
||||
```powershell
|
||||
git diff --check
|
||||
```
|
||||
|
||||
验证结果：
|
||||
|
||||
```text
|
||||
backend pytest 81 passed
|
||||
backend compileall passed
|
||||
frontend vitest 11 files / 27 tests passed
|
||||
frontend type-check passed
|
||||
frontend build passed
|
||||
git diff --check passed
|
||||
```
|
||||
|
||||
本轮新增回归覆盖：
|
||||
|
||||
- 完成 Run 在新 Runtime 中恢复查询；
|
||||
- Trace 多页读取和游标无重复；
|
||||
- SSE `Last-Event-ID`、`after_sequence` 和 `id:` 帧；
|
||||
- Model Call、Permission Resolved 和 Tool parent ID；
|
||||
- 进程中断 Run 自动生成唯一终止事件；
|
||||
- API Key、Authorization 和超长 Tool 参数净化；
|
||||
- 超长 AgentRun input/output 在持久化和重启读取后保持完整；
|
||||
- 直接检查 SQLite，确认 Secret 未进入 Run/Request/Config Snapshot；
|
||||
- OpenAPI 发布 Trace 路径；
|
||||
- 前端 SSE Client 发送和解析恢复游标。
|
||||
|
||||
测试仍会出现本机 `.pytest_cache` 无写入权限警告，不影响 81 项用例结果，也不涉及产品代码。
|
||||
|
||||
## 13. 当前边界与后续工作
|
||||
|
||||
### 13.1 本阶段明确不做
|
||||
|
||||
- 不在后端生成前端 Trace 树形布局；
|
||||
- 不在进程重启后自动重放未完成 Tool 副作用；
|
||||
- 不把 Secret 明文放入 Trace、日志或 Benchmark；
|
||||
- 不为 Benchmark 建立绕过 Agent Runtime 的专用执行协议。
|
||||
|
||||
### 13.2 后续需要继续处理
|
||||
|
||||
- 增加 Trace 保留、归档和被 Benchmark 引用时的保护策略；
|
||||
- 引入保留窗口后实现 `TRACE_CURSOR_EXPIRED`；
|
||||
- 根据桌面网络策略增加有上限的指数退避自动重连；
|
||||
- 评估高频 Token Event 的批量写入，减少 SQLite 连接与事务开销；
|
||||
- Agent Benchmark 接入正式 Trace 并验证指标字段是否充足；
|
||||
- 前端完成 Trace Timeline/Tree、筛选、节点详情和 Citation 跳转；
|
||||
- 多进程或远程执行出现需求后，再设计带租约和幂等副作用的执行恢复。
|
||||
|
||||
## 14. 可复用经验
|
||||
|
||||
### 14.1 资源上限不等于持久化
|
||||
|
||||
限制内存 Run 和 Event 数量只能防止进程膨胀，不能解决重启、审计和报告复现。临时保护措施应在文档中明确标注，不能被误认为最终架构已经完成。
|
||||
|
||||
### 14.2 游标必须独立于缓存结构
|
||||
|
||||
只要 sequence 来源于 `len(list)`、数组下标或当前页位置，裁剪和分页就可能破坏唯一性。可恢复事件流必须使用独立、单调且可持久化的逻辑序号。
|
||||
|
||||
### 14.3 恢复读取不等于恢复执行
|
||||
|
||||
恢复 Run/Trace 查询相对安全；恢复一个包含 Tool 副作用的执行任务需要额外的幂等、租约和补偿机制。在没有这些机制时，明确失败比重复执行更可靠。
|
||||
|
||||
### 14.4 脱敏要覆盖全部持久化副本
|
||||
|
||||
同一敏感值可能同时出现在 Event、Run Snapshot、Request、Config、日志和报告中。只检查最终 API 响应无法证明数据没有落盘，安全测试应直接验证持久化介质。
|
||||
|
||||
### 14.5 生产者和消费者 Contract 必须同时更新
|
||||
|
||||
后端新增事件类型、字段或 SSE 规则时，至少同步 Pydantic、OpenAPI、TypeScript DTO、Service 和协议测试。可视化页面可以由另一成员开发，但不能让对方从后端实现反推 Contract。
|
||||
+1
-1
@@ -499,5 +499,5 @@ backend/app/retrieval/engine.py FTS/Vector/Hybrid 检索编排
|
||||
backend/app/services/index_service.py 全量重建与失败恢复
|
||||
backend/app/knowledge/parser.py Block ID 与 tags 解析语义
|
||||
backend/tests/test_retrieval.py 审阅回归测试
|
||||
docs/Knowledge与Retrieval-Core开发说明.md 模块开发说明
|
||||
docs/development/Knowledge与Retrieval-Core开发说明.md 模块开发说明
|
||||
```
|
||||
@@ -4,7 +4,7 @@
|
||||
> 审阅范围：FastAPI、Knowledge / Retrieval Core、Agent Core、Extension Core、Provider Adapter、公共接口和后端开发文档。
|
||||
> 文档用途：记录问题形成原因、实际影响、修复判断和落地方案，供后续开发文档、比赛材料与技术博客使用。
|
||||
|
||||
> 2026-08-30 状态补充：本文记录的缺陷均保持修复。此后又加入 Provider 预设、模型发现、DeepSeek/OpenAI 凭据解析和 Fernet 加密存储，当前完整后端回归基线为 71 项测试通过。
|
||||
> 2026-09-01 状态补充：本文记录的缺陷均保持修复。此后又加入 Provider 预设、模型发现、DeepSeek/OpenAI 凭据解析、Fernet 加密存储和 Agent Trace 持久化，当前完整后端回归基线为 81 项测试通过。
|
||||
|
||||
## 1. 审阅结论
|
||||
|
||||
@@ -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<string, unknown>
|
||||
}
|
||||
|
||||
export interface ToolCall {
|
||||
tool_call_id: string
|
||||
name: string
|
||||
|
||||
@@ -18,6 +18,10 @@ const eventLabels: Record<AgentEventType, string> = {
|
||||
PermissionRequired: '请求权限',
|
||||
Usage: '用量统计',
|
||||
Citation: '引用来源',
|
||||
ModelCallStarted: '模型调用开始',
|
||||
ModelCallCompleted: '模型调用完成',
|
||||
ModelCallFailed: '模型调用失败',
|
||||
PermissionResolved: '权限已处理',
|
||||
RunCompleted: '运行完成',
|
||||
RunFailed: '运行失败',
|
||||
RunCancelled: '运行取消',
|
||||
@@ -86,6 +90,10 @@ const detailLabels: Record<string, string> = {
|
||||
total_tokens: '令牌总数',
|
||||
status: '状态',
|
||||
duration_ms: '耗时（毫秒）',
|
||||
model_call_id: '模型调用 ID',
|
||||
parent_model_call_id: '上级模型调用 ID',
|
||||
finish_reason: '结束原因',
|
||||
decision: '授权决定',
|
||||
}
|
||||
|
||||
export function runStatusLabel(status?: AgentRunStatus): string {
|
||||
|
||||
@@ -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<OperationResponse>
|
||||
return apiClient.post(`/api/agent/runs/${runId}/cancel`)
|
||||
}
|
||||
|
||||
export async function getAgentTrace(
|
||||
runId: string,
|
||||
params?: { after_sequence?: number; limit?: number },
|
||||
): Promise<AgentTraceResponse> {
|
||||
return apiClient.get(`/api/agent/runs/${runId}/trace`, { params })
|
||||
}
|
||||
|
||||
export async function listTools(): Promise<ToolDefinition[]> {
|
||||
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'],
|
||||
|
||||
@@ -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',
|
||||
)
|
||||
})
|
||||
})
|
||||
@@ -1,12 +1,17 @@
|
||||
import { resolveApiUrl } from './apiClient'
|
||||
|
||||
export type SseEventHandler = (event: string, data: Record<string, unknown>) => void
|
||||
export type SseEventHandler = (
|
||||
event: string,
|
||||
data: Record<string, unknown>,
|
||||
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<string, string> = {
|
||||
@@ -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<string, unknown>
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user