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/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/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/workflow.go b/internal/squad/workflow.go index d8ca8a33..08b0365c 100644 --- a/internal/squad/workflow.go +++ b/internal/squad/workflow.go @@ -264,25 +264,25 @@ func (d *WorkflowDriver) Execute(ctx context.Context, wsDir string, backend agen for _, s := range parallelGroup { sCopy := s wg.Add(1) - go func() { - defer wg.Done() - defer func() { - if rec := recover(); rec != nil { - mu.Lock() - if groupErr == nil { - groupErr = fmt.Errorf("parallel step panic: %v", rec) - } - mu.Unlock() - } - }() - if err := d.runOneStep(runCtx, wsDir, sCopy, backend, timeout, opts.StepNotify, &mu, &producedFiles, ledger); err != nil { - mu.Lock() - if groupErr == nil { - groupErr = err - } - mu.Unlock() - } - }() + go func() { + defer wg.Done() + defer func() { + if rec := recover(); rec != nil { + mu.Lock() + if groupErr == nil { + groupErr = fmt.Errorf("parallel step panic: %v", rec) + } + mu.Unlock() + } + }() + if err := d.runOneStep(runCtx, wsDir, sCopy, backend, timeout, opts.StepNotify, &mu, &producedFiles, ledger); err != nil { + mu.Lock() + if groupErr == nil { + groupErr = err + } + mu.Unlock() + } + }() } wg.Wait() if groupErr != nil { @@ -335,25 +335,25 @@ func (d *WorkflowDriver) Execute(ctx context.Context, wsDir string, backend agen Duration: time.Since(start), }, rErr } - // 治理：受控迁移至执行态（legacy 重跑场景直接放行） - markStepExecuting(wf, step.ID) - if err := d.runOneStep(runCtx, wsDir, step, backend, timeout, opts.StepNotify, &mu, &producedFiles, ledger); err != nil { - // A5 非关键步骤降级：记录 degraded 继续推进，收口统一标 degraded - if !step.IsCritical() { - markStepDegraded(wsDir, wf, step, err) - stepsRun++ + // 治理：受控迁移至执行态（legacy 重跑场景直接放行） + markStepExecuting(wf, step.ID) + if err := d.runOneStep(runCtx, wsDir, step, backend, timeout, opts.StepNotify, &mu, &producedFiles, ledger); err != nil { + // A5 非关键步骤降级：记录 degraded 继续推进，收口统一标 degraded + if !step.IsCritical() { + markStepDegraded(wsDir, wf, step, err) + stepsRun++ + _ = d.SaveWorkflow(wsDir, wf) + continue + } + wf.Status = "failed" _ = d.SaveWorkflow(wsDir, wf) - continue + _ = d.updateWorkspaceStatus(wsDir, "failed") + return &WorkflowExecutionResult{ + Success: false, + ErrorReason: err.Error(), + Duration: time.Since(start), + }, err } - wf.Status = "failed" - _ = d.SaveWorkflow(wsDir, wf) - _ = d.updateWorkspaceStatus(wsDir, "failed") - return &WorkflowExecutionResult{ - Success: false, - ErrorReason: err.Error(), - Duration: time.Since(start), - }, err - } stepsRun++ // 治理：FSM 受控迁移 + 步骤归档锁定 + 软线提醒 if tErr := TransitionStep(wf, step.ID, StepStateCompleted); tErr == nil { @@ -547,16 +547,20 @@ 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 - execErr error - matErr error - duration int64 - ok bool - lastAssess agent.FailureAssessment - basePrompt = string(taskData) + res *agent.Result + execErr error + matErr error + duration int64 + ok bool + 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 0500d316..521ebb48 100644 --- a/internal/taskrun/manager.go +++ b/internal/taskrun/manager.go @@ -361,7 +361,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