7.0 KiB
以下只基于当前代码。
1. 当前请求 AgentClientMessage.oneof message 类型
协议定义了 8 类上行消息,agent_v1.proto:
run_request- 新建/恢复一次 Agent 执行。
- 当前提取
conversation_id、conversation state、action、用户消息、request context、模型、thinking effort、mode、subagent 信息。
prewarm_request- 建立运行态和 checkpoint,但不启动 provider。
conversation_action- 会启动 Run:
user_message、resume、summarize、start_plan、execute_plan。 - 会取消:
cancel。 - 其他 action 当前基本按 metadata 处理。
- 会启动 Run:
exec_client_message- 客户端工具执行数据或结果。
- 当前主要处理 Read、Write、Delete、Glob/Grep、Diagnostics、Ls、ShellStream、MCP、Subagent、WriteShellStdin、ForceBackgroundShell、ExecuteHook。
exec_client_control_messagestream_close、throw、heartbeat。
interaction_response- 当前处理 AskQuestion、CreatePlan、WebSearch、WebFetch、SwitchMode 的客户端响应。
kv_client_message- proto 支持
get_blob_result、set_blob_result;当前业务主要消费set_blob_result,用于 checkpoint blob 确认。
- proto 支持
client_heartbeat- 当前归为 metadata,不推进执行状态。
识别和内部 intent 映射集中在 inbound.go 与 service.go。
2. 当前返回 AgentServerMessage.oneof message 类型
外层 6 类全部有实际使用,agent_v1.proto:
-
interaction_updatetext_deltathinking_deltathinking_completedsummary_startedsummarysummary_completedtool_call_startedpartial_tool_calltool_call_deltatool_call_completedshell_output_deltaheartbeatturn_ended
-
exec_server_message- 服务端要求客户端执行工具。
- 当前包括 Read、Write、Delete、Grep、Ls、Diagnostics、ShellStream、WriteShellStdin、ForceBackgroundShell、MCP、MCP resource、Subagent、ExecuteHook。
-
exec_server_control_message- 当前只有
abort,取消尚未完成的客户端执行。
- 当前只有
-
conversation_checkpoint_update- 返回完整的
ConversationStateStructure投影。
- 返回完整的
-
kv_server_message- 当前主要发送
set_blob_args,要求客户端保存 checkpoint blob。
- 当前主要发送
-
interaction_query- 当前包括 AskQuestion、CreatePlan、WebSearch、WebFetch、SwitchMode。
构造入口分别在 events.go、exec bridge 和 interaction bridge。
另外,成功、取消、provider 错误不一定表现为 oneMessage:最终通过 StreamEvent.End 转换成 Connect end-stream 或结构化错误。
3. 当前需要处理的协议信息
传输层:
POST /aiserver.v1.BidiService/BidiAppend:Connect unary。POST /agent.v1.AgentService/RunSSE:Connect server stream。- RunSSE 响应头被强制兼容成
text/event-stream。 - 实际消息仍由 Connect handler 负责 framing。
Bidi 外层:
request_id:整条活动流的主键。append_seqno:同一 request 上行消息排序和去重。data:十六进制字符串,解码后才是AgentClientMessage protobuf。data_binary:proto 中存在,但当前实现没有使用。BidiAppendResponse:始终是空 ACK。
业务关联标识:
conversation_id:持久化会话与历史。request_id:一次活跃请求以及 Bidi/RunSSE 配对。turn_seq:会话中的轮次。model_call_id:一次 provider pass。tool_call_id:模型工具调用。ExecServerMessage.id + exec_id:客户端执行请求和回包关联。InteractionQuery.id:交互查询和响应关联。KvServerMessage.id:checkpoint blob 请求与确认关联。
还需要解析 conversation state、action、mode、requested model、thinking effort、request context、workspace/MCP/skill 信息。当前归一化后的协议载体是 InboundIntent。
4. 当前怎样维护 Bidi 和 RunSSE 状态
Bidi 顺序状态:
appendSequenceTracker以request_id建立状态。- 维护
next、processing、ready。 - 小于
next的消息视为重复并忽略。 - 大于
next的消息等待前序完成。 - Cursor 复用
request_id且重新从append_seqno=1开始时,会在空闲状态重置序列。 - 状态空闲十分钟后清理。
append_seqno <= 0会绕过这个顺序机制。
运行状态:
StreamBroker使用map[requestID]*ActiveStream。- 每个
ActiveStream保存 provider、phase、backlog、subscriber、pending exec、pending interaction、checkpoint 和工具运行状态。 - Bidi、provider event、timer 和 compaction event 都投递到该 stream 的单一 actor mailbox 串行处理。
- Phase 包括
idle、provider_running、waiting_external、awaiting_user、compacting、checkpointing、completed/failed/canceled。
RunSSE 状态:
- RunSSE 可以先于 Bidi 到达,此时 Broker 创建只有
request_id的占位 stream。 - 每个 RunSSE 连接注册独立 subscriber,但数据事实源是共享
Backlog []StreamEvent。 Publish先追加 backlog,再用容量为 1 的 signal 唤醒订阅者;signal 可以合并,但事件不会丢,因为客户端重新读取 backlog。- 每个连接从本地
cursor=0开始,所以重新连接会从头回放当前内存 backlog。 - backlog 暂时为空时,每 5 秒直接发送 heartbeat;heartbeat 不进入 backlog。
- 最后一个订阅者断开后,给活跃请求 30 秒重连宽限期,之后 actor 执行取消。
- 终态 stream 在无订阅者时保留 30 秒,然后从 Broker 删除。
见 broker.go 和 service.go。
关键结论:Bidi/RunSSE 的活动状态、backlog、cursor、pending exec/interaction 都是内存态;持久化的是 conversation history/checkpoint,不是活动流本身。进程重启后无法恢复原 RunSSE backlog 和正在等待的桥接请求。