merge: candidate subagent-process-view-impl-2026-09-28 into dev

# Conflicts:
#	.doc/ops/CURRENT.md
This commit is contained in:
weichen
2026-09-28 10:36:26 +00:00
47 changed files with 2405 additions and 18 deletions
@@ -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 剥离对象修正为 `<think>…</think>`(原文误写为特殊符号)
- 无硬编码模型名(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)
@@ -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>` 块从产物中剥离(丢弃)。设计上不改变"正文不含 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<N>` |
| `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<N>` |
| `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<string, ProcessItem>, ev: DetailEvent): Record<string, ProcessItem>
// 纯函数(与 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`
+8
View File
@@ -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)
| 项 | 链接 |
+62 -1
View File
@@ -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)) {
+13
View File
@@ -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 (
+87 -1
View File
@@ -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()
+9
View File
@@ -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 {
+12
View File
@@ -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 {
+33
View File
@@ -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)
}
}
+6 -4
View File
@@ -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
+7
View File
@@ -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:
+1
View File
@@ -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)
+3
View File
@@ -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
+42
View File
@@ -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 {
+406
View File
@@ -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
}
+215
View File
@@ -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)
}
}
}
+60
View File
@@ -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
}
+38
View File
@@ -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")
}
}
+4
View File
@@ -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)
+19 -3
View File
@@ -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轨正式合规文档交付载荷
+4 -1
View File
@@ -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
+42 -2
View File
@@ -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<SubagentsTarget | null>(null);
const [docVisibilityModalOpen, setDocVisibilityModalOpen] = useState(false);
const [targetDocForVisibility, setTargetDocForVisibility] = useState<DeliverableItem | null>(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 (
<OnDemand
label="小队过程"
load={loadSubagentsView}
componentProps={{
runId: subagentsTarget.runId,
sessionId: subagentsTarget.sessionId,
roleId: subagentsTarget.roleId,
onExit: () => {
setView('workbench');
const next = `${window.location.pathname}${window.location.search}`;
window.history.replaceState(null, '', next);
},
}}
/>
);
}
// 全屏系统管理台:与工作台并列的顶层视图,不经过 AppShell
if (view === 'admin') {
return (
+18
View File
@@ -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;
+23
View File
@@ -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<SquadProjection> {
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();
}
+29
View File
@@ -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<TaskRunCardProps> = ({ message }) => {
<span style={{ wordBreak: 'break-word' }}>{artifactBits.join(' · ')}</span>
</div>
)}
{message.taskRunId && message.sessionId && (
<button
type="button"
data-testid="task-run-card-subagents"
onClick={() => {
window.location.hash = subagentsHash({
sessionId: message.sessionId,
runId: message.taskRunId || '',
});
}}
style={{
display: 'block',
width: '100%',
textAlign: 'left',
border: 0,
borderTop: '1px solid var(--border)',
background: 'transparent',
padding: '8px 12px',
fontSize: '12px',
color: 'var(--brand-500)',
cursor: 'pointer',
}}
>
{isLive ? '查看小队过程 →' : summarizeProcess(message.processItems) || '查看小队过程 →'}
</button>
)}
</div>
);
};
+1 -1
View File
@@ -43,7 +43,7 @@ export interface WorkbenchViewProps {
currentDocRole: UserRole;
setRightsModalOpen: Dispatch<SetStateAction<boolean>>;
setUserSettingsOpen: Dispatch<SetStateAction<boolean>>;
setView: (view: 'workbench' | 'admin') => void;
setView: (view: 'workbench' | 'admin' | 'subagents') => void;
setProjectMembersModalOpen: Dispatch<SetStateAction<boolean>>;
currentWfState: WorkbenchWorkflowState;
workflowGovernanceEnabled: boolean;
@@ -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 (
<div data-testid={props.testId} style={{ border: '1px solid var(--border)', borderRadius: 8, marginBottom: 8 }}>
<button
type="button"
aria-expanded={open}
onClick={() => setOpen((value) => !value)}
style={{
width: '100%',
textAlign: 'left',
background: 'transparent',
border: 0,
padding: '8px 10px',
cursor: 'pointer',
fontSize: 13,
color: 'var(--text)',
}}
>
{props.summary}
</button>
{open ? <div style={{ padding: '0 10px 10px', fontSize: 13 }}>{props.children}</div> : null}
</div>
);
}
@@ -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 (
<CollapsibleBlock testId="process-step-output" summary={`产出 · ${title}`}>
<div dangerouslySetInnerHTML={{ __html: renderBlockMarkdown(item.text) }} />
</CollapsibleBlock>
);
}
@@ -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, /思考 · /);
});
@@ -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 (
<div data-testid="subagents-view" style={{ minHeight: '100vh', background: 'var(--bg, #f8fafc)', color: 'var(--text, #0f172a)' }}>
<header style={{ display: 'flex', justifyContent: 'space-between', gap: 12, padding: '12px 16px', borderBottom: '1px solid var(--border, #e2e8f0)' }}>
<div>
<div style={{ fontWeight: 700 }}>小队过程 · {props.title || '任务'}</div>
<div style={{ fontSize: 12, color: 'var(--text-faint, #64748b)' }}>run:{runLabel}</div>
</div>
<button type="button" onClick={props.onExit}>返回对话</button>
</header>
{props.notice ? <p role="status" style={{ margin: '8px 16px' }}>{props.notice}</p> : null}
<div style={{ display: 'grid', gridTemplateColumns: '240px 1fr', minHeight: 'calc(100vh - 120px)' }}>
<aside data-testid="subagents-members" style={{ borderRight: '1px solid var(--border, #e2e8f0)', padding: 8 }}>
{props.members.length === 0 ? <p>还没有小队编制</p> : null}
{props.members.map((item) => {
const status = memberRunStatus(item.role, props.phases);
const active = item.role === selected;
return (
<button
key={item.role}
type="button"
data-testid="subagents-member"
data-role={item.role}
aria-pressed={active}
onClick={() => props.onSelectRole(item.role)}
style={{
display: 'block',
width: '100%',
textAlign: 'left',
marginBottom: 6,
padding: 8,
borderRadius: 8,
border: active ? '1px solid var(--brand-500, #2563eb)' : '1px solid transparent',
background: active ? 'rgba(37, 99, 235, 0.08)' : 'transparent',
}}
>
<div style={{ fontWeight: 650 }}>{item.name || item.role}</div>
<div style={{ fontSize: 12 }}>{STATUS_LABEL[status]}</div>
</button>
);
})}
</aside>
<section data-testid="subagents-detail" style={{ padding: 12 }}>
{member ? (
<p style={{ marginTop: 0 }}>
{member.name} · {member.role}
{member.responsibilities?.length ? ` · ${member.responsibilities.join('、')}` : ''}
</p>
) : (
<p>选择一名专员查看过程</p>
)}
{visible.length === 0 ? <p>还没有过程记录</p> : null}
{visible.map((item) => {
if (item.category === 'tool_call') return <ToolCallCard key={item.key} item={item} />;
if (item.category === 'step_output') return <ProcessStepOutput key={item.key} item={item} />;
if (item.category === 'agent_thought') return <ThinkingBlock key={item.key} item={item} />;
return <ProcessStepOutput key={item.key} item={item} />;
})}
</section>
</div>
<ol data-testid="subagents-timeline" style={{ listStyle: 'none', margin: 0, padding: '8px 16px', borderTop: '1px solid var(--border, #e2e8f0)', fontSize: 12 }}>
{props.phases.map((phase) => (
<li key={phase.step}>{phase.icon} {phase.label}</li>
))}
</ol>
</div>
);
}
export function itemsForLayout(items: Record<string, ProcessItem>): ProcessItem[] {
return processItemsOf(items);
}
@@ -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<SquadMemberDTO[]>([]);
const [title, setTitle] = useState('');
const [notice, setNotice] = useState<string | undefined>();
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 (
<SubagentsLayout
title={title}
runId={props.runId}
members={members}
items={itemsForLayout(stream.items)}
phases={stream.phases}
selectedRole={props.roleId}
onSelectRole={selectRole}
onExit={props.onExit}
notice={streamNotice}
/>
);
}
@@ -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 (
<CollapsibleBlock
testId="thinking-block"
defaultOpen={item.status === 'streaming'}
summary={summaryOf(item)}
>
<div dangerouslySetInnerHTML={{ __html: renderBlockMarkdown(item.text) }} />
</CollapsibleBlock>
);
}
@@ -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 (
<CollapsibleBlock testId="tool-call-card" summary={`工具 ${name} · ${statusLabel(status)}${duration}`}>
{body ? <div dangerouslySetInnerHTML={{ __html: renderBlockMarkdown(body) }} /> : <p>无入参或结果</p>}
</CollapsibleBlock>
);
}
+18
View File
@@ -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: '法规' });
});
+27
View File
@@ -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 元素的锚点高亮脉冲
*/
+4
View File
@@ -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,
+9 -5
View File
@@ -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 },
];
}
+86
View File
@@ -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({}));
});
+133
View File
@@ -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<ProcessCategory>(['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<string, ProcessItem>,
ev?: DetailEvent | null,
now = Date.now(),
): Record<string, ProcessItem> {
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<string, ProcessItem> | 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<string, ProcessItem> | 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<string, ProcessItem> | null): string {
return processItemsOf(items)
.map((item) => `${item.key}:${item.status}:${item.text.length}:${item.tool?.status ?? ''}`)
.join('|');
}
@@ -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 <script>\n```json\n{"a":1}\n```');
assert.doesNotMatch(html, /<script>/);
assert.match(html, /&lt;script&gt;/);
assert.match(html, /<pre><code class="lang-json">/);
assert.match(html, /&quot;a&quot;: 1/);
});
+39
View File
@@ -0,0 +1,39 @@
/** 工具入参/结果用的轻量块级渲染:转义优先,只识别 fenced code。不引入新依赖。 */
function escapeHtml(value: string): string {
return value
.replace(/&/g, '&amp;')
.replace(/</g, '&lt;')
.replace(/>/g, '&gt;')
.replace(/"/g, '&quot;');
}
function fenceBody(lang: string, body: string): string {
let text = body.replace(/\n$/, '');
if (lang === 'json') {
try {
text = JSON.stringify(JSON.parse(text), null, 2);
} catch {
// 保持原文,仍然转义。
}
}
const klass = lang ? ` class="lang-${escapeHtml(lang)}"` : '';
return `<pre><code${klass}>${escapeHtml(text)}</code></pre>`;
}
export function renderBlockMarkdown(source: string): string {
const text = source ?? '';
const parts: string[] = [];
const re = /```([a-zA-Z0-9_-]*)\n([\s\S]*?)```/g;
let last = 0;
for (const match of text.matchAll(re)) {
const index = match.index ?? 0;
if (index > last) parts.push(`<p>${escapeHtml(text.slice(last, index)).replace(/\n/g, '<br>')}</p>`);
parts.push(fenceBody(match[1] || '', match[2] || ''));
last = index + match[0].length;
}
if (last < text.length) {
parts.push(`<p>${escapeHtml(text.slice(last)).replace(/\n/g, '<br>')}</p>`);
}
return parts.join('');
}
+2
View File
@@ -12,6 +12,7 @@ import {
saveSessionDraft,
} from '../data/storage';
import { enrichMessagesWithCategories } from '../engine/categoryHelper';
import { processStamp } from '../engine/processEvent';
import {
filterPendingLiveMessages,
isLiveMessageInHistory,
@@ -529,6 +530,7 @@ export function useChatWindow({
liveM.status !== m.status ||
liveM.livePhaseLabel !== m.livePhaseLabel ||
(liveM.phaseEvents?.length || 0) !== (m.phaseEvents?.length || 0) ||
processStamp(liveM.processItems) !== processStamp(m.processItems) ||
liveM.errorInfo !== m.errorInfo ||
liveM.serverId !== m.serverId)
) {
+59
View File
@@ -0,0 +1,59 @@
import { useEffect, useState } from 'react';
import type { StreamEvent } from '../api/chatStream';
import { readTaskRunSSEOnce, subscribeTaskRunEvents } from '../api/taskRuns';
import { applyPhaseEvent } from '../engine/phaseEventMapper';
import { applyDetailEvent, type ProcessItem } from '../engine/processEvent';
import type { PhaseEvent } from '../types/workbench';
export type RunProcessStatus = 'loading' | 'live' | 'done' | 'error';
/** 子页面自持订阅:先重放已落库事件,未终态再续订 live。 */
export function useRunProcessStream(runId: string): {
items: Record<string, ProcessItem>;
phases: PhaseEvent[];
status: RunProcessStatus;
} {
const [items, setItems] = useState<Record<string, ProcessItem>>({});
const [phases, setPhases] = useState<PhaseEvent[]>([]);
const [status, setStatus] = useState<RunProcessStatus>('loading');
useEffect(() => {
const controller = new AbortController();
let itemState: Record<string, ProcessItem> = {};
let phaseState: PhaseEvent[] = [];
let terminal = false;
const onEvent = (ev: StreamEvent) => {
if (ev.type === 'detail') {
itemState = applyDetailEvent(itemState, ev.detail);
setItems(itemState);
} else if (ev.type === 'phase') {
phaseState = applyPhaseEvent(phaseState, ev.phase);
setPhases(phaseState);
} else if (ev.type === 'done' || ev.type === 'error') {
terminal = true;
}
};
void (async () => {
try {
const replay = await readTaskRunSSEOnce(runId, 0, onEvent, controller.signal, { replayOnly: true });
if (controller.signal.aborted) return;
if (replay.terminal || terminal) {
setStatus('done');
return;
}
setStatus('live');
await subscribeTaskRunEvents(
runId,
{ onEvent, afterSeq: replay.lastSeq },
controller.signal,
);
if (!controller.signal.aborted) setStatus('done');
} catch {
if (!controller.signal.aborted) setStatus('error');
}
})();
return () => controller.abort();
}, [runId]);
return { items, phases, status };
}
+9
View File
@@ -12,6 +12,7 @@ import { createTaskRun, getTaskRun, resumeTaskRun, subscribeTaskRunEvents } from
import { readActiveTasks, removeActiveTask, touchActiveTaskPhase, upsertActiveTask } from '../lib/activeTaskCache';
import { buildTaskRequest, streamTask } from '../api/taskStream';
import { applyPhaseEvent } from '../engine/phaseEventMapper';
import { applyDetailEvent } from '../engine/processEvent';
import { resolveTaskTemplateForMessage, wrongWorkflowGuidance } from '../engine/templateMatch';
import { GOVERNANCE_BLOCKED_HINT, isBlockedGovernanceStatus } from '../lib/blockedGovernanceCopy';
import {
@@ -501,6 +502,14 @@ export function useTaskRunAttempts(input: TaskRunAttemptsInput) {
}));
break;
}
case 'detail': {
// 过程帧只进 processItems,不拼进气泡正文。
patchAsst((m) => ({
status: m.status === 'pending' ? 'streaming_status' : m.status,
processItems: applyDetailEvent(m.processItems || {}, ev.detail),
}));
break;
}
case 'status': {
if (ev.status?.message) {
patchAsst({ status: 'streaming_status', livePhaseLabel: ev.status.message });
+22
View File
@@ -0,0 +1,22 @@
import assert from 'node:assert/strict';
import test from 'node:test';
import { defaultMemberRole, memberRunStatus } from './squadMembers';
import type { PhaseEvent } from '../types/workbench';
const phases: PhaseEvent[] = [
{ step: 'step_executing#1', icon: '⚙️', label: '法规', status: 'completed', role: 'regulatory' },
{ step: 'step_executing#2', icon: '⚙️', label: '撰写', status: 'running', role: 'writer' },
];
test('memberRunStatus prefers running over older completed', () => {
assert.equal(memberRunStatus('writer', phases), 'running');
assert.equal(memberRunStatus('regulatory', phases), 'completed');
assert.equal(memberRunStatus('leader', phases), 'waiting');
});
test('defaultMemberRole selects the running member', () => {
assert.equal(
defaultMemberRole([{ role: 'regulatory' }, { role: 'writer' }, { role: 'leader' }], phases),
'writer',
);
});
+24
View File
@@ -0,0 +1,24 @@
import type { PhaseEvent } from '../types/workbench';
export type MemberRunStatus = 'waiting' | 'running' | 'completed' | 'failed';
export function memberRunStatus(role: string, phases: PhaseEvent[]): MemberRunStatus {
const related = phases.filter((phase) => phase.role === role);
if (related.some((phase) => phase.status === 'running')) return 'running';
if (related.some((phase) => phase.status === 'failed')) return 'failed';
if (related.some((phase) => phase.status === 'completed')) return 'completed';
return 'waiting';
}
/** 未指定 role 时:当前执行中,否则最近完成,否则编制第一人。 */
export function defaultMemberRole(
members: { role: string }[],
phases: PhaseEvent[],
): string | undefined {
const running = members.find((member) => memberRunStatus(member.role, phases) === 'running');
if (running) return running.role;
for (let i = members.length - 1; i >= 0; i -= 1) {
if (memberRunStatus(members[i].role, phases) === 'completed') return members[i].role;
}
return members[0]?.role;
}
+8
View File
@@ -1,3 +1,4 @@
import type { ProcessItem } from '../engine/processEvent';
import { DocLifecycleState, UserRole } from './docgov';
import { DocVisibilityRule } from './rbac';
import { MessageContentKind, MessageExportCapability } from './replyExport';
@@ -57,6 +58,8 @@ export interface PhaseEvent {
status: 'pending' | 'running' | 'completed' | 'failed';
/** 后端 phase.message 原文(刷新快照 / 时间线当前行展开用) */
detail?: string;
/** 专员 role(step_executing 等按成员归集状态) */
role?: string;
}
/** 服务端写入 chat_messages.metadata.task_phase_snapshot 的折叠进度快照 */
@@ -218,6 +221,11 @@ export interface ChatMessage {
branchTag?: string;
timestamp: string;
phaseEvents?: PhaseEvent[];
/**
* 过程 detail 聚合(思考/工具/步骤产出)。只在气泡摘要与小队子页面使用,
* 不得拼进 content。
*/
processItems?: Record<string, ProcessItem>;
/** 任务总耗时(毫秒);done 事件或 snapshot.duration_ms */
taskDurationMs?: number;
/** 历史消息带来的 phase 快照标记(有则刷新不必再 replay) */