diff --git a/.doc/ops/short-term/2026-09-28-subagent-process-view-design.md b/.doc/ops/short-term/2026-09-28-subagent-process-view-design.md new file mode 100644 index 00000000..546574df --- /dev/null +++ b/.doc/ops/short-term/2026-09-28-subagent-process-view-design.md @@ -0,0 +1,54 @@ +# 任务:小队过程流式展示与 subagent 子页面设计 + +## 状态 + +- status: done(2026-09-28) +- owner: Cursor +- type: 方案研究(设计文档,无代码实施) +- created: 2026-09-28 +- base_dev_sha: 4c42d1bf74e262930bcf127e517ab6b76a0dda71 +- branch: feature/subagent-process-view-design-2026-09-28 +- worktree: .worktrees/subagent-process-view-design-2026-09-28 +- write_scope: `.doc/system-design/`、`.doc/ops/short-term/`(本文件) +- mutex: subagent-process-view-design + +## 用户目标与授权 + +> 是否能让助手的工作过程,像豆包、kimi、codex 等工具一样,输出思考、工具调用、过程内容输出等信息;使用流式输出,持续更新助手的工作过程。多智能体小队先设计如何使用 subagent 子页面来展示上面那样的信息,但可以先不实现。 + +(来源:2026-09-28 Cursor 会话,用户已确认计划) + +- 授权范围:撰写系统设计文档;**不实施任何代码变更**(用户明示"可以先不实现") +- 交付终点:设计文档评审通过(可审产物验收) +- 禁止项:修改 Go/TS 代码、数据库迁移、环境操作、推送远端 + +## 需求重叠探测 + +- 2026-09-28 `catalog search`(关键词:流式输出/思考/工具调用/subagent 页面、streaming agent thought tool call、小队过程可视化流式)均无命中 +- 既有关联文档为归档/演进方向(2026-09-08 三层信息架构、小队编排 §18.5),无进行中任务重叠 + +## 研究输入 + +- 三路并行只读探索(后端流式链路 / squad+taskrun 架构 / 前端聊天架构)+ 主线直接调研,交叉验证 +- 关键现状结论:协议/持久化/前端骨架已就绪,七处断点导致过程数据(思考链、工具调用、增量输出)无法到达前端;前端无 subagent 页面载体 + +## 交付物 + +- `.doc/system-design/2026-09-28-agent-process-streaming-and-subagent-view-design.md`(460 行,A–H 八章:目标参照/协议扩展/七断点打通/subagent 子页面/分期路线/风险边界/断点对照表/参考文件) +- 已登记 `.doc/system-design/README.md` 报告清单(0c 条目) + +## 验证结果(2026-09-28 实测) + +- 行号核对:设计文档引用的 14 处 文件:行 全部与真实代码一致(sed 逐处核对:stream/types.go:68、pinative.go:102、manager.go:364、loop.go:178-183、backend.go:38、runner.mjs:600/24、sse.go:476、workflow.go:543、useTaskRunAttempts.ts:474、taskRuns.ts:113、session.go:624、phase_snapshot.go:225 等) +- 内链核对:文档内 21 个代码文件相对链接、6 个 .doc 文档引用全部存在 +- 术语修正:stripThinkBlocks 剥离对象修正为 `…`(原文误写为特殊符号) +- 无硬编码模型名(grep MiniMax/M2.7 零命中于方案正文,仅协议描述提及 provider 家族作为 reasoning_content 语义来源) +- 产品 PI 测试 NOT REQUIRED(依据:验证契约「范围判定」——纯设计文档,无运行模板/提示词/业务变更;本任务不产生产品代码) + +## 五态收口 + +- review_state: done(设计文档自检完成;待用户评审确认) +- integration_state: not_applicable(不合入代码;文档随本任务分支提交) +- deployment: not_applicable +- cleanup: pending(worktree 待独立维护者复查后清理) +- remote_sync: not_applicable(未授权 push) diff --git a/.doc/system-design/2026-09-28-agent-process-streaming-and-subagent-view-design.md b/.doc/system-design/2026-09-28-agent-process-streaming-and-subagent-view-design.md new file mode 100644 index 00000000..f15130b2 --- /dev/null +++ b/.doc/system-design/2026-09-28-agent-process-streaming-and-subagent-view-design.md @@ -0,0 +1,460 @@ +# 智能体工作过程流式展示与 subagent 子页面设计 + +> 版本:v1.0(设计稿,未实施) +> 日期:2026-09-28 +> 任务:subagent-process-view-design-2026-09-28 +> 作者:Cursor(owner),三路并行代码探索 + 主线调研交叉验证 +> 交付物性质:系统设计文档,**不含任何代码实施**(用户明示"可以先不实现") + +--- + +## A. 目标与参照 + +### A.1 用户需求 + +1. 让助手的工作过程像豆包、kimi、codex 等工具一样对外输出:**思考(thinking)、工具调用(tool call)、过程内容输出**三类信息; +2. 使用**流式输出**持续更新助手的工作过程,而非终态一次性呈现; +3. 多智能体小队设计 **subagent 子页面**承载上述展示(先设计,不实现)。 + +### A.2 参照产品的展示形态拆解 + +| 产品 | 思考展示 | 工具调用展示 | 增量输出 | 交互模式 | +|---|---|---|---|---| +| 豆包 | "深度思考"折叠块:思考中呼吸态滚动,完成后折叠为一行摘要,可点击展开全文 | 工具卡片:图标+工具名+入参摘要+结果摘要+耗时 | 正文逐 token 流式 | 主流内嵌折叠,不打断阅读 | +| kimi | 推理过程折叠区,默认收起,进行中有进度感 | 搜索/工具步骤条 | 逐段流式 | 同上 | +| codex(CLI/Web) | 思考行内灰字流式 | 命令执行块(命令+输出+退出码) | 流式 | 工程向,密度更高 | + +共同设计语言:**默认折叠(低认知负荷)→ 状态可感知(进行中动效)→ 按需展开(全文/详情)→ 归档后可回放**。 + +### A.3 本仓已有基座与约束 + +本设计不是从零发明,而是延伸既有三层信息架构(2026-09-08 streaming-events 设计,已归档但语义仍有效): + +- **L1 phase/status**("现在在干嘛"):已实现并持久化(`task_run_events`); +- **L2 summary/delta/briefing**("结论是什么"):已实现(delta 为整段伪流式); +- **L3 detail**("推演依据"):**协议已定义(`DetailPayload.category: agent_thought/tool_call/step_output/log`),但无生产者、双重丢弃**——本设计的主战场。 + +既有 UX 约束必须继承: + +1. **不打扰原则**:detail 默认折叠,不默认展开思考链/工具日志(2026-09-09 chat-wait-error-retry 设计 P1 明确); +2. **透明可追溯**:详细思考与草稿不丢失,按需可查(工作区文件留存兜底); +3. **治理红线**:过程展示只是可视化,**不改变验收证据边界**——PI 叙事、模型输出不能冒充 PASS/测试证据(AGENTS.md 安全与证据)。 + +### A.4 概念区分(重要) + +本设计的 "subagent" 指**产品内小队成员**(`internal/squad` 的 AgentSpec:organizer/regulatory/writer/leader 等 8 标准专员 + ad_hoc 派生专员),**不是**本仓开发流程的工程工具 PI 子任务派发(`scripts/pi_subagents.py` + `.doc/ops/2026-09-24-pi-subagent-executor.md`)。两者名字相近但完全不同域,实施时不得混用数据源。 + +--- + +## B. 过程事件协议扩展 + +### B.1 现状协议回顾 + +`internal/stream/types.go` 的 `StreamEvent` 已定义 12 种事件类型;其中 `DetailPayload`: + +```go +// internal/stream/types.go:68 +type DetailPayload struct { + Category string `json:"category"` // agent_thought, step_output, tool_call, log + StepID string `json:"step_id,omitempty"` // 步骤标识 (如 tasks/01_regulatory) + Role string `json:"role,omitempty"` // 对应角色 + Title string `json:"title"` + Content string `json:"content"` // 完整推演正文或技术草稿 + Collapsible bool `json:"collapsible"` +} +``` + +现状问题: + +1. 只有 `Content` 全文单帧,无增量模式(思考链逐 token 需要数百帧,全文帧会放大 N 倍流量); +2. 无 `RunID` 冗余(事件重放时按 run 已隔离,可省;但 live 合并需要前端从通道上下文取); +3. 并行步骤下 `Role`/`StepID` 已有,但前端折叠键没有用它们(见 B.4)。 + +### B.2 DetailPayload 扩展(增量模式) + +```go +// 设计草案(不实施) +type DetailPayload struct { + Category string `json:"category"` // agent_thought | tool_call | step_output | log(沿用四类) + StepID string `json:"step_id,omitempty"` // 步骤标识(如 tasks/03_regulatory) + Role string `json:"role,omitempty"` // 成员角色 key(如 regulatory) + RunID string `json:"run_id,omitempty"` // 冗余 run 标识(重放通道按 run 已隔离,可选) + Seq int64 `json:"seq,omitempty"` // 同一 (category, step_id, role, item_id) 流内的序号 + ItemID string `json:"item_id,omitempty"` // 聚合项 ID:一个思考块/一次工具调用/一份产出 + // 增量模式三选一: + Delta string `json:"delta,omitempty"` // 增量帧:追加到 ItemID 项 + Content string `json:"content,omitempty"` // 终帧/快照帧:ItemID 项的完整内容(或无增量能力的发送方) + Done bool `json:"done,omitempty"` // 终帧标记(ItemID 项结束) + Title string `json:"title,omitempty"` + Collapsible bool `json:"collapsible"` + // tool_call 特化字段(Category == "tool_call" 时): + Tool *ToolCallInfo `json:"tool,omitempty"` +} + +type ToolCallInfo struct { + Name string `json:"name"` // write_file / read_file / grep ... + Status string `json:"status"` // calling | succeeded | failed + Input string `json:"input,omitempty"` // 入参摘要(JSON,可能分帧) + Output string `json:"output,omitempty"` // 结果摘要(可能分帧) + DurationMs int64 `json:"duration_ms,omitempty"` +} +``` + +**增量语义**(关键设计决策): + +- 发送方优先逐帧 `Delta`;`Content` 用于两类场景:(a) 无流能力的发送方(如既有 `step_output` 整段产出),(b) 周期性**快照对账帧**(防止增量丢帧后内容漂移); +- 客户端按 `(StepID, Role, Category, ItemID)` 聚合:`Delta` 追加、`Content` 覆盖、`Done` 定稿; +- 节流(发送方义务,见 F.1):逐 token 事件合并为 ~100–300ms 批帧,单帧上限(如 8KB)超限分帧。 + +### B.3 llm 层事件扩展(reasoning 链) + +`internal/agentx/llm` 现有事件只有 `text_delta/tool_call_delta/message_end/error`。OpenAI 兼容国产模型(MiniMax/DeepSeek 系)在 delta 中返回 `reasoning_content` 字段(思考链)。扩展: + +```go +// internal/agentx/llm/types.go 设计草案 +const ( + KindTextDelta EventKind = "text_delta" + KindReasoningDelta EventKind = "reasoning_delta" // 新增 + KindToolCallDelta EventKind = "tool_call_delta" + ... +) + +type Event struct { + Kind EventKind + Text string + ReasoningDelta string // 新增:reasoning_content 增量 + ... +} +``` + +`ParseSSE`(internal/agentx/llm/sse.go)补充解析 `delta.reasoning_content` → `KindReasoningDelta`。 + +**容器侧对应**:`deploy/pi-agent/runner.mjs:24` 的 `stripThinkBlocks` 目前把 `…` 块从产物中剥离(丢弃)。设计上不改变"正文不含 think"的既有约定,而是在**剥离前**把 think 块作为 reasoning 事件分流转发(见 C.5)。 + +### B.4 并行步骤归属路由(修复既有缺陷) + +现状 `PhasePayload` 折叠键为 `phase_id#step_index`([taskrun/phase_snapshot.go:225](../../../internal/taskrun/phase_snapshot.go) 与 [web/src/engine/phaseEventMapper.ts:65](../../../web/src/engine/phaseEventMapper.ts)),**并行组成员**(多个 role 同时 `step_executing`)若无 step_index 会互相覆盖/误折叠。 + +设计:折叠键升级为 `phase_id#step_index@role`(role 存在时必选参与键)。前端 `applyPhaseEvent` 与后端 `FoldPhaseEvents` 同步修改,语义保持对齐(两处折叠逻辑本就要求一致,见 phase_snapshot.go 注释)。detail 事件天然带 `role/step_id`,聚合不受影响。 + +### B.5 通道选择:沿用 stream/StreamEvent,不新增域事件 + +理由: + +1. `events.StaticContracts`(internal/events/registry.go)是静态 allowlist(13 种业务事件 + YAML parity 测试),加 agent 过程事件需改 Go+YAML+测试三处,且域事件总线面向**业务投影**(通知/阶段推进),不面向用户实时流; +2. `task_run_events`(JSONB payload)+ `after_seq` 重放已为过程事件准备好持久化与断线恢复; +3. chat SSE 通道(链路 A)无持久化,但 B 轨参谋对话的过程展示可接受"live-only + 终态落库"(chat_turns 记录终态),与现状一致。 + +--- + +## C. 后端链路打通设计(七断点逐条) + +### C.0 总览 + +``` +【链路 B:长任务(本设计主通道,先做)】 +worker goroutine (taskrun.Manager.runWorker) + └ bizdev.ExecuteTask → squad.WorkflowDriver.runOneStep + └ agent.Backend.Execute(OnEvent 回调) ← 断点 4 + ├ pi-native: agentx.Run(OnEvent) ← 断点 2/3(直通) + │ └ llm.Stream: reasoning/text/tool delta ← 断点 1 + └ pi-agent: POST runner /v1/run/stream ← 断点 5(NDJSON 流) + └ Streamer.EmitDetail(category, stepID, role, ...) ← 断点 6 之前:detail 已能出 Streamer + └ Manager.appendEvent → task_run_events (seq) ← 持久化(已就绪) + └ SSE: GET /tasks/runs/{id}/events ← 透传(已就绪) + └ 前端 detail 渲染分支 ← 断点 7 + +【链路 A:聊天 SSE(P4 再做)】 +chat.Session → coordinator.generateDialogueReply + └ dialogueBackend.Execute(OnEvent) → token 级 EmitDelta(替换一次性 EmitDelta 伪流式) +``` + +### C.1 断点 1:reasoning_content 解析 + +- 文件:[internal/agentx/llm/sse.go](../../../internal/agentx/llm/sse.go)(`sseChoice.Delta` 增 `ReasoningContent string \`json:"reasoning_content"\``)、[internal/agentx/llm/types.go](../../../internal/agentx/llm/types.go)(`Event` 增 `ReasoningDelta` + `KindReasoningDelta`) +- 语义:reasoning 与 text 分开累计;`message_end` 的 `Message.Content` 仍只含 text(与 runner `stripThinkBlocks` 语义对齐:正文无 think)。 +- 兼容性:无 reasoning 的模型零行为变化(字段空即无事件)。 + +### C.2 断点 2:agentx 循环转发 tool_call_delta + +- 文件:[internal/agentx/loop.go:178-183](../../../internal/agentx/loop.go) `consume()`:`case llm.KindToolCallDelta` 增加转发(emit 一个 `tool_call_delta` Event,带 ToolName/ArgumentsDelta);`KindReasoningDelta` 同理转发 `reasoning_delta`。 +- `agentx.Event`(loop.go:19)增字段:`Kind string`(llm 层事件细分)或复用 `Type` 前缀约定(`text_delta`/`reasoning_delta`/`tool_call_delta`/`tool_execution_start`/`tool_execution_end`…已够用,推荐复用 Type)。 + +### C.3 断点 3:Backend 接口过程回调 + +**设计取舍**:不改 `Backend.Execute` 签名(改签名会波及 stub/fallback/limiter 全部实现与调用方),改为 **Request 增可选回调**: + +```go +// internal/agent/backend.go 设计草案 +type Request struct { + ... + // OnEvent 过程事件回调(nil = 不需要过程,行为与现状完全一致) + // 实现方必须保证:回调在 Execute 返回前完成(同步或等待收尾),无泄漏 goroutine。 + OnEvent func(ev ProcessEvent) +} + +// ProcessEvent 对 agentx.Event 与 runner NDJSON 事件的统一投影 +type ProcessEvent struct { + Type string // text_delta | reasoning_delta | tool_call_delta | tool_execution_start | tool_execution_end | message_end | ... + Text string // delta 文本(Type 为 *delta 时) + Tool string // 工具名 + Detail string // 工具参数/结果摘要(tool_execution_* 时) +} +``` + +- pi-native([internal/agent/pinative.go](../../../internal/agent/pinative.go)):`agentx.Run` 的 `OnEvent` 双写——既拼 `execution.log`(现状保留),又投影到 `req.OnEvent`; +- `WrapWithConcurrencyLimit` 透传 Request 即透传回调,无需改; +- fallback/stub:fallback 转发 primary 的回调事件即可;stub 无事件。 + +### C.4 断点 4:squad 层桥接 Streamer + +[internal/squad/workflow.go](../../../internal/squad/workflow.go) `runOneStep`(构造 `agent.Request` 处,约 :543): + +```go +// 设计草案 +if req.OnEvent == nil { + req.OnEvent = func(ev agent.ProcessEvent) { /* 仅拼日志(现状) */ } +} +// 注入流式桥(streamer 可为 nil,安全) +req.OnEvent = chainEvents(req.OnEvent, squadDetailBridge(streamer, step)) +``` + +`squadDetailBridge`(新函数,~40 行)把 `ProcessEvent` 翻译为 `EmitDetail`: + +| agentx/runner 事件 | EmitDetail 映射 | +|---|---| +| `reasoning_delta` | category=agent_thought, delta 帧, ItemID=`step.ID#thought` | +| `text_delta` | category=agent_thought 的草稿子项(或 P2 可选:不改,任务正文走终态 step_output,避免与最终 deliverable 重复)→ **推荐 P0 不发 text_delta,仅 reasoning + tool** | +| `tool_call_delta` | category=tool_call, delta 帧(入参增量), ItemID=`step.ID#tool` | +| `tool_execution_start` | tool_call 项 status=calling 帧 | +| `tool_execution_end` | tool_call 项 status=succeeded/failed + Output 摘要 + duration 帧 | +| `message_end` | 对应 ItemID 的 done 帧 | + +节流在 bridge 内做(100–300ms 合并、8KB 分帧、工具输出截断 N KB——工具结果可能巨大,只发首尾摘要)。 + +**B 轨参谋(链路 A,P4)**:[internal/coordinator/dialogue_llm.go](../../../internal/coordinator/dialogue_llm.go) `generateDialogueReply` 同样注入 `OnEvent`,把 `text_delta` 逐帧转 `streamer.EmitDelta`(替换 [internal/chat/session.go:624](../../../internal/chat/session.go) 一次性 `EmitDelta(NaturalReply)`),`reasoning_delta` 转 `EmitDetail(agent_thought)`。注意 chat SSE 无持久化——断线后靠 `chat_turns` 终态查询(现状机制),过程帧不可恢复(可接受,与豆包/kimi 行为一致)。 + +### C.5 断点 5:远程 runner 流式端点 + +[deploy/pi-agent/runner.mjs](../../../deploy/pi-agent/runner.mjs) 增加端点(不动 `/v1/run` 既有契约——`ResultV1` typed 校验、wire 预算 8MB 等全部保留): + +``` +POST /v1/run/stream Content-Type: application/x-ndjson(chunked) +请求体 = /v1/run 同构 + stream: true +响应 = 逐行 NDJSON: + {"type":"proc","ev":{...pi RPC stdout 事件行...}} // 过程事件(原样转发已解析的行) + ...(执行中持续 flush) + {"type":"result","body":{...与 /v1/run 完全相同的 typed 结果...}} // 终帧 +``` + +- runner 侧改动集中在 `handleRun`:stdout `on("data")` 里既有的逐行 JSON 解析循环(runner.mjs:365-388)旁路一个转发器;`stripThinkBlocks` 之前先把 think 块作为 `{"type":"reasoning_delta"}` proc 帧发出; +- Go 侧 [internal/agent/piagent.go](../../../internal/agent/piagent.go) `executeRemote`:`req.OnEvent != nil` 时改走流式端点——`http.Client` 无总超时的流式变体(`Timeout` 由 ctx 控制),bufio 逐行读:`proc` 帧 → `req.OnEvent` 投影;`result` 帧 → 走既有 `DecodeRemoteRun` 校验路径(**终态校验逻辑零改动**,保证治理边界不放松); +- 降级:流式端点 404(旧 runner)→ 回落 `/v1/run` 一次性调用 + 无过程事件(`capabilities` 探测亦可,但 404 回落更简单)。 + +### C.6 断点 6:includeDetails 参数化 + +两处硬编码/丢弃: + +1. [internal/taskrun/manager.go:364](../../../internal/taskrun/manager.go) `stream.WithListener(ctx, false, listener)`——worker 侧 Streamer 是否发 detail 由**创建时**决定。设计:`CreateAndStart`/`StartClaimedRun` 透传 `IncludeDetails`(`RunRecord` 或 `CreateOptions` 增字段;默认 false 保持现状)。**注意**:detail 是否持久化与是否推送解耦——建议 **detail 一律持久化**(`task_run_events` JSONB 足够),`includeDetails` 只控制 live 推送与 chat 通道;重放端点用 query `include_details=0` 过滤,历史 subagent 页面仍可全量回放。取舍理由:持久化膨胀可控(F.1 节流后估算),而"刷新后过程消失"违背透明可追溯原则; +2. [internal/server/sse.go:476](../../../internal/server/sse.go) `_ = includeDetails`——`StreamChatRequest.IncludeDetails` 已存在(chat 通道)但 task 通道读取后丢弃;接线到 worker 创建。 + +### C.7 断点 7:前端协议与分支 + +- [web/src/api/chatStream.ts](../../../web/src/api/chatStream.ts) `StreamEvent` 增 `detail?: { category?: string; step_id?: string; role?: string; item_id?: string; seq?: number; delta?: string; content?: string; done?: boolean; title?: string; collapsible?: boolean; tool?: { name?: string; status?: string; input?: string; output?: string; duration_ms?: number } }`; +- [web/src/hooks/useTaskRunAttempts.ts](../../../web/src/hooks/useTaskRunAttempts.ts) `onEvent` switch 增 `case 'detail'`:聚合进消息的 `processItems`(见 D.4),不进正文。 + +--- + +## D. subagent 子页面设计(重点) + +### D.1 入口与路由 + +**决策:扩展 hash 路由,不引入 react-router。** + +理由: + +| 方案 | 优点 | 代价 | +|---|---|---| +| A. hash 扩展(推荐) | 复用 [web/src/engine/anchorNavigator.ts](../../../web/src/engine/anchorNavigator.ts) 既有解析/监听先例(`#/s/{sessionId}/m/{messageId}`);零新依赖;与"工作台为唯一应用视图"的产品形态匹配 | 顶层 view 状态机要加一态;深链参数解析手写 | +| B. react-router | 标准 | 新依赖 + 全仓路由心智迁移;App.tsx 1758 行重构风险;本需求只有一个子页面,收益不成比例 | + +路由形态: + +``` +#/s/{sessionId}/run/{runId}/agents → 子页面(默认选中当前 running/最近完成成员) +#/s/{sessionId}/run/{runId}/agents/{roleId} → 子页面(选中指定成员) +``` + +实现骨架(设计草案): + +- App.tsx 顶层 view 状态机扩展:`'workbench' | 'admin' | 'subagents'`;`parseAnchorHash` 扩展识别 `/run/{runId}/agents` 段; +- `handleSubagentsView` 由 TaskRunCard 的"查看小队过程"入口触发(写 hash 触发 hashchange,与消息深链同一机制); +- 子页面经 `OnDemand`([web/src/components/lazy/OnDemand.tsx](../../../web/src/components/lazy/OnDemand.tsx))懒加载,**props 最小化**:仅 `runId`、`sessionId`、`onExit`(返回 workbench)——规避 App.tsx props 钻透:子页面自持 SSE 订阅与状态,不依赖父容器消息流。 + +### D.2 页面信息架构 + +``` +┌──────────────────────────────────────────────────────────────┐ +│ 小队过程 · {任务标题} · run:{runId 截断} [刷新] [返回对话] │ +├──────────────┬───────────────────────────────────────────────┤ +│ 成员卡列表 │ 选中成员(role: regulatory 法规专员) │ +│ │ ┌ 状态条: ⚙️ 正在执行 · 已用 2m13s · 第2/5步 ┐ │ +│ 👤 organizer │ └──────────────────────────────────────────┘ │ +│ ✅ 完成 1m2s│ ┌ ThinkingBlock(默认折叠)──────────────┐ │ +│ 👤 searcher │ │ 💭 思考中…(流式增量,呼吸态) │ │ +│ ✅ 完成 3m4s│ └───────────────────────────────────────┘ │ +│ 👤 regulatory│ ┌ ToolCallCard ─────────────────────────┐ │ +│ ⚙️ 执行中 │ │ 🔧 read_file tasks/01.../output.md │ │ +│ 👤 writer │ │ ✅ 213ms · 12KB │ │ +│ ⏳ 等待 │ └──────────────────────────────────────┘ │ +│ 👤 leader │ ┌ ThinkingBlock(第二段)──────────────┐ │ +│ ⏳ 等待 │ ... │ │ +│ │ ┌ 步骤产出 output.md(终态 step_output)┐ │ +│ [成员详情] │ │ 📄 阶段性产出(可展开全文) │ │ +│ │ └──────────────────────────────────────┘ │ +├──────────────┴───────────────────────────────────────────────┤ +│ 事件时间线(跨成员 phase 事件折叠,复用 applyPhaseEvent 语义) │ +└──────────────────────────────────────────────────────────────┘ +``` + +- **左栏数据源**:`squad.json`(成员编制:role/Name/Responsibilities——工作区文件,经新只读端点或既有 workspace API 投影)+ phase/detail 事件聚合出状态与耗时(`step_executing`/`step_completed`/`step_failed` by role); +- **右栏数据源**:detail 事件流按 `role == selected` 过滤,聚合为 `processItems`(见 D.4); +- **底部时间线**:全量 phase 事件(复用 `TaskRunCard` 折叠逻辑与 `PhaseEvent` 类型)。 + +### D.3 三类过程组件 + +| 组件 | 状态 | 内容来源 | 交互 | +|---|---|---|---| +| `ThinkingBlock` | thinking(呼吸态+增量滚动)/ done(折叠为一行摘要:前 40 字 + 字数)/ error | `agent_thought` 增量帧聚合 | 默认折叠;点击展开全文(Markdown 渲染);进行中可"实时展开" | +| `ToolCallCard` | calling(转轮)/ succeeded(绿勾+耗时)/ failed(红叹+错误摘要) | `tool_call` 帧聚合(name/input/output/status/duration) | 入参/结果默认摘要(N KB 截断),点击展开 JSON/代码(**需补块级渲染**,见 D.6) | +| `ProcessStepOutput` | done | 既有 `step_output` 整段 content | 默认折叠,展开渲染 Markdown | + +组件均为纯展示组件(props 驱动),放置 `web/src/components/subagents/`,配 `node:test` 静态断言测试(与仓内组件测试形态一致)。 + +### D.4 前端数据流与状态 + +```typescript +// 设计草案:web/src/hooks/useRunProcessStream.ts +type ProcessItem = { + key: string; // `${stepId}/${itemId}` + category: 'agent_thought' | 'tool_call' | 'step_output' | 'log'; + role: string; + stepId: string; + title?: string; + text: string; // delta 累计或 content 快照 + status: 'streaming' | 'done' | 'error'; + tool?: { name: string; status: string; input?: string; output?: string; durationMs?: number }; + updatedAt: number; +}; + +function applyDetailEvent(items: Record, ev: DetailEvent): Record +// 纯函数(与 applyPhaseEvent 同风格,node:test 可测): +// delta → 追加 text、status=streaming +// content → 覆盖 text(快照对账) +// done → status=done +// tool 帧 → 合并 tool 字段 +``` + +订阅策略(复用既有骨架): + +- **live**:`subscribeTaskRunEvents(runId, { onEvent, afterSeq })`([web/src/api/taskRuns.ts:113](../../../web/src/api/taskRuns.ts)),事件分发:`phase` → 左栏状态 + 底部时间线;`detail` → `applyDetailEvent`; +- **刷新/进入已结束 run**:先 `readTaskRunSSEOnce(runId, 0, …, { replayOnly: true })` 全量重放(事件已持久化,见 C.6),再决定是否续订 live; +- **超量回放保护**:`after_seq` 分页重放(既有能力),detail 大 run 可先只重放 phase + 按需拉成员 detail(`GET /runs/{id}/events?role=xxx&include_details=1` 过滤参数为可选增强,P2.5)。 + +### D.5 与主对话气泡的联动 + +- 气泡内 `TaskRunCard` 保持现状(L1 时间线),新增两处: + 1. 运行中:卡片底部一行"查看小队过程 →"(写 hash 深链进子页面); + 2. 完成后:L3 摘要条(如"5 名专员 · 12 次工具调用 · 3 段推演"),点击同样进子页面; +- **不把全量 detail 渲染进气泡**——认知负荷原则:气泡只承载 L1/L2,L3 全文在子页面。`MessageBubble` 现有 `message.phaseEvents` 通道不动。 + +### D.6 前端基础设施缺口补齐(实施前置) + +1. **块级 Markdown + 代码高亮**:现状 `renderInlineMarkdown` 仅行内语法([web/src/engine/renderMarkdown.ts](../../../web/src/engine/renderMarkdown.ts))。工具调用入参/结果需要 fenced code block + JSON 高亮。方案:`renderMarkdown` 增块级分支(heading/fence/list——解析器自研或引入 `marked`/`react-markdown`;倾向自研轻量扩展,避免新依赖与 XSS 面扩大,沿用现有 escape 纪律); +2. **高频增量渲染性能**:processItems 更新频率受后端节流(100–300ms 批帧)保护;子页面右栏**不用虚拟列表**(成员过程项数量级 ~10²),但聚合函数必须纯函数 + `React.memo` 行项; +3. **通用折叠组件**:自研 `CollapsibleBlock`(受控 `open` + `Summary` 插槽),三组件复用。 + +### D.7 后端配套只读端点(成员编制) + +子页面左栏需要 `squad.json`。设计(P2 内最小): + +``` +GET /api/v1/tasks/runs/{run_id}/squad +→ { members: [{ role, name, responsibilities, source }], task_key, title } +``` + +实现:taskrun Manager 读 `record.WorkspaceID` → 工作区 `squad.json` 反序列化(只读投影,权限沿用 run 事件的 `RequireProjectAccess`)。 + +--- + +## E. 分期实施路线 + +| 期 | 内容 | 验证形态(按验证契约) | 依赖 | +|---|---|---|---| +| **P0 协议+直通** | B.2/B.3 协议扩展;C.1–C.4(pi-native 全链路);C.6 includeDetails 参数化 + detail 持久化;前端 chatStream 类型 + detail 分支(仅聚合不渲染) | Go 单测(llm sse_test 增 reasoning case、squad bridge 节流测试);stub/pi-native E2E;真实 PI Runner(dev 环境)冒烟 | 无 | +| **P1 气泡过程块** | D.5 气泡内 L3 摘要 + 折叠 ThinkingBlock(最小渲染,块级 markdown 只做 fence+JSON) | 前端 node:test + 真实浏览器验收(交互变更) | P0 | +| **P2 subagent 子页面** | D.1–D.7 全量(hash 路由、成员卡、过程流、squad 端点) | 前端单测 + Playwright spec(新增 e2e/subagents-view.spec.ts)+ 真实 PI 全链路验收 | P1 | +| **P2.5(可选)** | 事件过滤参数(role/include_details query);大 run 分页回放 | 接口测试 | P2 | +| **P3 runner 流式** | C.5 `/v1/run/stream` + piagent 流式消费 + 404 回落 | runner 单测(mjs 侧 settle-policy 风格)+ 真实容器验收(dev) | P0(可并行) | +| **P4 B 轨参谋** | C.4 尾段:dialogue_llm OnEvent → token 级 EmitDelta + reasoning detail | 真实 PI 对话链路验收(chat SSE) | P0 | +| **P5(演进)** | 历史回放端点(transcript.jsonl/execution.log 投影 API);per-member 持久化表(如 `task_run_step_events`,仅当 JSONB 容量实测超限) | 视实测 | P2/P3 | + +每期独立合入(Conventional Commits:`feat(stream)/feat(web)` 前后端分原子提交),P0 即可演示 pi-native 路径的"思考流+工具卡"。 + +--- + +## F. 风险与边界 + +### F.1 事件量与持久化膨胀 + +- **节流义务在 bridge**(C.4):reasoning/text delta 合并 100–300ms 批帧;单帧 ≤8KB;工具输出截断(如 4KB 首部 + 省略标记);`step_output` 保持现状单帧(终态产物已有 `deliverable` 通道与文档存储,不重复全文持久化时可评估只存摘要——P0 先全文,P2.5 实测再收紧); +- **容量估算**:单 run 5 成员 × (1 思考块 2–8KB + 6 工具调用 × 1KB 摘要) ≈ 50KB/run 量级(节流后),`task_run_events` JSONB 单行无硬限,可控; +- **兜底**:`execution.log`/`transcript.jsonl` 仍全量落盘工作区(现状不变),P5 投影 API 作历史回放。 + +### F.2 认知负荷与安全 + +- detail 默认折叠贯穿三处(气泡摘要、子页面 ThinkingBlock、ToolCallCard 入参/结果摘要)——继承 2026-09-09 设计 P1 约束; +- **脱敏**:bridge 层过滤——不透传 prompt/AGENTS.md 全文(这些是任务输入不是过程);工具入参只发 workspace 相对路径相关字段;配置/密钥字段(provider/key)永不进事件(`ProcessEvent` 设计上就不含这些来源); +- **治理红线**:过程事件不写 `events` 域总线、不参与验收判定;测试/审查结论仍以报告路径与 receipt 为准(pi-subagents 规则),本设计纯可视化。 + +### F.3 兼容与降级 + +- 旧 runner(无 `/v1/run/stream`)→ 404 回落一次性调用,功能退化为 L1 时间线(现状); +- `OnEvent == nil` → 所有后端行为与现状完全一致(回调是纯增量); +- 前端不识别 detail(旧前端 + 新后端)→ JSON 字段忽略,无破坏; +- `includeDetails` 默认 false → chat 通道现状不变。 + +### F.4 并行正确性 + +- 并行组多成员同时发 detail:`Streamer.Emit` 已有 RWMutex;bridge 无共享状态(每 step 独立闭包); +- 前端聚合键含 role/stepID(B.4),并行不串扰; +- `step_output` 终态帧在并行完成回调内发出(现状 workflow.go:746 位置不变)。 + +### F.5 明确不做 + +- 不引入 WebSocket(SSE + after_seq 重放已覆盖需求); +- 不做过程事件的域总线登记(B.5 理由); +- 不做 B 轨 chat 过程帧的持久化与断线重放(P4 live-only,终态落库现状); +- 本期不实现任何代码(用户授权边界)。 + +--- + +## G. 附录:断点-文件-方案对照表 + +| # | 断点 | 文件:行 | 方案章节 | +|---|---|---|---| +| 1 | reasoning_content 未解析 | internal/agentx/llm/sse.go、types.go | C.1 / B.3 | +| 2 | agentx 丢弃 tool_call_delta | internal/agentx/loop.go:178-183 | C.2 | +| 3 | Backend 黑盒无回调 | internal/agent/backend.go:38、pinative.go:102 | C.3 | +| 4 | squad 层无成员过程流 | internal/squad/workflow.go:543(runOneStep) | C.4 | +| 5 | runner /v1/run 一次性 | deploy/pi-agent/runner.mjs:600 | C.5 | +| 6 | includeDetails 硬编码 false / 丢弃 | internal/taskrun/manager.go:364、server/sse.go:476 | C.6 | +| 7 | 前端无 detail 协议与分支 | web/src/api/chatStream.ts、hooks/useTaskRunAttempts.ts:474 | C.7 | + +## H. 参考文件 + +- 三层信息架构基座:`.doc/archive/design/superpowers/specs/2026-09-08-streaming-events-and-digestible-ux-design.md` +- 认知负荷约束:`.doc/archive/design/superpowers/specs/2026-09-09-chat-wait-error-retry-ux-design.md` +- 小队编排(含 §18.5 演进方向):`.doc/system-design/agent-squad/02-智能体小队编排-设计-思路-架构-实现-难点-解决方案-亮点与优势.md` +- 长任务 resume:`.doc/archive/design/superpowers/specs/2026-09-09-longtask-phase-live-resume-ux-design.md` +- 任务记录:`.doc/ops/short-term/2026-09-28-subagent-process-view-design.md` diff --git a/.doc/system-design/README.md b/.doc/system-design/README.md index b6a2f936..70c64361 100644 --- a/.doc/system-design/README.md +++ b/.doc/system-design/README.md @@ -7,6 +7,14 @@ ## 📋 报告清单 +### 0c. 小队过程流式展示与 subagent 子页面(2026-09-28) + +| 项 | 链接 | +|-----|------| +| 主设计 | [2026-09-28-agent-process-streaming-and-subagent-view-design.md](2026-09-28-agent-process-streaming-and-subagent-view-design.md) | +| 任务 | [../ops/short-term/2026-09-28-subagent-process-view-design.md](../ops/short-term/2026-09-28-subagent-process-view-design.md) | +| 一句话总结 | 思考/工具调用/过程内容七断点打通(reasoning 解析→Backend 回调→runner 流式→detail 持久化)+ subagent 子页面(hash 路由 + 成员卡 + 过程流组件);设计稿未实施 | + ### 0b. 自由协作与认领提醒(2026-09-24) | 项 | 链接 | diff --git a/deploy/pi-agent/runner.mjs b/deploy/pi-agent/runner.mjs index e30bcb73..e60aa4db 100755 --- a/deploy/pi-agent/runner.mjs +++ b/deploy/pi-agent/runner.mjs @@ -51,6 +51,33 @@ function extractAssistantText(ev) { return chunks.join(""); } +// toProcFrame 把 pi RPC stdout 事件行投影为 Go 侧 agent.ProcessEvent 同构帧 +// (2026-09-28 过程流式设计 C.5)。只保留过程可视化需要的类型,其他类型原样 +// 透传 type 供前端扩展。think 块在 stripThinkBlocks 剥离前分流为 reasoning_delta。 +function toProcFrame(ev) { + const t = ev.type || ""; + const msg = ev.message && typeof ev.message === "object" ? ev.message : null; + if (t === "message_end" || t === "turn_end") { + return { type: "message_end" }; + } + if (msg && msg.role === "assistant") { + const parts = Array.isArray(msg.content) ? msg.content : []; + for (const part of parts) { + if (part && typeof part === "object" && part.type === "reasoning" && typeof part.text === "string") { + return { type: "reasoning_delta", text: part.text }; + } + } + } + if (t === "tool_execution_start" || t === "tool_execution_end") { + return { + type: t, + tool: typeof ev.tool === "string" ? ev.tool : (msg && typeof msg.name === "string" ? msg.name : ""), + detail: typeof ev.detail === "string" ? ev.detail : "", + }; + } + return { type: t }; +} + function materializeReasoningTargets({ workspace, targets, @@ -254,7 +281,7 @@ function authorizeRunner(req, res) { return false; } -async function handleRun(body) { +async function handleRun(body, onProcEvent) { const rawWs = body.workspace || ""; const workspace = resolveWorkspace(rawWs); const model = body.model || runtimeModel || DEFAULT_MODEL; @@ -382,6 +409,11 @@ async function handleRun(body) { lastAssistantText = assistantChunk; } applyAgentEvent(settle, ev); + if (onProcEvent) { + try { onProcEvent(toProcFrame(ev)); } catch (e) { + console.warn("[runner] proc-event forward failed:", e.message); + } + } if (t === "agent_settled") { child.stdin.end(); } @@ -596,6 +628,35 @@ const server = http.createServer(async (req, res) => { return; } + // 4a. Streaming Agent Task Run Endpoint (NDJSON chunked, 2026-09-28 process streaming design C.5) + if (method === "POST" && url.pathname === "/v1/run/stream") { + if (!authorizeRunner(req, res)) { + return; + } + res.writeHead(200, { + "Content-Type": "application/x-ndjson", + "Cache-Control": "no-cache", + "X-Accel-Buffering": "no", + }); + const writeFrame = (obj) => { + try { + res.write(JSON.stringify(obj) + "\n"); + } catch { + // client disconnect: keep running so the run still settles; write errors ignored + } + }; + const onProcEvent = (frame) => writeFrame({ type: "proc", ev: frame }); + try { + const body = await parseJsonBody(req); + const result = await handleRun(body, onProcEvent); + writeFrame({ type: "result", body: withResultV1(result) }); + } catch (err) { + writeFrame({ type: "result", body: withResultV1({ ok: false, error: "runner_failed", patches: [], run_id: "unknown" }) }); + } + res.end(); + return; + } + // 4. Agent Task Run Endpoint if (method === "POST" && url.pathname === "/v1/run") { if (!authorizeRunner(req, res)) { diff --git a/internal/agent/backend.go b/internal/agent/backend.go index 994cbe93..7e9de90f 100644 --- a/internal/agent/backend.go +++ b/internal/agent/backend.go @@ -15,6 +15,19 @@ type Request struct { Timeout time.Duration `json:"timeout"` // ToolProfile is workspace (default, document tools) or reasoning (no tools). ToolProfile string `json:"tool_profile,omitempty"` + // OnEvent 可选过程事件回调(nil = 调用方不需要过程,后端行为与现状完全一致)。 + // 实现方必须保证:回调在 Execute 返回前完成(同步调用或等待收尾),无泄漏 goroutine; + // 回调内不得阻塞(消费方自行节流/异步入队)。 + OnEvent func(ev ProcessEvent) +} + +// ProcessEvent 对 agentx 事件与远程 runner NDJSON 过程帧的统一投影, +// 供小队过程流式展示消费(见 .doc/system-design/2026-09-28-agent-process-streaming-and-subagent-view-design.md)。 +type ProcessEvent struct { + Type string // text_delta | reasoning_delta | tool_call_delta | tool_execution_start | tool_execution_end | message_end | ... + Text string // delta 文本(Type 为 *_delta 时) + Tool string // 工具名(tool_* 时) + Detail string // 工具参数/结果摘要(tool_execution_* 时) } const ( diff --git a/internal/agent/piagent.go b/internal/agent/piagent.go index e3d6c20d..e4a8705c 100644 --- a/internal/agent/piagent.go +++ b/internal/agent/piagent.go @@ -1,6 +1,7 @@ package agent import ( + "bufio" "bytes" "context" "encoding/json" @@ -217,6 +218,14 @@ func (p *PiAgentBackend) executeRemote(ctx context.Context, runnerURL string, re } reqBody, _ := json.Marshal(p.runnerPayload(req, workDir, true, timeout)) + // 过程流式路径(2026-09-28 设计 C.5):调用方带 OnEvent 时优先走 NDJSON 流式 + // 端点,实时转发 proc 帧;旧 runner 404 时回落一次性 /v1/run(无过程事件)。 + if req.OnEvent != nil { + if res, attempted, err := p.executeRemoteStream(ctx, runnerURL, reqBody, req, expectedRunID, timeout); attempted { + return res, err + } + } + httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, runnerURL+"/v1/run", bytes.NewReader(reqBody)) if err != nil { return &Result{ @@ -244,8 +253,85 @@ func (p *PiAgentBackend) executeRemote(ctx context.Context, runnerURL string, re if len(limited) > MaxResultBodyBytes { return &Result{Success: false, ErrorReason: "runner result body too large"}, nil } + return decodeRemoteResult(limited, expectedRunID, workDir, req) +} - decoded, decErr := DecodeRemoteRun(limited, expectedRunID, true) +// executeRemoteStream 调用 runner /v1/run/stream:proc 帧实时转发 OnEvent, +// result 终帧走与一次性路径完全相同的 DecodeRemoteRun 校验。 +// attempted=false 表示端点不可用(404/405),调用方应回落 /v1/run。 +func (p *PiAgentBackend) executeRemoteStream(ctx context.Context, runnerURL string, reqBody []byte, req Request, expectedRunID string, timeout time.Duration) (*Result, bool, error) { + httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, runnerURL+"/v1/run/stream", bytes.NewReader(reqBody)) + if err != nil { + return nil, false, nil + } + httpReq.Header.Set("Content-Type", "application/json") + httpReq.Header.Set("Accept", "application/x-ndjson") + ApplyRunnerAuth(httpReq, p.cfg.PiAgentRunnerAuthToken) + + // 无总超时的 client:超时由 ctx 控制,逐行流式读取期间保持连接。 + client := &http.Client{} + resp, err := client.Do(httpReq) + if err != nil { + return nil, false, nil // 网络错误也回落一次性路径 + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + // 任何非 200(含旧 runner 404 与瞬态 5xx)都回落一次性路径, + // 保证错误分类与既有 /v1/run 语义完全一致(不改变治理边界)。 + resp.Body.Close() + return nil, false, nil + } + + var resultBody []byte + var sawResult bool + sc := bufio.NewScanner(resp.Body) + sc.Buffer(make([]byte, 0, 64*1024), MaxResultBodyBytes) + for sc.Scan() { + if ctx.Err() != nil { + return &Result{Success: false, ErrorReason: "stream context cancelled", Retryable: true}, true, nil + } + line := strings.TrimSpace(sc.Text()) + if line == "" { + continue + } + var frame struct { + Type string `json:"type"` + Ev *ProcessEvent `json:"ev,omitempty"` + Body json.RawMessage `json:"body,omitempty"` + } + if err := json.Unmarshal([]byte(line), &frame); err != nil { + continue + } + switch frame.Type { + case "proc": + if frame.Ev != nil && req.OnEvent != nil { + req.OnEvent(*frame.Ev) + } + case "result": + resultBody = frame.Body + sawResult = true + } + if sawResult { + break // result 是终帧 + } + } + if !sawResult { + return &Result{ + Success: false, + ErrorReason: "pi runner stream ended without result frame", + Retryable: true, + }, true, nil + } + res, decErr := decodeRemoteResult(resultBody, expectedRunID, req.WorkDir, req) + if decErr != nil { + return res, true, decErr + } + return res, true, nil +} + +// decodeRemoteResult 一次性与流式路径共用的终态解码与落盘。 +func decodeRemoteResult(body []byte, expectedRunID, workDir string, req Request) (*Result, error) { + decoded, decErr := DecodeRemoteRun(body, expectedRunID, true) if decErr != nil { reason := "typed result rejected" msg := decErr.Error() diff --git a/internal/agent/pinative.go b/internal/agent/pinative.go index ed18e197..831d296c 100644 --- a/internal/agent/pinative.go +++ b/internal/agent/pinative.go @@ -93,6 +93,7 @@ func (p *PiNativeBackend) Execute(ctx context.Context, req Request) (*Result, er defer cancel() var logs strings.Builder + onProcess := req.OnEvent runRes, err := agentx.Run(runCtx, agentx.Config{ Model: model, SystemPrompt: nativeSystemPrompt(req, profile), @@ -110,6 +111,14 @@ func (p *PiNativeBackend) Execute(ctx context.Context, req Request) (*Result, er logs.WriteString(ev.Text) } logs.WriteByte('\n') + if onProcess != nil { + onProcess(ProcessEvent{ + Type: ev.Type, + Text: ev.Text, + Tool: ev.Tool, + Detail: ev.Detail, + }) + } }, }, llm.Message{Role: llm.RoleUser, Content: req.Prompt}) if err != nil { diff --git a/internal/agentx/llm/sse.go b/internal/agentx/llm/sse.go index 258f8261..63899178 100644 --- a/internal/agentx/llm/sse.go +++ b/internal/agentx/llm/sse.go @@ -11,6 +11,7 @@ import ( type sseChoice struct { Delta struct { Content string `json:"content"` + Reasoning string `json:"reasoning_content"` ToolCalls []struct { Index int `json:"index"` ID string `json:"id"` @@ -94,6 +95,17 @@ func emitParsed(ctx context.Context, r io.Reader, out chan<- Event) { case out <- Event{Kind: KindTextDelta, Text: choice.Delta.Content}: } } + // reasoning_content(国产 OpenAI 兼容模型的思考链)与正文分流: + // 只作为 reasoning_delta 事件外发,不计入 Message.Content(与 runner + // stripThinkBlocks"正文无 think"语义对齐)。 + if choice.Delta.Reasoning != "" { + select { + case <-ctx.Done(): + emitAbort(out, content.String(), ctx.Err()) + return + case out <- Event{Kind: KindReasoningDelta, ReasoningDelta: choice.Delta.Reasoning}: + } + } for _, tc := range choice.Delta.ToolCalls { cur, ok := calls[tc.Index] if !ok { diff --git a/internal/agentx/llm/sse_test.go b/internal/agentx/llm/sse_test.go index a1ddf447..56c04558 100644 --- a/internal/agentx/llm/sse_test.go +++ b/internal/agentx/llm/sse_test.go @@ -189,3 +189,36 @@ func TestStreamErrorString(t *testing.T) { t.Fatal("error string") } } + +func TestParseSSEReasoningDelta(t *testing.T) { + body := strings.Join([]string{ + `data: {"choices":[{"delta":{"reasoning_content":"先分析需求"},"index":0}]}`, + `data: {"choices":[{"delta":{"reasoning_content":"再检索标准"},"index":0}]}`, + `data: {"choices":[{"delta":{"content":"结论"},"finish_reason":"stop"}],"usage":{"prompt_tokens":1,"completion_tokens":2,"total_tokens":3}}`, + `data: [DONE]`, + "", + }, "\n") + var reasoning []string + var text []string + var end Event + for ev := range ParseSSE(context.Background(), strings.NewReader(body)) { + switch ev.Kind { + case KindReasoningDelta: + reasoning = append(reasoning, ev.ReasoningDelta) + case KindTextDelta: + text = append(text, ev.Text) + case KindMessageEnd: + end = ev + } + } + if strings.Join(reasoning, "") != "先分析需求再检索标准" { + t.Fatalf("reasoning deltas %v", reasoning) + } + // 正文与思考链分流:Content 只含 text,不含 reasoning + if strings.Join(text, "") != "结论" { + t.Fatalf("text deltas %v", text) + } + if end.Message.Content != "结论" { + t.Fatalf("end content %q must not include reasoning", end.Message.Content) + } +} diff --git a/internal/agentx/llm/types.go b/internal/agentx/llm/types.go index bf19df2f..7cae6036 100644 --- a/internal/agentx/llm/types.go +++ b/internal/agentx/llm/types.go @@ -29,10 +29,11 @@ const ( type EventKind string const ( - KindTextDelta EventKind = "text_delta" - KindToolCallDelta EventKind = "tool_call_delta" - KindMessageEnd EventKind = "message_end" - KindError EventKind = "error" + KindTextDelta EventKind = "text_delta" + KindReasoningDelta EventKind = "reasoning_delta" + KindToolCallDelta EventKind = "tool_call_delta" + KindMessageEnd EventKind = "message_end" + KindError EventKind = "error" ) // ToolCall is one function call requested by the assistant. @@ -72,6 +73,7 @@ type Usage struct { type Event struct { Kind EventKind Text string + ReasoningDelta string ToolIndex int ToolID string ToolName string diff --git a/internal/agentx/loop.go b/internal/agentx/loop.go index 6841a3dc..fff26aeb 100644 --- a/internal/agentx/loop.go +++ b/internal/agentx/loop.go @@ -22,6 +22,9 @@ type Event struct { Message llm.Message Tool string Text string + // Detail carries tool arguments/results or reasoning deltas for + // process-visualization consumers; log writers may ignore it. + Detail string } // Config drives one agent session. @@ -178,6 +181,10 @@ func consume(ctx context.Context, cfg Config, messages []llm.Message) (llm.Event switch ev.Kind { case llm.KindTextDelta: emit(cfg, Event{Type: "text_delta", Text: ev.Text}) + case llm.KindReasoningDelta: + emit(cfg, Event{Type: "reasoning_delta", Text: ev.ReasoningDelta}) + case llm.KindToolCallDelta: + emit(cfg, Event{Type: "tool_call_delta", Tool: ev.ToolName, Detail: ev.ArgumentsDelta}) case llm.KindError: return ev, ev.Err case llm.KindMessageEnd: diff --git a/internal/server/server.go b/internal/server/server.go index b95426e6..1c523b9c 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -233,6 +233,7 @@ func (s *Server) Routes() http.Handler { mux.HandleFunc("GET /api/v1/tasks/runs", s.handleListTaskRuns) mux.HandleFunc("GET /api/v1/tasks/runs/{run_id}", s.handleGetTaskRun) mux.HandleFunc("GET /api/v1/tasks/runs/{run_id}/events", s.handleTaskRunEvents) + mux.HandleFunc("GET /api/v1/tasks/runs/{run_id}/squad", s.handleGetTaskRunSquad) mux.HandleFunc("POST /api/v1/tasks/runs/{run_id}/cancel", s.handleCancelTaskRun) mux.HandleFunc("POST /api/v1/tasks/runs/{run_id}/resume", s.handleResumeTaskRun) mux.HandleFunc("POST /api/v1/admin/"+workspace.WorkspacesDirName+"/reconcile", s.handleReconcileWorkspace) diff --git a/internal/server/sse.go b/internal/server/sse.go index 39b632be..5fa4c542 100644 --- a/internal/server/sse.go +++ b/internal/server/sse.go @@ -473,6 +473,9 @@ func (s *Server) handleRunTaskStream(w http.ResponseWriter, r *http.Request) { return } } + // includeDetails 参数保留解析兼容,但 worker 侧已一律产出 detail 事件 + // (2026-09-28 过程流式设计 C.6:detail 全量持久化与推送,认知负荷由前端 + // 默认折叠保证;该参数不再控制任务通道的事件产出)。 _ = includeDetails _ = requestID diff --git a/internal/server/task_run_handlers.go b/internal/server/task_run_handlers.go index 021986e9..b6fbd557 100644 --- a/internal/server/task_run_handlers.go +++ b/internal/server/task_run_handlers.go @@ -672,6 +672,48 @@ func (s *Server) handleGetTaskRun(w http.ResponseWriter, r *http.Request) { writeJSONResponse(w, http.StatusOK, s.buildTaskRunResponse(rec)) } +// handleGetTaskRunSquad 只读投影工作区 squad.json,供小队过程子页面左栏。 +func (s *Server) handleGetTaskRunSquad(w http.ResponseWriter, r *http.Request) { + user, ok := s.requirePrincipal(w, r) + if !ok { + return + } + runID := r.PathValue("run_id") + if runID == "" { + writeSafeError(w, r, http.StatusBadRequest, "run_id required") + return + } + if !s.assertTaskRunOwnedByUser(w, r.Context(), runID, user) { + return + } + rec, err := s.taskRuns().GetRun(runID) + if err != nil { + writeDomainError(w, r, http.StatusNotFound, err) + return + } + if rec == nil || strings.TrimSpace(rec.WorkspaceID) == "" || s.resolver() == nil { + writeSafeError(w, r, http.StatusNotFound, "workspace not found") + return + } + dir, err := s.resolver().ResolveStoredWorkDir(rec.WorkspaceID) + if err != nil { + writeSafeError(w, r, http.StatusNotFound, "workspace not found") + return + } + proj, err := squad.ReadSquadProjection(dir) + if err != nil { + writeSafeError(w, r, http.StatusNotFound, "squad not found") + return + } + if proj.TaskKey == "" { + proj.TaskKey = rec.TaskKey + } + if proj.Title == "" { + proj.Title = rec.TaskKey + } + writeJSONResponse(w, http.StatusOK, proj) +} + func (s *Server) handleResumeTaskRun(w http.ResponseWriter, r *http.Request) { user, ok := s.requirePrincipal(w, r) if !ok { diff --git a/internal/squad/process_bridge.go b/internal/squad/process_bridge.go new file mode 100644 index 00000000..9945b78c --- /dev/null +++ b/internal/squad/process_bridge.go @@ -0,0 +1,406 @@ +package squad + +import ( + "fmt" + "strings" + "sync" + "time" + + "chat-agent-system/internal/agent" + "chat-agent-system/internal/stream" +) + +// 过程流常量:节流窗口 / 单帧上限 / 工具文本截断(设计文档 F.1)。 +const ( + processFlushInterval = 200 * time.Millisecond + processMaxFrameBytes = 8 * 1024 + processToolTrimBytes = 4 * 1024 +) + +// detailCategory 过程事件类别(与 stream.DetailPayload.Category 枚举对齐)。 +const ( + detailCategoryAgentThought = "agent_thought" + detailCategoryToolCall = "tool_call" +) + +// deltaStream 一个聚合项的增量缓冲(节流冲刷)。 +// 只被 bridge 主路径使用;timer 回调仅触发 flush 任务。 +type deltaStream struct { + mu sync.Mutex + itemID string + buf strings.Builder + timer *time.Timer + done bool +} + +func (d *deltaStream) append(text string) { + d.mu.Lock() + defer d.mu.Unlock() + if d.done || text == "" { + return + } + d.buf.WriteString(text) +} + +// takeIfReady 返回并清空缓冲;由外部节拍调用(非定时器自转)。 +func (d *deltaStream) take() string { + d.mu.Lock() + defer d.mu.Unlock() + out := d.buf.String() + d.buf.Reset() + return out +} + +func (d *deltaStream) finish() { + d.mu.Lock() + defer d.mu.Unlock() + if d.timer != nil { + d.timer.Stop() + d.timer = nil + } + d.done = true +} + +// processBridge 把 agent.ProcessEvent 翻译为 stream.EventDetail 增量帧。 +// 每个 step(小队成员的一次执行)独立持有一个 bridge;无跨 step 共享状态,并行安全。 +type processBridge struct { + streamer *stream.Streamer + step StepSpec + friendly string // 角色显示名 + flushTick *time.Ticker + stopCh chan struct{} + stopOnce sync.Once + wg sync.WaitGroup + + mu sync.Mutex + thoughtSeq int + toolSeq int + thought *thoughtBlock // 当前思考块(nil = 尚未开始) + currentTool *toolBlock + emitted bool // 是否已发出过事件(Stop 只在发过时才有意义) +} + +// thoughtBlock 一个 message 周期内的思考聚合项。 +type thoughtBlock struct { + itemID string + stream *deltaStream + title string +} + +// toolBlock 一次工具调用的聚合项。 +type toolBlock struct { + itemID string + name string + started time.Time + input strings.Builder +} + +// newProcessBridge 创建过程桥。返回 nil 表示该 step 无流式需求(streamer 缺失), +// 调用方对 nil bridge 调用任何方法都是安全 no-op。 +func newProcessBridge(streamer *stream.Streamer, step StepSpec, friendlyRole string) *processBridge { + if streamer == nil { + return nil + } + b := &processBridge{ + streamer: streamer, + step: step, + friendly: friendlyRole, + stopCh: make(chan struct{}), + } + // 定时冲刷循环:所有 delta 缓冲统一按 flushInterval 节拍冲刷。 + b.flushTick = time.NewTicker(processFlushInterval) + b.wg.Add(1) + go func() { + defer b.wg.Done() + for { + select { + case <-b.stopCh: + return + case <-b.flushTick.C: + b.flushPending() + } + } + }() + return b +} + +// Stop 结束 bridge:冲刷残余、定稿当前项、停止节拍循环。幂等(可安全多次调用)。 +func (b *processBridge) Stop() { + if b == nil { + return + } + b.stopOnce.Do(func() { + b.flushPending() + b.mu.Lock() + th := b.thought + tl := b.currentTool + b.mu.Unlock() + if th != nil { + b.finishThought(th) + } + if tl != nil { + b.finishTool(tl, "succeeded", "") + } + close(b.stopCh) + b.flushTick.Stop() + b.wg.Wait() + }) +} + +// Handle 把一个 agent.ProcessEvent 翻译为 detail 事件流。 +// 调用方保证事件到达顺序与 agentx 循环一致(同 step 内串行)。 +func (b *processBridge) Handle(ev agent.ProcessEvent) { + if b == nil { + return + } + select { + case <-b.stopCh: + return // Stop 后忽略(防御:Execute 返回后的残余事件) + default: + } + switch ev.Type { + case "reasoning_delta": + b.onReasoningDelta(ev.Text) + case "text_delta": + // P0 决策(设计文档 C.4):任务正文不发 detail——终态走既有 step_output + // 通道,避免与最终 deliverable 重复推送大文本。 + case "tool_call_delta": + b.onToolCallDelta(ev) + case "tool_execution_start": + b.onToolStart(ev.Tool) + case "tool_execution_end": + b.onToolEnd(ev.Tool, ev.Detail) + case "message_end": + b.onMessageEnd() + } +} + +func (b *processBridge) onReasoningDelta(text string) { + b.mu.Lock() + if b.thought == nil { + b.thoughtSeq++ + b.thought = &thoughtBlock{ + itemID: fmt.Sprintf("thought#%s#%d", b.step.ID, b.thoughtSeq), + stream: &deltaStream{}, + title: fmt.Sprintf("%s思考过程 %d", b.friendly, b.thoughtSeq), + } + b.emitted = true + } + th := b.thought + b.mu.Unlock() + th.stream.append(text) +} + +func (b *processBridge) onToolCallDelta(ev agent.ProcessEvent) { + b.mu.Lock() + t := b.currentTool + if t == nil { + // 模型在 tool_execution_start 前就吐参数增量:只预建内部块(不发 calling 帧, + // 正式的 calling 帧由随后的 tool_execution_start 发出)。 + b.toolSeq++ + t = &toolBlock{ + itemID: fmt.Sprintf("tool#%s#%d", b.step.ID, b.toolSeq), + name: ev.Tool, + started: time.Now(), + } + b.currentTool = t + b.emitted = true + } + b.mu.Unlock() + t.input.WriteString(ev.Detail) +} + +func (b *processBridge) onToolStart(name string) { + b.mu.Lock() + var cur *toolBlock + if b.currentTool != nil { + // 已由 tool_call_delta 预建(罕见时序):复用同一聚合项,只补名与时间。 + cur = b.currentTool + if cur.name == "" { + cur.name = name + } + cur.started = time.Now() + } else { + b.toolSeq++ + cur = &toolBlock{ + itemID: fmt.Sprintf("tool#%s#%d", b.step.ID, b.toolSeq), + name: name, + started: time.Now(), + } + b.currentTool = cur + } + b.emitted = true + itemID, toolName := cur.itemID, cur.name + b.mu.Unlock() + // tool_execution_start 是正式的"开始调用"信号:无论是否预建,都要发 calling 帧。 + b.emitDetail(&stream.DetailPayload{ + Category: detailCategoryToolCall, + StepID: b.step.ID, + Role: b.step.Role, + ItemID: itemID, + Tool: &stream.ToolCallInfo{Name: toolName, Status: "calling"}, + Title: fmt.Sprintf("%s调用工具 %s", b.friendly, toolName), + Collapsible: true, + }) +} + +func (b *processBridge) onToolEnd(name, result string) { + b.mu.Lock() + t := b.currentTool + b.currentTool = nil + b.mu.Unlock() + if t == nil { + return + } + status := "succeeded" + if strings.Contains(strings.ToLower(result), "error") || strings.Contains(strings.ToLower(result), "failed") { + status = "failed" + } + duration := time.Since(t.started).Milliseconds() + info := &stream.ToolCallInfo{ + Name: name, + Status: status, + Input: trimProcessText(t.input.String()), + Output: trimProcessText(result), + DurationMs: duration, + } + b.emitDetail(&stream.DetailPayload{ + Category: detailCategoryToolCall, + StepID: b.step.ID, + Role: b.step.Role, + ItemID: t.itemID, + Tool: info, + Collapsible: true, + }) + // 工具项即时定稿(状态帧 + done 标记合并语义:终帧即最后状态) + b.emitDetail(&stream.DetailPayload{ + Category: detailCategoryToolCall, + StepID: b.step.ID, + Role: b.step.Role, + ItemID: t.itemID, + Tool: info, + Done: true, + Collapsible: true, + }) +} + +func (b *processBridge) onMessageEnd() { + b.mu.Lock() + th := b.thought + b.thought = nil + b.mu.Unlock() + if th != nil { + b.finishThought(th) + } +} + +// flushPending 冲刷当前思考块缓冲(节拍或 Stop 调用)。 +func (b *processBridge) flushPending() { + if b == nil { + return + } + b.mu.Lock() + th := b.thought + b.mu.Unlock() + if th == nil { + return + } + text := th.stream.take() + if text == "" { + return + } + if len(text) > processMaxFrameBytes { + text = text[:processMaxFrameBytes] + "\n…(截断)" + } + b.emitDetail(&stream.DetailPayload{ + Category: detailCategoryAgentThought, + StepID: b.step.ID, + Role: b.step.Role, + ItemID: th.itemID, + Delta: text, + Collapsible: true, + }) +} + +func (b *processBridge) finishThought(th *thoughtBlock) { + th.stream.finish() + // 冲残余 + if text := th.stream.take(); text != "" { + b.emitDetail(&stream.DetailPayload{ + Category: detailCategoryAgentThought, + StepID: b.step.ID, + Role: b.step.Role, + ItemID: th.itemID, + Delta: text, + Collapsible: true, + }) + } + b.emitDetail(&stream.DetailPayload{ + Category: detailCategoryAgentThought, + StepID: b.step.ID, + Role: b.step.Role, + ItemID: th.itemID, + Title: th.title, + Done: true, + Collapsible: true, + }) +} + +func (b *processBridge) finishTool(t *toolBlock, status, result string) { + b.emitDetail(&stream.DetailPayload{ + Category: detailCategoryToolCall, + StepID: b.step.ID, + Role: b.step.Role, + ItemID: t.itemID, + Tool: &stream.ToolCallInfo{ + Name: t.name, + Status: status, + Input: trimProcessText(t.input.String()), + Output: trimProcessText(result), + DurationMs: time.Since(t.started).Milliseconds(), + }, + Done: true, + Collapsible: true, + }) +} + +func (b *processBridge) emitDetail(p *stream.DetailPayload) { + if b == nil || b.streamer == nil { + return + } + b.streamer.Emit(stream.StreamEvent{Type: stream.EventDetail, Detail: p}) +} + +// trimProcessText 工具入参/结果截断(防超大帧;保留首尾)。 +func trimProcessText(s string) string { + if len(s) <= processToolTrimBytes { + return s + } + headN := processToolTrimBytes / 2 + tailN := processToolTrimBytes / 4 + head := s[:headN] + tail := s[len(s)-tailN:] + return head + "\n…(中间省略 " + fmt.Sprint(len(s)-headN-tailN) + " 字节)…\n" + tail +} + +// AttachProcessBridge 给 agent.Request 装配过程桥(若 streamer 缺失或无流式需求则原样返回)。 +// 返回的 stop 函数在 Execute 返回后必须调用一次(定稿与冲刷残余)。 +func AttachProcessBridge(streamer *stream.Streamer, step StepSpec, req *agent.Request) (stop func()) { + if streamer == nil || req == nil { + return func() {} + } + bridge := newProcessBridge(streamer, step, FriendlyRoleName(step.Role)) + if bridge == nil { + return func() {} + } + prev := req.OnEvent + if prev == nil { + req.OnEvent = bridge.Handle + } else { + req.OnEvent = func(ev agent.ProcessEvent) { + prev(ev) + bridge.Handle(ev) + } + } + return bridge.Stop +} diff --git a/internal/squad/process_bridge_test.go b/internal/squad/process_bridge_test.go new file mode 100644 index 00000000..e0c1fdcf --- /dev/null +++ b/internal/squad/process_bridge_test.go @@ -0,0 +1,215 @@ +package squad + +import ( + "strings" + "sync" + "testing" + "time" + + "chat-agent-system/internal/agent" + "chat-agent-system/internal/stream" +) + +func TestProcessBridgeNilStreamerIsNoop(t *testing.T) { + step := StepSpec{ID: "step_01", Role: "regulatory"} + req := &agent.Request{} + stop := AttachProcessBridge(nil, step, req) + if req.OnEvent != nil { + t.Fatal("nil streamer must not attach OnEvent") + } + req.OnEvent = func(agent.ProcessEvent) {} // 直接赋值不应有 panic + stop() // no-op stop 不 panic +} + +func TestProcessBridgeReasoningFlow(t *testing.T) { + var mu sync.Mutex + var events []stream.StreamEvent + streamer := stream.NewStreamer(true, func(ev stream.StreamEvent) { + mu.Lock() + events = append(events, ev) + mu.Unlock() + }) + step := StepSpec{ID: "step_02", Role: "engineer"} + req := &agent.Request{} + stop := AttachProcessBridge(streamer, step, req) + defer stop() + + req.OnEvent(agent.ProcessEvent{Type: "reasoning_delta", Text: "第一段思考"}) + req.OnEvent(agent.ProcessEvent{Type: "reasoning_delta", Text: "继续推理"}) + req.OnEvent(agent.ProcessEvent{Type: "message_end"}) + stop() + // 再发事件(stop 后)不应 panic 且不应再发帧 + req.OnEvent(agent.ProcessEvent{Type: "reasoning_delta", Text: "ignored"}) + + mu.Lock() + defer mu.Unlock() + var deltas, dones int + var joined strings.Builder + for _, ev := range events { + if ev.Type != stream.EventDetail || ev.Detail == nil { + continue + } + if ev.Detail.Category != detailCategoryAgentThought { + t.Fatalf("unexpected category %s", ev.Detail.Category) + } + if ev.Detail.Delta != "" { + deltas++ + joined.WriteString(ev.Detail.Delta) + } + if ev.Detail.Done { + dones++ + } + if ev.Detail.StepID != step.ID || ev.Detail.Role != step.Role { + t.Fatalf("step identity missing: %+v", ev.Detail) + } + } + if joined.String() != "第一段思考继续推理" { + t.Fatalf("joined delta %q", joined.String()) + } + if dones != 1 { + t.Fatalf("expected exactly 1 done frame, got %d", dones) + } + if deltas < 1 { + t.Fatal("expected at least one delta frame") + } +} + +func TestProcessBridgeToolFlow(t *testing.T) { + var mu sync.Mutex + var events []stream.StreamEvent + streamer := stream.NewStreamer(true, func(ev stream.StreamEvent) { + mu.Lock() + events = append(events, ev) + mu.Unlock() + }) + step := StepSpec{ID: "step_03", Role: "writer"} + b := newProcessBridge(streamer, step, "撰写专员") + if b == nil { + t.Fatal("bridge expected") + } + + b.Handle(agent.ProcessEvent{Type: "tool_call_delta", Tool: "read_file", Detail: `{"path":"a.md"`}) + b.Handle(agent.ProcessEvent{Type: "tool_execution_start", Tool: "read_file"}) + b.Handle(agent.ProcessEvent{Type: "tool_call_delta", Tool: "read_file", Detail: `}`}) + b.Handle(agent.ProcessEvent{Type: "tool_execution_end", Tool: "read_file", Detail: "file content ..."}) + b.Stop() + + mu.Lock() + defer mu.Unlock() + var calling, done, withInput int + for _, ev := range events { + if ev.Type != stream.EventDetail || ev.Detail == nil || ev.Detail.Category != detailCategoryToolCall { + continue + } + if ev.Detail.Tool == nil { + t.Fatal("tool_call frame must carry Tool info") + } + switch { + case ev.Detail.Tool.Status == "calling": + calling++ + case ev.Detail.Done: + done++ + } + if strings.Contains(ev.Detail.Tool.Input, "a.md") { + withInput++ + } + } + if calling != 1 { + t.Fatalf("calling frames %d", calling) + } + if done < 1 { + t.Fatal("expected done frame for tool item") + } + if withInput == 0 { + t.Fatal("tool input aggregate lost") + } +} + +func TestProcessBridgeToolFailureStatus(t *testing.T) { + var mu sync.Mutex + var events []stream.StreamEvent + streamer := stream.NewStreamer(true, func(ev stream.StreamEvent) { + mu.Lock() + events = append(events, ev) + mu.Unlock() + }) + b := newProcessBridge(streamer, StepSpec{ID: "s", Role: "r"}, "测试") + b.Handle(agent.ProcessEvent{Type: "tool_execution_start", Tool: "write_file"}) + b.Handle(agent.ProcessEvent{Type: "tool_execution_end", Tool: "write_file", Detail: "error: disk full"}) + b.Stop() + + mu.Lock() + defer mu.Unlock() + sawFailed := false + for _, ev := range events { + if ev.Detail != nil && ev.Detail.Tool != nil && ev.Detail.Tool.Status == "failed" { + sawFailed = true + } + } + if !sawFailed { + t.Fatal("error result must map to failed status") + } +} + +func TestTrimProcessText(t *testing.T) { + small := strings.Repeat("a", 100) + if got := trimProcessText(small); got != small { + t.Fatal("small text must pass through") + } + big := strings.Repeat("b", processToolTrimBytes*4) + got := trimProcessText(big) + if len(got) >= len(big) { + t.Fatal("big text must be trimmed") + } + if !strings.Contains(got, "…(中间省略") { + t.Fatal("trim marker missing") + } +} + +func TestProcessBridgeConcurrentSteps(t *testing.T) { + // 并行组成员各自持有独立 bridge,事件互不串扰(设计 F.4)。 + var mu sync.Mutex + seen := map[string]string{} // itemID -> role + streamer := stream.NewStreamer(true, func(ev stream.StreamEvent) { + if ev.Type != stream.EventDetail || ev.Detail == nil { + return + } + mu.Lock() + seen[ev.Detail.ItemID] = ev.Detail.Role + mu.Unlock() + }) + b1 := newProcessBridge(streamer, StepSpec{ID: "p1", Role: "regulatory"}, "法规专员") + b2 := newProcessBridge(streamer, StepSpec{ID: "p2", Role: "market"}, "市场专员") + var wg sync.WaitGroup + wg.Add(2) + go func() { + defer wg.Done() + for i := 0; i < 20; i++ { + b1.Handle(agent.ProcessEvent{Type: "reasoning_delta", Text: "r1"}) + } + b1.Stop() + }() + go func() { + defer wg.Done() + for i := 0; i < 20; i++ { + b2.Handle(agent.ProcessEvent{Type: "reasoning_delta", Text: "r2"}) + } + b2.Stop() + }() + wg.Wait() + time.Sleep(50 * time.Millisecond) // 等冲刷 + mu.Lock() + defer mu.Unlock() + for itemID, role := range seen { + if !strings.Contains(itemID, "p1") && !strings.Contains(itemID, "p2") { + t.Fatalf("unexpected item %s", itemID) + } + want := "regulatory" + if strings.Contains(itemID, "p2") { + want = "market" + } + if role != want { + t.Fatalf("item %s mapped to role %s, want %s", itemID, role, want) + } + } +} diff --git a/internal/squad/squad_projection.go b/internal/squad/squad_projection.go new file mode 100644 index 00000000..ccdab454 --- /dev/null +++ b/internal/squad/squad_projection.go @@ -0,0 +1,60 @@ +package squad + +import ( + "encoding/json" + "os" + "path/filepath" +) + +// SquadMemberView 是子页面左栏的只读编制投影(不含模型名与工具权限)。 +type SquadMemberView struct { + Role string `json:"role"` + Name string `json:"name"` + Responsibilities []string `json:"responsibilities"` + Source string `json:"source,omitempty"` +} + +// SquadProjection GET /tasks/runs/{id}/squad 的响应体。 +type SquadProjection struct { + Members []SquadMemberView `json:"members"` + TaskKey string `json:"task_key"` + Title string `json:"title"` +} + +// ReadSquadProjection 读取工作区 squad.json(及可选 workspace.json 标题)。 +// 缺少 squad.json 时返回错误,由调用方决定 404。 +func ReadSquadProjection(workDir string) (SquadProjection, error) { + data, err := os.ReadFile(filepath.Join(workDir, "squad.json")) + if err != nil { + return SquadProjection{}, err + } + var cfg SquadConfig + if err := json.Unmarshal(data, &cfg); err != nil { + return SquadProjection{}, err + } + members := make([]SquadMemberView, 0, len(cfg.Agents)) + for _, agent := range cfg.Agents { + resp := agent.Responsibilities + if resp == nil { + resp = []string{} + } + members = append(members, SquadMemberView{ + Role: agent.Role, + Name: agent.Name, + Responsibilities: resp, + Source: string(agent.Source), + }) + } + title := cfg.SquadName + taskKey := "" + if metaRaw, err := os.ReadFile(filepath.Join(workDir, "workspace.json")); err == nil { + var meta WorkspaceMetadata + if json.Unmarshal(metaRaw, &meta) == nil { + if meta.StageTitle != "" { + title = meta.StageTitle + } + taskKey = meta.TaskKey + } + } + return SquadProjection{Members: members, TaskKey: taskKey, Title: title}, nil +} diff --git a/internal/squad/squad_projection_test.go b/internal/squad/squad_projection_test.go new file mode 100644 index 00000000..57ab1402 --- /dev/null +++ b/internal/squad/squad_projection_test.go @@ -0,0 +1,38 @@ +package squad + +import ( + "os" + "path/filepath" + "testing" +) + +func TestReadSquadProjection(t *testing.T) { + dir := t.TempDir() + squadJSON := `{ + "squad_name": "立项小队", + "version": "1", + "agents": [ + {"role": "regulatory", "name": "法规专员", "source": "standard", "responsibilities": ["筛法规"]} + ] + }` + if err := os.WriteFile(filepath.Join(dir, "squad.json"), []byte(squadJSON), 0o644); err != nil { + t.Fatal(err) + } + ws := `{"task_key":"intake","stage_title":"用户需求收集"}` + if err := os.WriteFile(filepath.Join(dir, "workspace.json"), []byte(ws), 0o644); err != nil { + t.Fatal(err) + } + got, err := ReadSquadProjection(dir) + if err != nil { + t.Fatal(err) + } + if got.TaskKey != "intake" || got.Title != "用户需求收集" { + t.Fatalf("meta: %+v", got) + } + if len(got.Members) != 1 || got.Members[0].Role != "regulatory" || got.Members[0].Name != "法规专员" { + t.Fatalf("members: %+v", got.Members) + } + if _, err := ReadSquadProjection(t.TempDir()); err == nil { + t.Fatal("missing squad.json must error") + } +} diff --git a/internal/squad/workflow.go b/internal/squad/workflow.go index 9889d43d..4e12d6cf 100644 --- a/internal/squad/workflow.go +++ b/internal/squad/workflow.go @@ -551,6 +551,9 @@ func (d *WorkflowDriver) runOneStep( Targets: targets, Timeout: stepTimeout, } + // 过程流桥:把成员执行的思考/工具调用增量翻译为 detail 事件(streamer 缺失时为 no-op)。 + // 注意 retry 循环内多次 Execute 共用一个 bridge 语义(同 step 聚合),Stop 只在首次调用。 + stopProcessBridge := AttachProcessBridge(streamer, step, &req) var ( res *agent.Result @@ -561,6 +564,7 @@ func (d *WorkflowDriver) runOneStep( lastAssess agent.FailureAssessment basePrompt = string(taskData) ) + defer stopProcessBridge() for attempt := 1; attempt <= agent.MaxTransientStepInvocations; attempt++ { onDisk, _ := os.ReadFile(prevOutPath) wroteThisInvocation := deliverableWroteThisInvocation(prevContent, onDisk) diff --git a/internal/stream/types.go b/internal/stream/types.go index 70f48483..e7b6863f 100644 --- a/internal/stream/types.go +++ b/internal/stream/types.go @@ -69,9 +69,25 @@ type DetailPayload struct { Category string `json:"category"` // agent_thought, step_output, tool_call, log StepID string `json:"step_id,omitempty"` // 步骤标识 (如 tasks/01_regulatory) Role string `json:"role,omitempty"` // 对应角色 - Title string `json:"title"` // 详情标题 - Content string `json:"content"` // 完整推演正文或技术草稿 - Collapsible bool `json:"collapsible"` // 客户端是否应当默认折叠显示 + Title string `json:"title"` + Content string `json:"content"` // 完整推演正文或技术草稿 + Collapsible bool `json:"collapsible"` // 客户端是否应当默认折叠显示 + // ---- 过程流扩展(2026-09-28 subagent 过程展示设计 B.2)---- + ItemID string `json:"item_id,omitempty"` // 聚合项 ID:一个思考块/一次工具调用 + // 增量模式:Delta 追加到 ItemID 项;Content 为快照对账帧;Done 定稿。 + Delta string `json:"delta,omitempty"` + Done bool `json:"done,omitempty"` + // ToolCallInfo 工具调用状态帧(Category == "tool_call" 时)。 + Tool *ToolCallInfo `json:"tool,omitempty"` +} + +// ToolCallInfo 一次工具调用的状态投影(thinking 流式展示用)。 +type ToolCallInfo struct { + Name string `json:"name"` // write_file / read_file / ... + Status string `json:"status"` // calling | succeeded | failed + Input string `json:"input,omitempty"` // 入参摘要(截断) + Output string `json:"output,omitempty"` // 结果摘要(截断) + DurationMs int64 `json:"duration_ms,omitempty"` } // DeliverablePayload A轨正式合规文档交付载荷 diff --git a/internal/taskrun/manager.go b/internal/taskrun/manager.go index 452ba5a5..b2c7b74d 100644 --- a/internal/taskrun/manager.go +++ b/internal/taskrun/manager.go @@ -476,7 +476,10 @@ func (m *Manager) runWorker(ctx context.Context, ar *activeRun, taskReq bizdev.T } m.appendEvent(ar, ev) } - ctxWithS, streamer := stream.WithListener(ctx, false, listener) + // includeDetails=true:任务通道的 detail(思考链/工具调用过程)全量持久化并推送 + // (2026-09-28 过程流式设计 C.6——"刷新后过程消失"违背透明可追溯原则; + // 认知负荷由前端默认折叠保证,旧前端忽略 detail 字段无破坏)。 + ctxWithS, streamer := stream.WithListener(ctx, true, listener) var taskResult *bizdev.TaskRunResult var execErr error diff --git a/web/src/App.tsx b/web/src/App.tsx index 897d51a6..1fa2b7dd 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -112,7 +112,7 @@ import { RequestEpoch, shareInflight } from './data/initFanout'; import { resolveSessionProjectId, } from './lib/sessionAdapter'; -import { navigateToAnchor, parseAnchorHash } from './engine/anchorNavigator'; +import { navigateToAnchor, parseAnchorHash, parseSubagentsHash, type SubagentsTarget } from './engine/anchorNavigator'; import { useUnreadMarks } from './hooks/useUnreadMarks'; import { fetchChatCategories, fetchChatDates } from './api/chatWindow'; import { ChatCategoryStat, ChatDateStat, QuotedMessage } from './types/workbench'; @@ -130,6 +130,8 @@ export { const loadAdminConsoleView = () => import('./components/admin/console/AdminConsoleView').then((m) => ({ default: m.AdminConsoleView })); +const loadSubagentsView = () => + import('./components/subagents/SubagentsView').then((m) => ({ default: m.SubagentsView })); export default function App() { // Wave 2:启动时幂等剔除 localStorage 中的 demo seed 残留(须先于 load*) purgeDemoSeedsFromLocalStorage(); @@ -324,7 +326,8 @@ export default function App() { const [rightsModalOpen, setRightsModalOpen] = useState(false); const [userSettingsOpen, setUserSettingsOpen] = useState(false); // 顶层视图切换:单屏工作台 / 全屏系统管理台(仅管理员可进入) - const [view, setView] = useState<'workbench' | 'admin'>('workbench'); + const [view, setView] = useState<'workbench' | 'admin' | 'subagents'>('workbench'); + const [subagentsTarget, setSubagentsTarget] = useState(null); const [docVisibilityModalOpen, setDocVisibilityModalOpen] = useState(false); const [targetDocForVisibility, setTargetDocForVisibility] = useState(null); @@ -1134,6 +1137,16 @@ export default function App() { useEffect(() => { const handleHashChange = () => { if (sessionValidatedRef.current !== true) return; + const sub = parseSubagentsHash(window.location.hash); + if (sub) { + setSubagentsTarget(sub); + setView('subagents'); + if (sub.sessionId && sub.sessionId !== activeSessionIdRef.current) { + setActiveSessionId(sub.sessionId); + } + return; + } + setView((current) => (current === 'subagents' ? 'workbench' : current)); const parsed = parseAnchorHash(window.location.hash); if (!parsed) return; const currentActive = activeSessionIdRef.current; @@ -1161,6 +1174,14 @@ export default function App() { return () => window.removeEventListener('hashchange', handleHashChange); }, []); // 仅在挂载时监听 hashchange,避免 activeSessionId 变更时因残留 hash 产生回弹 + useEffect(() => { + if (sessionValidated !== true) return; + const sub = parseSubagentsHash(window.location.hash); + if (!sub) return; + setSubagentsTarget(sub); + setView('subagents'); + }, [sessionValidated]); + // 引用消息交互 const handleQuoteMessage = (message: ChatMessage) => { let msgId = message.serverId; @@ -1591,6 +1612,25 @@ export default function App() { ); } + if (view === 'subagents' && subagentsTarget) { + return ( + { + setView('workbench'); + const next = `${window.location.pathname}${window.location.search}`; + window.history.replaceState(null, '', next); + }, + }} + /> + ); + } + // 全屏系统管理台:与工作台并列的顶层视图,不经过 AppShell if (view === 'admin') { return ( diff --git a/web/src/api/chatStream.ts b/web/src/api/chatStream.ts index 70932fd6..d393aded 100644 --- a/web/src/api/chatStream.ts +++ b/web/src/api/chatStream.ts @@ -18,6 +18,24 @@ export type StreamEvent = { status?: { message?: string; intent?: string }; delta?: string; error?: string; + detail?: { + category?: string; + step_id?: string; + role?: string; + title?: string; + content?: string; + collapsible?: boolean; + item_id?: string; + delta?: string; + done?: boolean; + tool?: { + name?: string; + status?: string; + input?: string; + output?: string; + duration_ms?: number; + }; + }; deliverable?: { task_key?: string; title?: string; diff --git a/web/src/api/taskRuns.ts b/web/src/api/taskRuns.ts index e09f9f17..d36b2e07 100644 --- a/web/src/api/taskRuns.ts +++ b/web/src/api/taskRuns.ts @@ -215,3 +215,26 @@ export async function getTaskRun(runId: string) { } return res.json(); } + +export type SquadMemberDTO = { + role: string; + name: string; + responsibilities: string[]; + source?: string; +}; + +export type SquadProjection = { + members: SquadMemberDTO[]; + task_key: string; + title: string; +}; + +export async function getTaskRunSquad(runId: string): Promise { + const res = await apiFetch(`/api/v1/tasks/runs/${encodeURIComponent(runId)}/squad`, { + headers: getBearerAuthHeaders(), + }); + if (!res.ok) { + throw new Error(`get task run squad failed: HTTP ${res.status}`); + } + return res.json(); +} diff --git a/web/src/components/chat/TaskRunCard.tsx b/web/src/components/chat/TaskRunCard.tsx index 0128516b..589494f4 100644 --- a/web/src/components/chat/TaskRunCard.tsx +++ b/web/src/components/chat/TaskRunCard.tsx @@ -9,6 +9,8 @@ import { runningPhaseDetail, type TaskRunCardBadgeTone, } from '../../lib/taskRunCard'; +import { subagentsHash } from '../../engine/anchorNavigator'; +import { summarizeProcess } from '../../engine/processEvent'; export interface TaskRunCardProps { message: ChatMessage; @@ -274,6 +276,33 @@ export const TaskRunCard: React.FC = ({ message }) => { {artifactBits.join(' · ')} )} + + {message.taskRunId && message.sessionId && ( + + )} ); }; diff --git a/web/src/components/layout/WorkbenchView.tsx b/web/src/components/layout/WorkbenchView.tsx index e2ece52a..77b27623 100644 --- a/web/src/components/layout/WorkbenchView.tsx +++ b/web/src/components/layout/WorkbenchView.tsx @@ -43,7 +43,7 @@ export interface WorkbenchViewProps { currentDocRole: UserRole; setRightsModalOpen: Dispatch>; setUserSettingsOpen: Dispatch>; - setView: (view: 'workbench' | 'admin') => void; + setView: (view: 'workbench' | 'admin' | 'subagents') => void; setProjectMembersModalOpen: Dispatch>; currentWfState: WorkbenchWorkflowState; workflowGovernanceEnabled: boolean; diff --git a/web/src/components/subagents/CollapsibleBlock.tsx b/web/src/components/subagents/CollapsibleBlock.tsx new file mode 100644 index 00000000..9c535f8e --- /dev/null +++ b/web/src/components/subagents/CollapsibleBlock.tsx @@ -0,0 +1,32 @@ +import { useState, type ReactNode } from 'react'; + +export function CollapsibleBlock(props: { + summary: string; + children: ReactNode; + defaultOpen?: boolean; + testId?: string; +}) { + const [open, setOpen] = useState(props.defaultOpen ?? false); + return ( +
+ + {open ?
{props.children}
: null} +
+ ); +} diff --git a/web/src/components/subagents/ProcessStepOutput.tsx b/web/src/components/subagents/ProcessStepOutput.tsx new file mode 100644 index 00000000..35865236 --- /dev/null +++ b/web/src/components/subagents/ProcessStepOutput.tsx @@ -0,0 +1,12 @@ +import type { ProcessItem } from '../../engine/processEvent'; +import { renderBlockMarkdown } from '../../engine/renderBlockMarkdown'; +import { CollapsibleBlock } from './CollapsibleBlock'; + +export function ProcessStepOutput({ item }: { item: ProcessItem }) { + const title = item.title || '步骤产出'; + return ( + +
+ + ); +} diff --git a/web/src/components/subagents/SubagentsLayout.test.ts b/web/src/components/subagents/SubagentsLayout.test.ts new file mode 100644 index 00000000..d446c8fb --- /dev/null +++ b/web/src/components/subagents/SubagentsLayout.test.ts @@ -0,0 +1,49 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import React from 'react'; +import { renderToStaticMarkup } from 'react-dom/server'; +import { SubagentsLayout } from './SubagentsLayout'; + +test('SubagentsLayout shows the running member and folded thought summary', () => { + const html = renderToStaticMarkup( + React.createElement(SubagentsLayout, { + title: '用户需求收集', + runId: 'run-abcdef123456', + members: [ + { role: 'regulatory', name: '法规专员', responsibilities: ['筛法规'] }, + { role: 'writer', name: '撰写专员', responsibilities: ['成稿'] }, + ], + phases: [ + { step: 'step_executing#2', icon: '⚙️', label: '撰写中', status: 'running', role: 'writer' }, + ], + items: [ + { + key: 'tasks/02/thought-1', + category: 'agent_thought', + role: 'writer', + stepId: 'tasks/02', + text: '先核对素材再写用户需求表的适用范围说明', + status: 'done', + updatedAt: 1, + }, + { + key: 'tasks/01/thought-9', + category: 'agent_thought', + role: 'regulatory', + stepId: 'tasks/01', + text: '法规侧不应出现在撰写栏', + status: 'done', + updatedAt: 2, + }, + ], + onSelectRole: () => undefined, + onExit: () => undefined, + }), + ); + assert.match(html, /data-testid="subagents-view"/); + assert.match(html, /撰写专员/); + assert.match(html, /先核对素材/); + assert.doesNotMatch(html, /法规侧不应出现/); + assert.match(html, /返回对话/); + assert.match(html, /思考 · /); +}); diff --git a/web/src/components/subagents/SubagentsLayout.tsx b/web/src/components/subagents/SubagentsLayout.tsx new file mode 100644 index 00000000..73861a48 --- /dev/null +++ b/web/src/components/subagents/SubagentsLayout.tsx @@ -0,0 +1,104 @@ +import type { SquadMemberDTO } from '../../api/taskRuns'; +import { processItemsOf, type ProcessItem } from '../../engine/processEvent'; +import { defaultMemberRole, memberRunStatus } from '../../lib/squadMembers'; +import type { PhaseEvent } from '../../types/workbench'; +import { ProcessStepOutput } from './ProcessStepOutput'; +import { ThinkingBlock } from './ThinkingBlock'; +import { ToolCallCard } from './ToolCallCard'; + +const STATUS_LABEL = { + waiting: '等待', + running: '执行中', + completed: '完成', + failed: '异常', +} as const; + +export function SubagentsLayout(props: { + title: string; + runId: string; + members: SquadMemberDTO[]; + items: ProcessItem[]; + phases: PhaseEvent[]; + selectedRole?: string; + onSelectRole: (role: string) => void; + onExit: () => void; + notice?: string; +}) { + const selected = props.selectedRole && props.members.some((member) => member.role === props.selectedRole) + ? props.selectedRole + : defaultMemberRole(props.members, props.phases); + const member = props.members.find((item) => item.role === selected); + const visible = props.items.filter((item) => item.role === selected || (!item.role && !selected)); + const runLabel = props.runId.length > 12 ? `${props.runId.slice(0, 8)}…` : props.runId; + + return ( +
+
+
+
小队过程 · {props.title || '任务'}
+
run:{runLabel}
+
+ +
+ {props.notice ?

{props.notice}

: null} +
+ +
+ {member ? ( +

+ {member.name} · {member.role} + {member.responsibilities?.length ? ` · ${member.responsibilities.join('、')}` : ''} +

+ ) : ( +

选择一名专员查看过程

+ )} + {visible.length === 0 ?

还没有过程记录

: null} + {visible.map((item) => { + if (item.category === 'tool_call') return ; + if (item.category === 'step_output') return ; + if (item.category === 'agent_thought') return ; + return ; + })} +
+
+
    + {props.phases.map((phase) => ( +
  1. {phase.icon} {phase.label}
  2. + ))} +
+
+ ); +} + +export function itemsForLayout(items: Record): ProcessItem[] { + return processItemsOf(items); +} diff --git a/web/src/components/subagents/SubagentsView.tsx b/web/src/components/subagents/SubagentsView.tsx new file mode 100644 index 00000000..8da68dcc --- /dev/null +++ b/web/src/components/subagents/SubagentsView.tsx @@ -0,0 +1,55 @@ +import { useEffect, useState } from 'react'; +import { getTaskRunSquad, type SquadMemberDTO } from '../../api/taskRuns'; +import { subagentsHash } from '../../engine/anchorNavigator'; +import { useRunProcessStream } from '../../hooks/useRunProcessStream'; +import { itemsForLayout, SubagentsLayout } from './SubagentsLayout'; + +export function SubagentsView(props: { + runId: string; + sessionId: string; + roleId?: string; + onExit: () => void; +}) { + const stream = useRunProcessStream(props.runId); + const [members, setMembers] = useState([]); + const [title, setTitle] = useState(''); + const [notice, setNotice] = useState(); + + useEffect(() => { + let cancelled = false; + void getTaskRunSquad(props.runId) + .then((squad) => { + if (cancelled) return; + setMembers(squad.members || []); + setTitle(squad.title || squad.task_key || ''); + setNotice(undefined); + }) + .catch(() => { + if (!cancelled) setNotice('暂时读不到小队编制,过程事件仍会显示。'); + }); + return () => { + cancelled = true; + }; + }, [props.runId]); + + const selectRole = (role: string) => { + const next = subagentsHash({ sessionId: props.sessionId, runId: props.runId, roleId: role }); + if (window.location.hash !== next) window.location.hash = next; + }; + + const streamNotice = stream.status === 'error' ? '过程流中断,可返回对话后重新进入。' : notice; + + return ( + + ); +} diff --git a/web/src/components/subagents/ThinkingBlock.tsx b/web/src/components/subagents/ThinkingBlock.tsx new file mode 100644 index 00000000..ff7cc694 --- /dev/null +++ b/web/src/components/subagents/ThinkingBlock.tsx @@ -0,0 +1,22 @@ +import type { ProcessItem } from '../../engine/processEvent'; +import { renderBlockMarkdown } from '../../engine/renderBlockMarkdown'; +import { CollapsibleBlock } from './CollapsibleBlock'; + +function summaryOf(item: ProcessItem): string { + const head = item.text.replace(/\s+/g, ' ').trim().slice(0, 40); + const extra = item.text.length > 40 ? '…' : ''; + const state = item.status === 'streaming' ? '思考中' : item.status === 'error' ? '思考中断' : '思考'; + return `${state} · ${head}${extra}(${item.text.length} 字)`; +} + +export function ThinkingBlock({ item }: { item: ProcessItem }) { + return ( + +
+ + ); +} diff --git a/web/src/components/subagents/ToolCallCard.tsx b/web/src/components/subagents/ToolCallCard.tsx new file mode 100644 index 00000000..8d99aba8 --- /dev/null +++ b/web/src/components/subagents/ToolCallCard.tsx @@ -0,0 +1,24 @@ +import type { ProcessItem } from '../../engine/processEvent'; +import { renderBlockMarkdown } from '../../engine/renderBlockMarkdown'; +import { CollapsibleBlock } from './CollapsibleBlock'; + +function statusLabel(status: string): string { + if (status === 'succeeded') return '完成'; + if (status === 'failed') return '失败'; + return '调用中'; +} + +export function ToolCallCard({ item }: { item: ProcessItem }) { + const tool = item.tool; + const name = tool?.name || item.title || '工具'; + const status = tool?.status || (item.status === 'error' ? 'failed' : 'calling'); + const duration = tool?.durationMs != null ? ` · ${tool.durationMs}ms` : ''; + const body = [tool?.input ? `入参\n\`\`\`json\n${tool.input}\n\`\`\`` : '', tool?.output ? `结果\n\`\`\`\n${tool.output}\n\`\`\`` : ''] + .filter(Boolean) + .join('\n\n'); + return ( + + {body ?
:

无入参或结果

} + + ); +} diff --git a/web/src/engine/anchorNavigator.test.ts b/web/src/engine/anchorNavigator.test.ts index 0ca100d6..bc82c6b5 100644 --- a/web/src/engine/anchorNavigator.test.ts +++ b/web/src/engine/anchorNavigator.test.ts @@ -4,7 +4,9 @@ import { parseAnchorHash, formatAnchorHash, navigateToAnchor, + parseSubagentsHash, resolveTurnAnchorId, + subagentsHash, } from './anchorNavigator'; test('parseAnchorHash: 正确解析标准深链 hash', () => { @@ -153,3 +155,19 @@ test('resolveTurnAnchorId: 方案 A 轮次归一化测试', () => { assert.equal(resolveTurnAnchorId(102, []), 102, 'empty messages returns raw id'); }); +test('parseSubagentsHash 解析小队过程深链且不误伤消息锚点', () => { + assert.equal(parseAnchorHash('#/s/sess/run/run-1/agents'), null); + assert.deepEqual(parseSubagentsHash('#/s/sess/run/run-1/agents'), { + sessionId: 'sess', + runId: 'run-1', + roleId: undefined, + }); + assert.deepEqual(parseSubagentsHash('#/s/sess/run/run-1/agents/regulatory'), { + sessionId: 'sess', + runId: 'run-1', + roleId: 'regulatory', + }); + const hashed = subagentsHash({ sessionId: 'a/b', runId: 'run 1', roleId: '法规' }); + assert.deepEqual(parseSubagentsHash(hashed), { sessionId: 'a/b', runId: 'run 1', roleId: '法规' }); +}); + diff --git a/web/src/engine/anchorNavigator.ts b/web/src/engine/anchorNavigator.ts index 335afc4a..8cdec77d 100644 --- a/web/src/engine/anchorNavigator.ts +++ b/web/src/engine/anchorNavigator.ts @@ -110,6 +110,33 @@ export function parseAnchorHash(hash: string): AnchorTarget | null { return { sessionId, messageId }; } +export type SubagentsTarget = { + sessionId: string; + runId: string; + roleId?: string; +}; + +/** + * 小队过程子页面深链:#/s/{sessionId}/run/{runId}/agents[/{roleId}] + * 与消息锚点 #/s/{sessionId}/m/{messageId} 互斥。 + */ +export function parseSubagentsHash(hash: string): SubagentsTarget | null { + if (!hash) return null; + const cleaned = hash.startsWith('#') ? hash.substring(1) : hash; + const match = cleaned.match(/^\/s\/([^/]+)\/run\/([^/]+)\/agents(?:\/([^/]+))?$/); + if (!match) return null; + const sessionId = decodeURIComponent(match[1]); + const runId = decodeURIComponent(match[2]); + if (!sessionId || !runId) return null; + const roleId = match[3] ? decodeURIComponent(match[3]) : undefined; + return { sessionId, runId, roleId }; +} + +export function subagentsHash(target: SubagentsTarget): string { + const base = `#/s/${encodeURIComponent(target.sessionId)}/run/${encodeURIComponent(target.runId)}/agents`; + return target.roleId ? `${base}/${encodeURIComponent(target.roleId)}` : base; +} + /** * 触发 DOM 元素的锚点高亮脉冲 */ diff --git a/web/src/engine/messageIdentity.ts b/web/src/engine/messageIdentity.ts index 885d5975..1ded23ec 100644 --- a/web/src/engine/messageIdentity.ts +++ b/web/src/engine/messageIdentity.ts @@ -114,6 +114,10 @@ export function mergeLiveOverHistory( clientTurnId: live.clientTurnId ?? history.clientTurnId, clientMessageId: live.clientMessageId ?? history.clientMessageId, phaseEvents: live.phaseEvents?.length ? live.phaseEvents : history.phaseEvents, + processItems: + live.processItems && Object.keys(live.processItems).length > 0 + ? live.processItems + : history.processItems, livePhaseLabel: preferHistoryTerminal ? undefined : live.livePhaseLabel ?? history.livePhaseLabel, diff --git a/web/src/engine/phaseEventMapper.ts b/web/src/engine/phaseEventMapper.ts index 2bf5c611..5a79927c 100644 --- a/web/src/engine/phaseEventMapper.ts +++ b/web/src/engine/phaseEventMapper.ts @@ -83,17 +83,19 @@ export function applyPhaseEvent(strip: PhaseEvent[], phase?: PhaseEventPayload): const base = detail || meta.label; const label = decorateLabel(base, phase); const key = stepKeyOf(phase); + const role = phase.role?.trim() || undefined; // 已存在的条目:刷新文案;非失败事件再次到达时,failed 条目恢复 running(重试/续跑重新执行该步骤) const existingIdx = strip.findIndex((e) => e.step === key); if (existingIdx >= 0) { return strip.map((e, i) => { if (i !== existingIdx) return e; + const nextRole = role || e.role; if (e.status === 'failed' && phase.phase_id !== 'step_failed') { - return { ...e, label, detail, status: 'running' as const }; + return { ...e, label, detail, role: nextRole, status: 'running' as const }; } // 已收口条目被重复事件刷新文案时,保持 label 与 status 语义一致。 - return settleStep({ ...e, label, detail }); + return settleStep({ ...e, label, detail, role: nextRole }); }); } @@ -101,10 +103,12 @@ export function applyPhaseEvent(strip: PhaseEvent[], phase?: PhaseEventPayload): if (phase.phase_id === 'step_failed') { if (strip.some((e) => e.status === 'running')) { return strip.map((e) => - e.status === 'running' ? { ...e, status: 'failed' as const, detail: detail ?? e.detail } : e, + e.status === 'running' + ? { ...e, status: 'failed' as const, detail: detail ?? e.detail, role: role || e.role } + : e, ); } - return [...strip, { step: key, icon: meta.icon, label, detail, status: 'failed' as const }]; + return [...strip, { step: key, icon: meta.icon, label, detail, role, status: 'failed' as const }]; } // 收口型 phase:running 条目置 completed,不追加新节点 @@ -118,6 +122,6 @@ export function applyPhaseEvent(strip: PhaseEvent[], phase?: PhaseEventPayload): ); return [ ...settled, - { step: key, icon: meta.icon, label, detail, status: 'running' as const }, + { step: key, icon: meta.icon, label, detail, role, status: 'running' as const }, ]; } \ No newline at end of file diff --git a/web/src/engine/processEvent.test.ts b/web/src/engine/processEvent.test.ts new file mode 100644 index 00000000..7ce0b2d9 --- /dev/null +++ b/web/src/engine/processEvent.test.ts @@ -0,0 +1,86 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { applyDetailEvent, processStamp, summarizeProcess } from './processEvent'; + +test('applyDetailEvent appends reasoning deltas then closes on done', () => { + let items = applyDetailEvent({}, { + category: 'agent_thought', + step_id: 'tasks/01', + role: 'regulatory', + item_id: 'thought-1', + delta: '先看', + }, 1); + items = applyDetailEvent(items, { + category: 'agent_thought', + step_id: 'tasks/01', + item_id: 'thought-1', + delta: '法规', + }, 2); + assert.equal(items['tasks/01/thought-1'].text, '先看法规'); + assert.equal(items['tasks/01/thought-1'].status, 'streaming'); + items = applyDetailEvent(items, { + category: 'agent_thought', + step_id: 'tasks/01', + item_id: 'thought-1', + done: true, + }, 3); + assert.equal(items['tasks/01/thought-1'].status, 'done'); + assert.equal(items['tasks/01/thought-1'].text, '先看法规'); +}); + +test('content snapshot replaces accumulated delta', () => { + let items = applyDetailEvent({}, { + category: 'agent_thought', + item_id: 't', + delta: '旧', + }, 1); + items = applyDetailEvent(items, { + category: 'agent_thought', + item_id: 't', + content: '对账后的全文', + }, 2); + assert.equal(items['/t'].text, '对账后的全文'); +}); + +test('tool frames merge status without dropping the name', () => { + let items = applyDetailEvent({}, { + category: 'tool_call', + role: 'writer', + step_id: 'tasks/02', + item_id: 'call-1', + tool: { name: 'read_file', status: 'calling', input: 'output.md' }, + }, 1); + items = applyDetailEvent(items, { + category: 'tool_call', + item_id: 'call-1', + step_id: 'tasks/02', + tool: { status: 'failed', output: 'missing' }, + done: true, + }, 2); + const tool = items['tasks/02/call-1'].tool; + assert.equal(tool?.name, 'read_file'); + assert.equal(tool?.status, 'failed'); + assert.equal(tool?.input, 'output.md'); + assert.equal(tool?.output, 'missing'); + assert.equal(items['tasks/02/call-1'].status, 'error'); +}); + +test('summarizeProcess counts roles tools and thoughts', () => { + const items = applyDetailEvent( + applyDetailEvent({}, { + category: 'agent_thought', + role: 'regulatory', + item_id: 'a', + delta: '想', + }, 1), + { + category: 'tool_call', + role: 'writer', + item_id: 'b', + tool: { name: 'write_file', status: 'succeeded' }, + }, + 2, + ); + assert.equal(summarizeProcess(items), '2 名专员 · 1 次工具调用 · 1 段推演'); + assert.notEqual(processStamp(items), processStamp({})); +}); diff --git a/web/src/engine/processEvent.ts b/web/src/engine/processEvent.ts new file mode 100644 index 00000000..5c50e0d5 --- /dev/null +++ b/web/src/engine/processEvent.ts @@ -0,0 +1,133 @@ +/** 小队过程 detail 帧的纯函数聚合。气泡与子页面共用,不写进消息正文。 */ + +export type ProcessCategory = 'agent_thought' | 'tool_call' | 'step_output' | 'log'; + +export type ProcessItemStatus = 'streaming' | 'done' | 'error'; + +export type ProcessTool = { + name: string; + status: string; + input?: string; + output?: string; + durationMs?: number; +}; + +export type ProcessItem = { + key: string; + category: ProcessCategory; + role: string; + stepId: string; + title?: string; + text: string; + status: ProcessItemStatus; + tool?: ProcessTool; + updatedAt: number; +}; + +export type DetailEvent = { + category?: string; + step_id?: string; + role?: string; + title?: string; + content?: string; + collapsible?: boolean; + item_id?: string; + delta?: string; + done?: boolean; + tool?: { + name?: string; + status?: string; + input?: string; + output?: string; + duration_ms?: number; + }; +}; + +const CATEGORIES = new Set(['agent_thought', 'tool_call', 'step_output', 'log']); + +function asCategory(raw?: string): ProcessCategory { + if (raw && CATEGORIES.has(raw as ProcessCategory)) return raw as ProcessCategory; + return 'log'; +} + +function mergeTool(prev: ProcessTool | undefined, incoming: DetailEvent['tool']): ProcessTool | undefined { + if (!incoming && !prev) return undefined; + return { + name: incoming?.name || prev?.name || '', + status: incoming?.status || prev?.status || 'calling', + input: incoming?.input ?? prev?.input, + output: incoming?.output ?? prev?.output, + durationMs: incoming?.duration_ms ?? prev?.durationMs, + }; +} + +function nextStatus(prev: ProcessItemStatus | undefined, ev: DetailEvent, tool?: ProcessTool): ProcessItemStatus { + if (tool?.status === 'failed') return 'error'; + if (ev.done || tool?.status === 'succeeded') return 'done'; + if (ev.delta || ev.content) return prev === 'done' || prev === 'error' ? prev : 'streaming'; + return prev ?? 'streaming'; +} + +/** + * 把一条 detail 帧折叠进过程项表。 + * delta 追加;content 是快照对账(覆盖);done / tool 成功收口。 + */ +export function applyDetailEvent( + items: Record, + ev?: DetailEvent | null, + now = Date.now(), +): Record { + if (!ev?.category && !ev?.item_id && !ev?.delta && !ev?.content) return items; + const stepId = ev.step_id || ''; + const itemId = ev.item_id || ev.title || ev.category || 'item'; + const key = `${stepId}/${itemId}`; + const prev = items[key]; + let text = prev?.text ?? ''; + if (typeof ev.content === 'string' && ev.content.length > 0) { + text = ev.content; + } else if (ev.delta) { + text += ev.delta; + } + const tool = mergeTool(prev?.tool, ev.tool); + const status = nextStatus(prev?.status, ev, tool); + return { + ...items, + [key]: { + key, + category: asCategory(ev.category || prev?.category), + role: ev.role || prev?.role || '', + stepId, + title: ev.title || prev?.title, + text, + status, + tool, + updatedAt: now, + }, + }; +} + +export function processItemsOf(items?: Record | null): ProcessItem[] { + if (!items) return []; + return Object.values(items).sort((a, b) => a.updatedAt - b.updatedAt || a.key.localeCompare(b.key)); +} + +/** 气泡 L3 摘要:专员数 · 工具次数 · 推演段数。空则返回空串。 */ +export function summarizeProcess(items?: Record | null): string { + const list = processItemsOf(items); + if (list.length === 0) return ''; + const roles = new Set(list.map((item) => item.role).filter(Boolean)); + const tools = list.filter((item) => item.category === 'tool_call').length; + const thoughts = list.filter((item) => item.category === 'agent_thought').length; + const parts: string[] = []; + if (roles.size > 0) parts.push(`${roles.size} 名专员`); + if (tools > 0) parts.push(`${tools} 次工具调用`); + if (thoughts > 0) parts.push(`${thoughts} 段推演`); + return parts.join(' · '); +} + +/** 供消息合并判断过程项是否变化(不计时间戳)。 */ +export function processStamp(items?: Record | null): string { + return processItemsOf(items) + .map((item) => `${item.key}:${item.status}:${item.text.length}:${item.tool?.status ?? ''}`) + .join('|'); +} diff --git a/web/src/engine/renderBlockMarkdown.test.ts b/web/src/engine/renderBlockMarkdown.test.ts new file mode 100644 index 00000000..39d66134 --- /dev/null +++ b/web/src/engine/renderBlockMarkdown.test.ts @@ -0,0 +1,11 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { renderBlockMarkdown } from './renderBlockMarkdown'; + +test('renderBlockMarkdown escapes html and pretty-prints json fences', () => { + const html = renderBlockMarkdown('see