feat(stream): 打通小队过程事件到持久化 detail 帧

pi-native 与 runner NDJSON 把推理和工具调用投影为 detail，任务流始终落库；非 200 的流式尝试回落到一次性 /v1/run，避免改变既有错误分类。

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)) {

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 (

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()

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 {

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 {

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)
+	}
+}

internal/agentx/llm/types.go

@@ -29,10 +29,11 @@ const (
 type EventKind string
 
 const (
-	KindTextDelta     EventKind = "text_delta"
+	KindTextDelta      EventKind = "text_delta"
-	KindToolCallDelta EventKind = "tool_call_delta"
+	KindReasoningDelta EventKind = "reasoning_delta"
-	KindMessageEnd    EventKind = "message_end"
+	KindToolCallDelta  EventKind = "tool_call_delta"
-	KindError         EventKind = "error"
+	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

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:

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
 

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
+}

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)
+		}
+	}
+}

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() {
+					go func() {
-			defer wg.Done()
+						defer wg.Done()
-			defer func() {
+						defer func() {
-				if rec := recover(); rec != nil {
+							if rec := recover(); rec != nil {
-					mu.Lock()
+								mu.Lock()
-					if groupErr == nil {
+								if groupErr == nil {
-						groupErr = fmt.Errorf("parallel step panic: %v", rec)
+									groupErr = fmt.Errorf("parallel step panic: %v", rec)
-					}
+								}
-					mu.Unlock()
+								mu.Unlock()
-				}
+							}
-			}()
+						}()
-			if err := d.runOneStep(runCtx, wsDir, sCopy, backend, timeout, opts.StepNotify, &mu, &producedFiles, ledger); err != nil {
+						if err := d.runOneStep(runCtx, wsDir, sCopy, backend, timeout, opts.StepNotify, &mu, &producedFiles, ledger); err != nil {
-				mu.Lock()
+							mu.Lock()
-				if groupErr == nil {
+							if groupErr == nil {
-					groupErr = err
+								groupErr = err
-				}
+							}
-				mu.Unlock()
+							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 重跑场景直接放行）
+			// 治理：受控迁移至执行态（legacy 重跑场景直接放行）
-		markStepExecuting(wf, step.ID)
+			markStepExecuting(wf, step.ID)
-		if err := d.runOneStep(runCtx, wsDir, step, backend, timeout, opts.StepNotify, &mu, &producedFiles, ledger); err != nil {
+			if err := d.runOneStep(runCtx, wsDir, step, backend, timeout, opts.StepNotify, &mu, &producedFiles, ledger); err != nil {
-			// A5 非关键步骤降级：记录 degraded 继续推进，收口统一标 degraded
+				// A5 非关键步骤降级：记录 degraded 继续推进，收口统一标 degraded
-			if !step.IsCritical() {
+				if !step.IsCritical() {
-				markStepDegraded(wsDir, wf, step, err)
+					markStepDegraded(wsDir, wf, step, err)
-				stepsRun++
+					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
+		res        *agent.Result
-		execErr     error
+		execErr    error
-		matErr      error
+		matErr     error
-		duration    int64
+		duration   int64
-		ok          bool
+		ok         bool
-		lastAssess  agent.FailureAssessment
+		lastAssess agent.FailureAssessment
-		basePrompt  = string(taskData)
+		basePrompt = string(taskData)
 	)
+	defer stopProcessBridge()
 	for attempt := 1; attempt <= agent.MaxTransientStepInvocations; attempt++ {
 		onDisk, _ := os.ReadFile(prevOutPath)
 		wroteThisInvocation := deliverableWroteThisInvocation(prevContent, onDisk)

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"`             // 详情标题
+	Title       string `json:"title"`
-	Content     string `json:"content"`           // 完整推演正文或技术草稿
+	Content     string `json:"content"`     // 完整推演正文或技术草稿
-	Collapsible bool   `json:"collapsible"`       // 客户端是否应当默认折叠显示
+	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轨正式合规文档交付载荷

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