byy/qiming
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
codex/client-refactor
qiming/qiming-rcoder/.qoder/repowiki/zh/content/API端点参考/代理进度流接口.md
Codex 4b1a580132 添加qiming-rcoder模块
2026-06-01 13:54:52 +08:00

17 KiB
Raw Permalink Blame History

代理进度流接口

**本文引用的文件** - [crates/agent_runner/src/router.rs](file://crates/agent_runner/src/router.rs) - [crates/agent_runner/src/handler/agent_session_notification.rs](file://crates/agent_runner/src/handler/agent_session_notification.rs) - [crates/rcoder/src/handler/agent_session_notification.rs](file://crates/rcoder/src/handler/agent_session_notification.rs) - [crates/shared_types/src/model/agent_session_notify.rs](file://crates/shared_types/src/model/agent_session_notify.rs) - [crates/agent_runner/src/service/session_cache.rs](file://crates/agent_runner/src/service/session_cache.rs) - [crates/agent_runner/src/proxy_agent/acp_agent.rs](file://crates/agent_runner/src/proxy_agent/acp_agent.rs) - [crates/rcoder/src/handler/chat_handler.rs](file://crates/rcoder/src/handler/chat_handler.rs) - [crates/shared_types/src/grpc/agent.rs](file://crates/shared_types/src/grpc/agent.rs)

目录

  1. 简介
  2. 项目结构
  3. 核心组件
  4. 架构总览
  5. 详细组件分析
  6. 依赖分析
  7. 性能考虑
  8. 故障排查指南
  9. 结论
  10. 附录

简介

本文件面向 RCoder 项目的“代理进度流”接口，聚焦于 GET /agent/progress/{session_id} 的 Server-Sent Events（SSE）实现。该接口提供实时的 AI 任务执行进度更新，涵盖代码生成、文件操作等事件类型；同时说明 SSE 文本流格式、事件类型分类、客户端重连机制，以及 session_id 的生成与有效期管理策略，并给出 JavaScript 客户端示例的对接思路与最佳实践。

项目结构

  • 该接口在两个服务中均有实现：
    • agent_runner：直接消费内部会话缓存，推送统一的 UnifiedSessionMessage。
    • rcoder：作为代理层，将外部容器的 SSE 流透传给客户端。
  • 路由注册位于 agent_runner 的路由文件中，暴露 /agent/progress/{session_id}。
graph TB
subgraph "Agent Runner 服务"
R["路由注册<br/>/agent/progress/{session_id}"]
H1["SSE处理器<br/>agent_session_notification"]
C["会话缓存<br/>SessionCache/SessionData"]
end
subgraph "Shared Types"
M["UnifiedSessionMessage<br/>统一消息模型"]
end
R --> H1
H1 --> C
C --> M

图表来源

章节来源

核心组件

  • SSE 处理器：负责建立 SSE 连接、发送心跳、将消息事件化并推送。
  • 会话缓存：维护每个 session_id 的消息通道与取消令牌，支持新连接覆盖旧连接。
  • 统一消息模型：将不同类型的会话更新统一为 UnifiedSessionMessage，便于前端解析。
  • 代理层（rcoder）：当需要透传外部容器的 SSE 时，将容器端 SSE 流透传至客户端。

章节来源

架构总览

SSE 进度流的总体交互如下：

  • 客户端发起 GET /agent/progress/{session_id} 建立 SSE 连接。
  • 服务端根据 session_id 获取或创建会话数据，建立消息通道与取消令牌。
  • 服务端立即发送一次心跳事件，随后周期性发送心跳，并将收到的统一消息转换为 SSE 事件推送。
  • 若连接被取消（如新连接建立或用户取消），旧连接会自然断开。
sequenceDiagram
participant Client as "客户端"
participant Handler as "SSE处理器"
participant Cache as "会话缓存"
participant Worker as "会话工作线程"
participant Model as "统一消息模型"
Client->>Handler : "GET /agent/progress/{session_id}"
Handler->>Cache : "创建/获取 SessionData 并建立新连接"
Handler->>Client : "发送初始心跳事件"
Worker-->>Handler : "推送 UnifiedSessionMessage"
Handler->>Client : "按事件类型发送事件如 prompt_start/prompt_end/tool_call 等"
Handler->>Client : "周期性发送心跳事件"
Handler-->>Client : "连接被取消/断开时终止推送"

图表来源

详细组件分析

SSE 文本流格式与事件类型

  • 文本流遵循标准 SSE 格式，每条消息包含事件类型与数据两部分。
  • 事件类型与 UnifiedSessionMessage 的映射关系：
    • prompt_start：会话开始
    • prompt_end：会话结束（含结束原因）
    • agent_message_chunk：Agent 文本响应片段
    • agent_thought_chunk：Agent 思考过程片段
    • tool_call：工具调用通知
    • tool_call_update：工具调用状态更新
    • available_commands_update：可用命令更新
    • current_mode_update：当前模式更新
    • heartbeat：心跳消息
  • 数据体为 JSON，结构遵循 UnifiedSessionMessage，包含 session_id、message_type、sub_type、data、timestamp。

章节来源

SSE 处理器（agent_runner）

  • 建立新连接：为每个 session_id 创建新的 SessionData 并插入缓存，确保每次连接全新开始。
  • 发送心跳：首次立即发送一次心跳事件，随后以固定间隔（如 30 秒）发送心跳。
  • 事件分发：将收到的 UnifiedSessionMessage 按 message_type 与 sub_type 转换为对应的事件名并推送。
  • 取消与断开：监听取消令牌与通道关闭，及时退出循环并记录日志。
flowchart TD
Start(["建立连接"]) --> NewSession["创建/获取 SessionData 并建立新连接"]
NewSession --> SendHB["发送初始心跳事件"]
SendHB --> Loop{"循环接收消息"}
Loop --> |收到消息| MapEvent["按消息类型映射事件名"]
MapEvent --> Yield["yield SSE 事件"]
Yield --> Loop
Loop --> |取消令牌触发| Close["断开连接"]
Loop --> |通道关闭| Close
Close --> End(["结束"])

图表来源

章节来源

会话缓存与连接管理

  • SessionData：持有命令通道、当前发送器与取消令牌，支持创建新连接并设置当前发送器与取消令牌。
  • SessionWorker：维护环形缓冲区，按需缓冲非心跳消息，并将消息实时推送到当前发送器。
  • 项目-会话映射：ensure_project_session 确保同一项目仅对应一个活跃会话，必要时清理旧会话数据。
  • 主动关闭：close_current_connection 主动触发取消令牌并丢弃发送器，促使接收端感知断开。
classDiagram
class SessionData {
+new(max_size)
+create_new_connection(buffer_size)
+push_message(message)
+close_current_connection()
}
class SessionWorker {
+spawn(max_size, command_rx, current_sender, current_cancel)
+run()
}
class UnifiedSessionMessage {
+session_id
+message_type
+sub_type
+data
+timestamp
}
SessionData --> SessionWorker : "spawn"
SessionWorker --> UnifiedSessionMessage : "推送"

图表来源

章节来源

代理层 SSE 透传（rcoder）

  • 当需要代理外部容器的 SSE 时，rcoder 会根据 session_id 查找对应容器，建立到容器 SSE 端点的连接，并将容器的 SSE 流透传给客户端。
  • 透传逻辑按双换行符切分事件，解析 event/data 字段，避免重复 data: 前缀，然后直接透传事件。
  • 错误处理：容器连接失败或读取失败时，发送 error 事件并记录日志。
sequenceDiagram
participant Client as "客户端"
participant Proxy as "rcoder SSE代理"
participant Container as "容器SSE端点"
Client->>Proxy : "GET /agent/progress/{session_id}"
Proxy->>Container : "建立到容器SSE的连接"
Container-->>Proxy : "SSE事件流按双换行符分隔"
Proxy->>Proxy : "解析event/data并去重"
Proxy-->>Client : "透传SSE事件"
Proxy-->>Client : "错误时发送error事件"

图表来源

章节来源

session_id 的生成与有效期管理

  • 生成机制：在聊天处理流程中，当创建项目会话状态时，会为项目设置 session_id，并更新最后活动时间。项目-会话映射通过 ensure_project_session 保证唯一性。
  • 有效期管理：服务端未对 session_id 设置硬性过期时间；连接断开或取消会触发清理。若需要过期控制，可在业务侧增加会话超时策略（例如定期清理长时间无活动的会话）。
  • 会话切换：当同一项目的新会话建立时，旧会话数据会被清理，确保不会跨会话污染。

章节来源

客户端重连机制与最佳实践

  • 心跳检测：服务端会周期性发送 heartbeat 事件，前端应基于心跳判断连接是否仍活跃。
  • 自动重连：建议实现指数退避的自动重连策略，避免频繁重试造成压力。
  • 事件处理：根据事件类型（prompt_start/prompt_end/tool_call 等）更新 UI 状态，避免阻塞主线程。
  • 错误处理：监听连接错误与 SessionPromptEnd 中的错误信息，向用户反馈并引导重试或取消。

章节来源

与 gRPC 订阅接口的关系

  • 项目中提供了基于 gRPC 的订阅进度流接口（subscribe_progress），可替代传统的 /agent/progress/{session_id} SSE 接口。
  • 若采用 gRPC 方案，前端需使用 gRPC 客户端订阅进度流，事件类型与数据结构保持一致。

章节来源

依赖分析

  • 路由依赖：/agent/progress/{session_id} 在 agent_runner 的路由中注册。
  • 处理器依赖：SSE 处理器依赖会话缓存与统一消息模型。
  • 代理层依赖：rcoder 的代理处理器依赖容器映射与 SSE 透传逻辑。
  • 项目-会话映射：通过 ensure_project_session 保证同一项目仅有一个活跃会话。
graph LR
Router["路由"] --> SSEHandler["SSE处理器"]
SSEHandler --> SessionCache["会话缓存"]
SessionCache --> UnifiedMsg["统一消息模型"]
ProxyHandler["rcoder代理处理器"] --> ContainerSSE["容器SSE端点"]

图表来源

章节来源

性能考虑

  • 心跳频率：默认 30 秒一次，可根据网络状况调整，避免过多心跳占用带宽。
  • 缓冲策略：SessionWorker 使用环形缓冲区，非心跳消息优先缓冲，心跳消息直通，减少延迟。
  • 取消与断开：通过 CancellationToken 与通道关闭快速退出，避免资源泄漏。
  • 透传效率：rcoder 代理层按双换行符切分事件，避免重复解析，提高透传效率。

章节来源

故障排查指南

  • 404 未找到容器：当 session_id 对应的活跃容器不存在时，返回 404 并携带错误码。
  • 建立 SSE 连接失败：容器 SSE 连接失败或读取失败时，发送 error 事件并记录错误。
  • 连接被取消：当新连接建立或用户取消任务时，旧连接会主动断开。
  • 心跳缺失：若长时间未收到 heartbeat，应触发重连流程。

章节来源

结论

RCoder 的代理进度流接口通过 SSE 提供了稳定、低延迟的实时进度推送能力。其核心在于：

  • 统一消息模型与事件类型映射，便于前端解析；
  • 会话缓存与取消令牌机制，确保连接可控与资源回收；
  • 代理层透传能力，满足多容器场景；
  • 心跳与断开策略，保障连接健康与稳定性。

建议在前端实现自动重连与指数退避、基于事件类型的状态机更新，并结合业务侧的会话超时策略，构建健壮的实时进度体验。

附录

API 定义与示例

  • 端点：GET /agent/progress/{session_id}
  • 内容类型：text/event-stream
  • 响应：SSE 事件流，事件类型与数据体见“SSE 文本流格式与事件类型”。

章节来源

JavaScript 客户端对接要点（概念性说明）

  • 使用浏览器内置 EventSource API 建立连接，监听不同事件类型并更新 UI。
  • 实现自动重连：监听 error 与 close 事件，采用指数退避策略重连。
  • 心跳检测：收到 heartbeat 事件后刷新心跳计时器，超时则触发重连。
  • 错误处理：监听 error 事件与 SessionPromptEnd 中的错误信息，向用户反馈并引导操作。

[本节为概念性说明，不直接分析具体源码文件]

Reference in New Issue View Git Blame Copy Permalink