WIP: 统一 AgentRun 提交流程与来源快照

[xerrors]

变更描述

🔥 务必注意，本项目不允许提交未经 “Human Review” 过的代码，不允许 Agent 未经人类审阅，直接提交 PR。不允许 Agent 直接提交回复。

Code is Cheap, Show Me Your Thoughts!

收敛消息型 AgentRun 的提交入口，让 Web Chat、Agent Call 与 Eval 共用同一套 Request/Run 创建、排队和来源快照流程。

背景与原因

此前 Web Chat 与 Invocation 分别维护提交逻辑，Request、Run、Conversation 和 Message 的创建边界存在重复，来源信息也主要依赖 Message metadata。随着异步调用、Eval 和后续 IM 接入扩展，这种双轨实现容易产生行为漂移。

实现方式

• 新增 run_submission_service.submit_run_command，统一 Web Chat、Agent Call 和 Eval 的消息型 Run 提交。
• Router 仅保留协议校验、输入转换、等待与响应装配，不再直接创建 Conversation、Request 或 Run。
• 在 AgentRunRequest 与 AgentRun 固化 source/channel/external_id/origin_metadata 来源快照。
• 将 Agent Invocation 拆分为 Call/Eval 两个 Router，同时保留现有 API 路径和响应结构。
• Eval 上下文继续使用 agent_invocation_meta.evaluation 传入 Message、worker 与 Langfuse。
• Resume、Subagent 与 Steer 生命周期保持不变。
• 重写相关仓储、Router、提交服务和 schema 演进单元测试，并更新 changelog。

flowchart TD
    Send["AgentChatComponent.handleSendMessage<br/>生成 request_id"]
    CreateRun["POST /api/agent/runs"]
    AgentCall["POST /api/agent-invocation/agent-call/runs"]
    Eval["POST /api/agent-invocation/eval/runs"]
    ResultAPI["POST /api/agent-invocation/agent-call/runs/result"]
    CallAdapt["agent_invocation_call_router<br/>解析 messages / 校验 stream<br/>async_mode=true: enqueue，立即返回<br/>async_mode=false: reject，等待终态"]
    EvalAdapt["agent_invocation_eval_router<br/>规范化 evaluation metadata<br/>固定 reject，等待终态"]
    ResumeDecision{"payload.resume 是否存在?"}
    SubmitCommand["run_submission_service.submit_run_command<br/>统一 RunSubmissionCommand"]

    Intake["agent_request_queue_service.intake_request"]
    LockConversation["锁 Conversation<br/>校验 uid + agent_slug + thread_id"]
    QueueState["读取 queued Request、active Run、最新 Run"]
    Persist["同一事务写入<br/>用户 Message + AgentRunRequest"]
    IntakeResult{"intake 结果"}
    Dispatch["_dispatch_locked_head<br/>创建 pending AgentRun<br/>Message 标为 dispatched<br/>Request 标为 dispatched"]
    DispatchCommit["finalize_dispatch<br/>COMMIT PostgreSQL"]
    QueueCommit["COMMIT PostgreSQL<br/>仅保留 queued Request"]
    RejectCommit["COMMIT PostgreSQL<br/>保留 rejected Request / Message"]
    EnqueueARQ["enqueue_agent_run<br/>再投递 ARQ"]

    QueuedResponse["返回 queued + request_events_url"]
    RejectedResponse["返回 rejected<br/>无 Request SSE / Run SSE"]
    RequestSSE["GET /api/agent/requests/{id}/events<br/>queued / position / heartbeat"]
    RunCreated["Request SSE: run_created<br/>携带 run_id + stream_url"]

    RunResponse["返回 dispatched + run_id + stream_url"]
    AsyncInvocationResponse["Agent Call async<br/>立即返回 pending/queued"]
    AwaitInvocation["await_agent_run_result<br/>等待 Agent Call sync / Eval 终态"]
    InvocationResponse["Call/Eval 响应适配<br/>choices 或评估结果"]
    ResultRead["get_agent_run_result<br/>读取 Run + output Message"]
    RunSSE["GET /api/agent/runs/{run_id}/events"]
    Worker["worker.process_agent_run<br/>pending -> running"]
    ChatService["stream_agent_chat<br/>构建 Runtime Context<br/>LangGraph 从 thread checkpoint 恢复"]
    EventMap["worker 映射 chunk -> Run event"]
    PersistOutput["保存 assistant/tool 消息<br/>回填 output_message_id"]
    RedisEvents["worker 写入 Redis Stream<br/>metadata/messages/custom/interrupt/error/end"]
    WorkerFinalize["worker 根据终止 chunk / 异常<br/>写 Run 终态"]
    Terminal["Run 写入 PostgreSQL 终态"]
    Next{"终态是否 completed?"}
    DispatchNext["dispatch_next_request<br/>派发下一条 Request"]
    StopQueue["failed/cancelled: 既有队列暂停<br/>interrupted: 统一 intake 被阻塞，需另建 resume Run"]

    ResumePath["create_agent_run_view<br/>校验 parent interrupted Run<br/>直接创建 run_type=resume"]
    ResumeCommit["_commit_and_enqueue<br/>COMMIT 后投递 ARQ"]

    Send --> CreateRun --> ResumeDecision
    AgentCall --> CallAdapt --> SubmitCommand
    Eval --> EvalAdapt --> SubmitCommand
    ResumeDecision -->|"否：普通 Chat"| SubmitCommand --> Intake
    Intake --> LockConversation --> QueueState --> Persist --> IntakeResult
    IntakeResult -->|"立即派发"| Dispatch --> DispatchCommit --> EnqueueARQ
    IntakeResult -->|"enqueue / steer 排队"| QueueCommit --> QueuedResponse --> RequestSSE
    IntakeResult -->|"reject 且不能立即派发"| RejectCommit --> RejectedResponse
    RequestSSE -->|"Request 被派发"| RunCreated --> RunSSE
    EnqueueARQ --> RunResponse --> RunSSE
    EnqueueARQ -->|"Agent Call async"| AsyncInvocationResponse
    EnqueueARQ -->|"Agent Call sync / Eval"| AwaitInvocation --> InvocationResponse
    ResultAPI --> ResultRead --> InvocationResponse
    EnqueueARQ --> Worker --> ChatService
    ChatService --> PersistOutput
    ChatService --> EventMap --> RedisEvents --> RunSSE
    ChatService --> WorkerFinalize --> Terminal --> Next
    Next -->|"是"| DispatchNext --> EnqueueARQ
    Next -->|"否"| StopQueue

    ResumeDecision -->|"是：恢复 interrupted Run"| ResumePath --> ResumeCommit --> EnqueueARQ

Loading

影响范围

• AgentRun Request/Run 持久化
• Web Chat 提交流程
• Agent Call/Eval API 适配
• worker/Langfuse 来源 metadata
• PostgreSQL 启动时 schema 演进

变更类型

• 新功能
• Bug 修复
• 文档更新
• 其他

测试

• 已在 Docker 环境测试
• 相关功能正常工作

相关日志或者截图

• docker compose exec api uv run --group test pytest test/unit/repositories/test_agent_run_repository.py test/unit/repositories/test_agent_run_request_repository.py test/unit/routers/test_agent_invocation_router_split.py test/unit/services/test_agent_invocation_router_adapters.py test/unit/services/test_run_submission_service.py test/unit/storage/test_postgres_manager_schema.py -q
  • 结果：31 passed
• docker compose exec api uv run --group test pytest test/e2e/test_agent_call_entrypoints_e2e.py -m e2e -q
  • 结果：1 passed
• 对相对 upstream/main 的 19 个变更 Python 文件执行 Ruff check 与 format check
  • 结果：全部通过
• git diff --check upstream/main...HEAD
  • 结果：通过

DEBUG：能力测试

额外扩展运行 worker 相关单元测试时，test_worker_startup_ensures_builtin_mcp_servers 因测试中的假数据库对象缺少 execute 失败；该测试不经过本次 Eval metadata 或统一提交链路，其余 18 项通过。本 PR 未宣称全量测试套件通过。

说明

• 未引入新的 IM 逻辑。
• 未改用独立 migration 系统，继续沿用项目现有的幂等 schema 演进机制。
• 未删除外部兼容 API 入口。
• GitHub Project 中未找到匹配任务条目，因此未创建或更新 Project 项。

贡献说明

本 PR 使用 Codex 辅助完成。贡献者已完成 Human Review，并明确确认已审阅两侧变更与 Review 结论：

• 检查实现与任务范围
• 检查敏感信息和安全风险
• 运行相关测试与静态检查
• 复核 Eval metadata 修复与下游消费契约

[gemini-code-assist]

Caution

The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased.

[gemini-code-assist]

Caution

The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased.

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.

[xerrors]

@codex review

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: f495613bb2

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".