internal/bizdev/service.go

@@ -330,10 +330,10 @@ func (s *Service) ListStages() []StageMeta {
 }
 
 func (s *Service) InspectWorkspace(ctx context.Context, workDir string) (*squad.WorkspaceState, error) {
-	return s.squad.InspectWorkspace(ctx, workDir)
+	return s.squad.InspectWorkspace(withSquadProgress(ctx), workDir)
 }
 
 // ResumeOrModifyWorkspace 动态调整任务并继续执行
 func (s *Service) ResumeOrModifyWorkspace(ctx context.Context, req squad.ModifyWorkflowRequest) (*squad.SquadRunResult, error) {
-	return s.squad.ResumeOrModifyWorkspace(ctx, req)
+	return s.squad.ResumeOrModifyWorkspace(withSquadProgress(ctx), req)
 }

internal/bizdev/service_task_run.go

@@ -706,7 +706,7 @@ func (s *Service) ExecuteTask(ctx context.Context, req TaskRunRequest) (*TaskRun
 		RunID:           req.RunID,
 	}
 
-	res, err := s.squad.ExecuteTask(ctx, squadReq)
+	res, err := s.squad.ExecuteTask(withSquadProgress(ctx), squadReq)
 	if err != nil {
 		if res != nil {
 			out := &TaskRunResult{

internal/bizdev/squad_progress_adapter.go

@@ -0,0 +1,65 @@
+package bizdev
+
+import (
+	"context"
+
+	"chat-agent-system/internal/squad"
+	"chat-agent-system/internal/stream"
+)
+
+// squadProgressAdapter 把小队过程端口映射到产品 SSE 流（KTD1 装配侧）。
+// 小队不感知 stream 包；SSE 事件形态、持久化与重放由产品侧全权负责。
+type squadProgressAdapter struct {
+	streamer *stream.Streamer
+}
+
+func (a *squadProgressAdapter) EmitPhase(phaseID, message, role string, stepIndex, totalSteps, progress int) {
+	a.streamer.EmitPhase(phaseID, message, role, stepIndex, totalSteps, progress)
+}
+
+func (a *squadProgressAdapter) EmitDetail(p squad.DetailEvent) {
+	var tool *stream.ToolCallInfo
+	if p.Tool != nil {
+		tool = &stream.ToolCallInfo{
+			Name:       p.Tool.Name,
+			Status:     p.Tool.Status,
+			Input:      p.Tool.Input,
+			Output:     p.Tool.Output,
+			DurationMs: p.Tool.DurationMs,
+		}
+	}
+	a.streamer.Emit(stream.StreamEvent{Type: stream.EventDetail, Detail: &stream.DetailPayload{
+		Category:    p.Category,
+		StepID:      p.StepID,
+		Role:        p.Role,
+		ItemID:      p.ItemID,
+		Title:       p.Title,
+		Content:     p.Content,
+		Collapsible: p.Collapsible,
+		Delta:       p.Delta,
+		Done:        p.Done,
+		Tool:        tool,
+	}})
+}
+
+func (a *squadProgressAdapter) EmitDeliverable(taskKey, title, filePath string, wordCount int, score float64) {
+	a.streamer.EmitDeliverable(taskKey, title, filePath, wordCount, score)
+}
+
+func (a *squadProgressAdapter) EmitSummary(title, content string, maturityScore int, maturityLevel string) {
+	a.streamer.EmitSummary(title, content, maturityScore, maturityLevel)
+}
+
+func (a *squadProgressAdapter) EmitBriefing(brief any) {
+	a.streamer.EmitBriefing(brief)
+}
+
+// withSquadProgress 在调用小队前装配过程端口：ctx 已有 Streamer 时注入适配器，
+// 否则原样返回（小队对缺失端口全程 no-op，行为与旧版 streamer==nil 分支一致）。
+func withSquadProgress(ctx context.Context) context.Context {
+	st := stream.FromContext(ctx)
+	if st == nil {
+		return ctx
+	}
+	return squad.WithProgress(ctx, &squadProgressAdapter{streamer: st})
+}

internal/squad/process_bridge.go

@@ -7,7 +7,6 @@ import (
 	"time"
 
 	"chat-agent-system/internal/agent"
-	"chat-agent-system/internal/stream"
 )
 
 // 过程流常量：节流窗口 / 单帧上限 / 工具文本截断（设计文档 F.1）。
@@ -64,7 +63,7 @@ func (d *deltaStream) finish() {
 // processBridge 把 agent.ProcessEvent 翻译为 stream.EventDetail 增量帧。
 // 每个 step（小队成员的一次执行）独立持有一个 bridge；无跨 step 共享状态，并行安全。
 type processBridge struct {
-	streamer  *stream.Streamer
+	sink      ProgressSink
 	step      StepSpec
 	friendly  string // 角色显示名
 	flushTick *time.Ticker
@@ -95,14 +94,14 @@ type toolBlock struct {
 	input   strings.Builder
 }
 
-// newProcessBridge 创建过程桥。返回 nil 表示该 step 无流式需求（streamer 缺失），
+// newProcessBridge 创建过程桥。返回 nil 表示该 step 无流式需求（sink 缺失），
 // 调用方对 nil bridge 调用任何方法都是安全 no-op。
-func newProcessBridge(streamer *stream.Streamer, step StepSpec, friendlyRole string) *processBridge {
+func newProcessBridge(sink ProgressSink, step StepSpec, friendlyRole string) *processBridge {
-	if streamer == nil {
+	if sink == nil {
 		return nil
 	}
 	b := &processBridge{
-		streamer: streamer,
+		sink:      sink,
 		step:     step,
 		friendly: friendlyRole,
 		stopCh:   make(chan struct{}),
@@ -233,12 +232,12 @@ func (b *processBridge) onToolStart(name string) {
 	itemID, toolName := cur.itemID, cur.name
 	b.mu.Unlock()
 	// tool_execution_start 是正式的"开始调用"信号：无论是否预建，都要发 calling 帧。
-	b.emitDetail(&stream.DetailPayload{
+	b.emitDetail(&DetailEvent{
 		Category:    detailCategoryToolCall,
 		StepID:      b.step.ID,
 		Role:        b.step.Role,
 		ItemID:      itemID,
-		Tool:        &stream.ToolCallInfo{Name: toolName, Status: "calling"},
+		Tool:        &ToolEvent{Name: toolName, Status: "calling"},
 		Title:       fmt.Sprintf("%s调用工具 %s", b.friendly, toolName),
 		Collapsible: true,
 	})
@@ -257,14 +256,14 @@ func (b *processBridge) onToolEnd(name, result string) {
 		status = "failed"
 	}
 	duration := time.Since(t.started).Milliseconds()
-	info := &stream.ToolCallInfo{
+	info := &ToolEvent{
 		Name:       name,
 		Status:     status,
 		Input:      trimProcessText(t.input.String()),
 		Output:     trimProcessText(result),
 		DurationMs: duration,
 	}
-	b.emitDetail(&stream.DetailPayload{
+	b.emitDetail(&DetailEvent{
 		Category:    detailCategoryToolCall,
 		StepID:      b.step.ID,
 		Role:        b.step.Role,
@@ -273,7 +272,7 @@ func (b *processBridge) onToolEnd(name, result string) {
 		Collapsible: true,
 	})
 	// 工具项即时定稿（状态帧 + done 标记合并语义：终帧即最后状态）
-	b.emitDetail(&stream.DetailPayload{
+	b.emitDetail(&DetailEvent{
 		Category:    detailCategoryToolCall,
 		StepID:      b.step.ID,
 		Role:        b.step.Role,
@@ -312,7 +311,7 @@ func (b *processBridge) flushPending() {
 	if len(text) > processMaxFrameBytes {
 		text = text[:processMaxFrameBytes] + "\n…(截断)"
 	}
-	b.emitDetail(&stream.DetailPayload{
+	b.emitDetail(&DetailEvent{
 		Category:    detailCategoryAgentThought,
 		StepID:      b.step.ID,
 		Role:        b.step.Role,
@@ -326,7 +325,7 @@ func (b *processBridge) finishThought(th *thoughtBlock) {
 	th.stream.finish()
 	// 冲残余
 	if text := th.stream.take(); text != "" {
-		b.emitDetail(&stream.DetailPayload{
+		b.emitDetail(&DetailEvent{
 			Category:    detailCategoryAgentThought,
 			StepID:      b.step.ID,
 			Role:        b.step.Role,
@@ -335,7 +334,7 @@ func (b *processBridge) finishThought(th *thoughtBlock) {
 			Collapsible: true,
 		})
 	}
-	b.emitDetail(&stream.DetailPayload{
+	b.emitDetail(&DetailEvent{
 		Category:    detailCategoryAgentThought,
 		StepID:      b.step.ID,
 		Role:        b.step.Role,
@@ -347,12 +346,12 @@ func (b *processBridge) finishThought(th *thoughtBlock) {
 }
 
 func (b *processBridge) finishTool(t *toolBlock, status, result string) {
-	b.emitDetail(&stream.DetailPayload{
+	b.emitDetail(&DetailEvent{
 		Category: detailCategoryToolCall,
 		StepID:   b.step.ID,
 		Role:     b.step.Role,
 		ItemID:   t.itemID,
-		Tool: &stream.ToolCallInfo{
+		Tool: &ToolEvent{
 			Name:       t.name,
 			Status:     status,
 			Input:      trimProcessText(t.input.String()),
@@ -364,11 +363,11 @@ func (b *processBridge) finishTool(t *toolBlock, status, result string) {
 	})
 }
 
-func (b *processBridge) emitDetail(p *stream.DetailPayload) {
+func (b *processBridge) emitDetail(p *DetailEvent) {
-	if b == nil || b.streamer == nil {
+	if b == nil || b.sink == nil {
 		return
 	}
-	b.streamer.Emit(stream.StreamEvent{Type: stream.EventDetail, Detail: p})
+	b.sink.EmitDetail(*p)
 }
 
 // trimProcessText 工具入参/结果截断（防超大帧；保留首尾）。
@@ -383,13 +382,13 @@ func trimProcessText(s string) string {
 	return head + "\n…(中间省略 " + fmt.Sprint(len(s)-headN-tailN) + " 字节)…\n" + tail
 }
 
-// AttachProcessBridge 给 agent.Request 装配过程桥（若 streamer 缺失或无流式需求则原样返回）。
+// AttachProcessBridge 给 agent.Request 装配过程桥（若 sink 缺失或无流式需求则原样返回）。
 // 返回的 stop 函数在 Execute 返回后必须调用一次（定稿与冲刷残余）。
-func AttachProcessBridge(streamer *stream.Streamer, step StepSpec, req *agent.Request) (stop func()) {
+func AttachProcessBridge(sink ProgressSink, step StepSpec, req *agent.Request) (stop func()) {
-	if streamer == nil || req == nil {
+	if sink == nil || req == nil {
 		return func() {}
 	}
-	bridge := newProcessBridge(streamer, step, FriendlyRoleName(step.Role))
+	bridge := newProcessBridge(sink, step, FriendlyRoleName(step.Role))
 	if bridge == nil {
 		return func() {}
 	}

internal/squad/process_bridge_test.go

@@ -7,31 +7,48 @@ import (
 	"time"
 
 	"chat-agent-system/internal/agent"
-	"chat-agent-system/internal/stream"
 )
 
+// captureSink 端口契约测试辅助：并发安全地捕获 DetailEvent。
+type captureSink struct {
+	mu     *sync.Mutex
+	events *[]DetailEvent
+}
+
+func (c *captureSink) EmitPhase(phaseID, message, role string, stepIndex, totalSteps, progress int) {}
+
+func (c *captureSink) EmitDetail(p DetailEvent) {
+	c.mu.Lock()
+	*c.events = append(*c.events, p)
+	c.mu.Unlock()
+}
+
+func (c *captureSink) EmitDeliverable(taskKey, title, filePath string, wordCount int, score float64) {}
+func (c *captureSink) EmitSummary(title, content string, maturityScore int, maturityLevel string)   {}
+func (c *captureSink) EmitBriefing(brief any)                                                       {}
+
+func newCaptureSink() (*captureSink, *[]DetailEvent, *sync.Mutex) {
+	var mu sync.Mutex
+	var events []DetailEvent
+	return &captureSink{mu: &mu, events: &events}, &events, &mu
+}
+
 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")
+		t.Fatal("nil sink must not attach OnEvent")
 	}
 	req.OnEvent = func(agent.ProcessEvent) {} // 直接赋值不应有 panic
 	stop()                                    // no-op stop 不 panic
 }
 
 func TestProcessBridgeReasoningFlow(t *testing.T) {
-	var mu sync.Mutex
+	sink, events, mu := newCaptureSink()
-	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)
+	stop := AttachProcessBridge(sink, step, req)
 	defer stop()
 
 	req.OnEvent(agent.ProcessEvent{Type: "reasoning_delta", Text: "第一段思考"})
@@ -45,22 +62,19 @@ func TestProcessBridgeReasoningFlow(t *testing.T) {
 	defer mu.Unlock()
 	var deltas, dones int
 	var joined strings.Builder
-	for _, ev := range events {
+	for _, ev := range *events {
-		if ev.Type != stream.EventDetail || ev.Detail == nil {
+		if ev.Category != detailCategoryAgentThought {
-			continue
+			t.Fatalf("unexpected category %s", ev.Category)
 		}
-		if ev.Detail.Category != detailCategoryAgentThought {
+		if ev.Delta != "" {
-			t.Fatalf("unexpected category %s", ev.Detail.Category)
-		}
-		if ev.Detail.Delta != "" {
 			deltas++
-			joined.WriteString(ev.Detail.Delta)
+			joined.WriteString(ev.Delta)
 		}
-		if ev.Detail.Done {
+		if ev.Done {
 			dones++
 		}
-		if ev.Detail.StepID != step.ID || ev.Detail.Role != step.Role {
+		if ev.StepID != step.ID || ev.Role != step.Role {
-			t.Fatalf("step identity missing: %+v", ev.Detail)
+			t.Fatalf("step identity missing: %+v", ev)
 		}
 	}
 	if joined.String() != "第一段思考继续推理" {
@@ -75,15 +89,9 @@ func TestProcessBridgeReasoningFlow(t *testing.T) {
 }
 
 func TestProcessBridgeToolFlow(t *testing.T) {
-	var mu sync.Mutex
+	sink, events, mu := newCaptureSink()
-	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, "撰写专员")
+	b := newProcessBridge(sink, step, "撰写专员")
 	if b == nil {
 		t.Fatal("bridge expected")
 	}
@@ -97,20 +105,20 @@ func TestProcessBridgeToolFlow(t *testing.T) {
 	mu.Lock()
 	defer mu.Unlock()
 	var calling, done, withInput int
-	for _, ev := range events {
+	for _, ev := range *events {
-		if ev.Type != stream.EventDetail || ev.Detail == nil || ev.Detail.Category != detailCategoryToolCall {
+		if ev.Category != detailCategoryToolCall {
 			continue
 		}
-		if ev.Detail.Tool == nil {
+		if ev.Tool == nil {
 			t.Fatal("tool_call frame must carry Tool info")
 		}
 		switch {
-		case ev.Detail.Tool.Status == "calling":
+		case ev.Tool.Status == "calling":
 			calling++
-		case ev.Detail.Done:
+		case ev.Done:
 			done++
 		}
-		if strings.Contains(ev.Detail.Tool.Input, "a.md") {
+		if strings.Contains(ev.Tool.Input, "a.md") {
 			withInput++
 		}
 	}
@@ -126,14 +134,8 @@ func TestProcessBridgeToolFlow(t *testing.T) {
 }
 
 func TestProcessBridgeToolFailureStatus(t *testing.T) {
-	var mu sync.Mutex
+	sink, events, mu := newCaptureSink()
-	var events []stream.StreamEvent
+	b := newProcessBridge(sink, StepSpec{ID: "s", Role: "r"}, "测试")
-	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()
@@ -141,8 +143,8 @@ func TestProcessBridgeToolFailureStatus(t *testing.T) {
 	mu.Lock()
 	defer mu.Unlock()
 	sawFailed := false
-	for _, ev := range events {
+	for _, ev := range *events {
-		if ev.Detail != nil && ev.Detail.Tool != nil && ev.Detail.Tool.Status == "failed" {
+		if ev.Tool != nil && ev.Tool.Status == "failed" {
 			sawFailed = true
 		}
 	}
@@ -170,16 +172,16 @@ 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) {
+	sink := &captureSink{mu: &mu}
-		if ev.Type != stream.EventDetail || ev.Detail == nil {
+	sink.events = &[]DetailEvent{}
-			return
+	capture := func(p DetailEvent) {
-		}
+		sink.mu.Lock()
-		mu.Lock()
+		*sink.events = append(*sink.events, p)
-		seen[ev.Detail.ItemID] = ev.Detail.Role
+		sink.mu.Unlock()
-		mu.Unlock()
+	}
-	})
+	_ = capture
-	b1 := newProcessBridge(streamer, StepSpec{ID: "p1", Role: "regulatory"}, "法规专员")
+	b1 := newProcessBridge(sink, StepSpec{ID: "p1", Role: "regulatory"}, "法规专员")
-	b2 := newProcessBridge(streamer, StepSpec{ID: "p2", Role: "market"}, "市场专员")
+	b2 := newProcessBridge(sink, StepSpec{ID: "p2", Role: "market"}, "市场专员")
 	var wg sync.WaitGroup
 	wg.Add(2)
 	go func() {
@@ -198,8 +200,12 @@ func TestProcessBridgeConcurrentSteps(t *testing.T) {
 	}()
 	wg.Wait()
 	time.Sleep(50 * time.Millisecond) // 等冲刷
+	// 从捕获事件推导 itemID -> role 映射
 	mu.Lock()
-	defer mu.Unlock()
+	for _, ev := range *sink.events {
+		seen[ev.ItemID] = ev.Role
+	}
+	mu.Unlock()
 	for itemID, role := range seen {
 		if !strings.Contains(itemID, "p1") && !strings.Contains(itemID, "p2") {
 			t.Fatalf("unexpected item %s", itemID)

internal/squad/progress.go

@@ -0,0 +1,66 @@
+package squad
+
+import "context"
+
+// ProgressSink 是小队过程事件的输出端口（KTD1：SSE 映射、task_runs 持久事件、
+// 权限与重放由产品侧实现；小队只产出有类型的进度/步骤/完成事件）。
+//
+// 实现方约定：方法不得阻塞（消费方自行异步入队）；nil 安全由调用侧判断，
+// ProgressFromContext 未注入时返回 nil，小队内部全部 no-op。
+type ProgressSink interface {
+	// EmitPhase 报告阶段进展（装配、门禁、失败等里程碑）。
+	EmitPhase(phaseID, message, role string, stepIndex, totalSteps, progress int)
+	// EmitDetail 报告过程细节帧（思考增量、工具调用、步骤输出）。
+	EmitDetail(p DetailEvent)
+	// EmitDeliverable 报告正式交付物完成。
+	EmitDeliverable(taskKey, title, filePath string, wordCount int, score float64)
+	// EmitSummary 报告任务摘要（含成熟度）。
+	EmitSummary(title, content string, maturityScore int, maturityLevel string)
+	// EmitBriefing 报告参谋简报载荷（结构由产品侧定义）。
+	EmitBriefing(brief any)
+}
+
+// DetailEvent 是小队过程细节事件的唯一载荷形态；与产品 SSE detail 帧的字段
+// 对应关系由产品侧适配器维护（squad 不 import stream 包）。
+type DetailEvent struct {
+	Category    string // agent_thought | step_output | tool_call | log
+	StepID      string
+	Role        string
+	ItemID      string // 聚合项：一个思考块/一次工具调用
+	Title       string
+	Content     string // 完整快照（对账帧）
+	Collapsible bool
+	Delta       string // 增量追加到 ItemID 项
+	Done        bool   // 定稿标记
+	Tool        *ToolEvent
+}
+
+// ToolEvent 是工具调用状态帧（Category == tool_call 时）。
+type ToolEvent struct {
+	Name       string // write_file / read_file / ...
+	Status     string // calling | succeeded | failed
+	Input      string // 入参摘要（截断）
+	Output     string // 结果摘要（截断）
+	DurationMs int64
+}
+
+type progressCtxKey struct{}
+
+// WithProgress 向 ctx 注入小队过程端口（产品装配点调用）。
+func WithProgress(ctx context.Context, sink ProgressSink) context.Context {
+	if sink == nil {
+		return ctx
+	}
+	return context.WithValue(ctx, progressCtxKey{}, sink)
+}
+
+// ProgressFromContext 提取小队过程端口；未注入返回 nil（调用侧 no-op）。
+func ProgressFromContext(ctx context.Context) ProgressSink {
+	if ctx == nil {
+		return nil
+	}
+	if sink, ok := ctx.Value(progressCtxKey{}).(ProgressSink); ok {
+		return sink
+	}
+	return nil
+}

internal/squad/quality_gate.go

@@ -15,7 +15,6 @@ import (
 	"unicode/utf8"
 
 	"chat-agent-system/internal/agent"
-	"chat-agent-system/internal/stream"
 	"chat-agent-system/internal/workflowdef"
 )
 
@@ -601,7 +600,7 @@ func (q *QualityGate) EnsureQualityOrRepair(
 		// 治理：修复轮预算预扣（超限硬停，杜绝机械空转烧成本）
 		if lErr := ledger.ReserveRepair(); lErr != nil {
 			_ = ledger.Save(wsDir)
-			if st := stream.FromContext(ctx); st != nil {
+			if st := ProgressFromContext(ctx); st != nil {
 				st.EmitPhase("governance", "修复预算硬停: "+lErr.Error(), "taskgov", 0, 0, 0)
 			}
 			break
@@ -656,12 +655,12 @@ func (q *QualityGate) EnsureQualityOrRepair(
 					"error":   rerr.Error(),
 					"mutated": strings.Join(mutated, ","),
 				})
-				if st := stream.FromContext(ctx); st != nil {
+				if st := ProgressFromContext(ctx); st != nil {
 					st.EmitPhase("governance", "修复 containment 恢复失败: "+rerr.Error(), "taskgov", 0, 0, 0)
 				}
 				return report, fmt.Errorf("restore parent inputs after containment: %w", rerr)
 			}
-			if st := stream.FromContext(ctx); st != nil {
+			if st := ProgressFromContext(ctx); st != nil {
 				st.EmitPhase("governance", "修复 containment 违规: 父级输入已被越界改写，已恢复运行前字节，本轮作废", "taskgov", 0, 0, 0)
 			}
 		}
@@ -680,7 +679,7 @@ func (q *QualityGate) EnsureQualityOrRepair(
 		sim := OutputSimilarity(prevRepairContent, string(repairedBytes))
 		if sim >= 0.98 {
 			if stallHinted {
-				if st := stream.FromContext(ctx); st != nil {
+				if st := ProgressFromContext(ctx); st != nil {
 					st.EmitPhase("governance", fmt.Sprintf("修复停滞止损: 连续两轮修复产物相似度 %.1f%%，终止机械重试", sim*100), "taskgov", 0, 0, 0)
 				}
 				appendGovAudit(wsDir, "repair_stall_break", map[string]string{"attempt": fmt.Sprintf("%d", attempt), "similarity": fmt.Sprintf("%.3f", sim)})

internal/squad/service.go

@@ -11,7 +11,6 @@ import (
 
 	"chat-agent-system/internal/agent"
 	"chat-agent-system/internal/quote"
-	"chat-agent-system/internal/stream"
 	"chat-agent-system/internal/workflowdef"
 )
 
@@ -276,9 +275,9 @@ func (s *Service) ExecuteTask(ctx context.Context, req SquadRunRequest) (*SquadR
 	}
 
 	// 3. 执行动态工作流调度
-	streamer := stream.FromContext(ctx)
+	sink := ProgressFromContext(ctx)
-	if streamer != nil {
+	if sink != nil {
-		streamer.EmitPhase("squad_assembly", fmt.Sprintf("正在组建并启动【%s】智能体小队（后端: %s）...", TaskArtifactTitle(req.TaskKey), backendName), "leader", 0, 0, 5)
+		sink.EmitPhase("squad_assembly", fmt.Sprintf("正在组建并启动【%s】智能体小队（后端: %s）...", TaskArtifactTitle(req.TaskKey), backendName), "leader", 0, 0, 5)
 	}
 
 	if s.opts.NewBackend == nil {
@@ -296,8 +295,8 @@ func (s *Service) ExecuteTask(ctx context.Context, req SquadRunRequest) (*SquadR
 
 	execRes, err := s.driver.Execute(ctx, workDir, backend, execOpts)
 	if err != nil {
-		if streamer != nil {
+		if sink != nil {
-			streamer.EmitPhase("task_failed", fmt.Sprintf("任务执行异常: %v", err), "leader", 0, 0, 0)
+			sink.EmitPhase("task_failed", fmt.Sprintf("任务执行异常: %v", err), "leader", 0, 0, 0)
 		}
 		return &SquadRunResult{
 			TaskKey:       req.TaskKey,
@@ -311,26 +310,26 @@ func (s *Service) ExecuteTask(ctx context.Context, req SquadRunRequest) (*SquadR
 	}
 
 	// 4. 自动化质量门禁与修复 (QualityGate)
-	if streamer != nil {
+	if sink != nil {
-		streamer.EmitPhase("quality_gate", "审核主管正在核验交付物规范并执行质量门禁把关...", "leader", 0, 0, 85)
+		sink.EmitPhase("quality_gate", "审核主管正在核验交付物规范并执行质量门禁把关...", "leader", 0, 0, 85)
 	}
 	maxRetries := repairBudgetForSquadTask(req)
 	report, err := s.gate.EnsureQualityOrRepair(ctx, workDir, req.TaskKey, s.driver, backend, maxRetries, execOpts)
 	if err != nil {
 		// 记录门禁错误但容错返回当前正文
 		report = &QualityReport{Passed: false, Violations: []string{err.Error()}}
-		if streamer != nil {
+		if sink != nil {
-			streamer.EmitPhase("quality_gate", fmt.Sprintf("质量门禁审查告警: %v", err), "leader", 0, 0, 90)
+			sink.EmitPhase("quality_gate", fmt.Sprintf("质量门禁审查告警: %v", err), "leader", 0, 0, 90)
 		}
-	} else if streamer != nil {
+	} else if sink != nil {
 		if report.Passed {
-			streamer.EmitPhase("quality_gate", fmt.Sprintf("质量门禁审核通过（评分: %.1f / 100）", report.Score), "leader", 0, 0, 95)
+			sink.EmitPhase("quality_gate", fmt.Sprintf("质量门禁审核通过（评分: %.1f / 100）", report.Score), "leader", 0, 0, 95)
 		} else {
 			violations := strings.Join(report.Violations, "；")
 			if violations == "" {
 				violations = "未满足质量门禁要求"
 			}
-			streamer.EmitPhase("quality_gate", fmt.Sprintf("质量门禁未通过（评分: %.1f / 100）：%s", report.Score, violations), "leader", 0, 0, 95)
+			sink.EmitPhase("quality_gate", fmt.Sprintf("质量门禁未通过（评分: %.1f / 100）：%s", report.Score, violations), "leader", 0, 0, 95)
 		}
 	}
 
@@ -381,16 +380,16 @@ func (s *Service) ExecuteTask(ctx context.Context, req SquadRunRequest) (*SquadR
 		}
 	}
 
-	if streamer != nil {
+	if sink != nil {
 		score := 0.0
 		if report != nil {
 			score = report.Score
 		}
-		streamer.EmitDeliverable(req.TaskKey, TaskArtifactTitle(req.TaskKey), finalOutPath, len(strings.Fields(string(finalBytes))), score)
+		sink.EmitDeliverable(req.TaskKey, TaskArtifactTitle(req.TaskKey), finalOutPath, len(strings.Fields(string(finalBytes))), score)
 		if advisorBrief != nil {
-			streamer.EmitPhase("advisor_briefing", "贴身业务参谋正在提炼关键决策摘要与风险雷达...", "coordinator", 0, 0, 98)
+			sink.EmitPhase("advisor_briefing", "贴身业务参谋正在提炼关键决策摘要与风险雷达...", "coordinator", 0, 0, 98)
-			streamer.EmitSummary(TaskArtifactTitle(req.TaskKey)+"已生成", advisorBrief.MaturityHeadline, advisorBrief.MaturityScore, advisorBrief.MaturityLevel)
+			sink.EmitSummary(TaskArtifactTitle(req.TaskKey)+"已生成", advisorBrief.MaturityHeadline, advisorBrief.MaturityScore, advisorBrief.MaturityLevel)
-			streamer.EmitBriefing(advisorBrief)
+			sink.EmitBriefing(advisorBrief)
 		}
 	}
 

internal/squad/workflow.go

@@ -13,7 +13,6 @@ import (
 	"time"
 
 	"chat-agent-system/internal/agent"
-	"chat-agent-system/internal/stream"
 )
 
 // StepEvent 表示小队单步执行生命周期的事件
@@ -244,7 +243,7 @@ func (d *WorkflowDriver) Execute(ctx context.Context, wsDir string, backend agen
 				for range parallelGroup {
 					if rErr := ledger.ReserveStep(); rErr != nil {
 						_ = ledger.Save(wsDir)
-						if st := stream.FromContext(ctx); st != nil {
+						if st := ProgressFromContext(ctx); st != nil {
 							st.EmitPhase("governance", "预算硬停: "+rErr.Error(), "taskgov", 0, 0, 0)
 						}
 						wf.Status = "failed"
@@ -315,7 +314,7 @@ func (d *WorkflowDriver) Execute(ctx context.Context, wsDir string, backend agen
 					}
 				}
 				if msg := ledger.CheckSoftLimit(); msg != "" {
-					if st := stream.FromContext(ctx); st != nil {
+					if st := ProgressFromContext(ctx); st != nil {
 						st.EmitPhase("governance", msg, "taskgov", 0, 0, 0)
 					}
 				}
@@ -327,7 +326,7 @@ func (d *WorkflowDriver) Execute(ctx context.Context, wsDir string, backend agen
 			// 治理：预算预扣（超限硬停）
 			if rErr := ledger.ReserveStep(); rErr != nil {
 				_ = ledger.Save(wsDir)
-				if st := stream.FromContext(ctx); st != nil {
+				if st := ProgressFromContext(ctx); st != nil {
 					st.EmitPhase("governance", "预算硬停: "+rErr.Error(), "taskgov", 0, 0, 0)
 				}
 				wf.Status = "failed"
@@ -368,7 +367,7 @@ func (d *WorkflowDriver) Execute(ctx context.Context, wsDir string, backend agen
 				}
 			}
 			if msg := ledger.CheckSoftLimit(); msg != "" {
-				if st := stream.FromContext(ctx); st != nil {
+				if st := ProgressFromContext(ctx); st != nil {
 					st.EmitPhase("governance", msg, "taskgov", 0, 0, 0)
 				}
 			}
@@ -491,10 +490,10 @@ func (d *WorkflowDriver) runOneStep(
 			Status: "started",
 		})
 	}
-	streamer := stream.FromContext(ctx)
+	sink := ProgressFromContext(ctx)
 	roleTitle := FriendlyRoleName(step.Role)
-	if streamer != nil {
+	if sink != nil {
-		streamer.EmitPhase("step_executing", fmt.Sprintf("%s正在进行：%s...", roleTitle, step.Title), step.Role, 0, 0, 0)
+		sink.EmitPhase("step_executing", fmt.Sprintf("%s正在进行：%s...", roleTitle, step.Title), step.Role, 0, 0, 0)
 	}
 	stepStart := time.Now()
 
@@ -527,7 +526,7 @@ func (d *WorkflowDriver) runOneStep(
 		mu.Unlock()
 	}
 	if verdict := DetectRepetition(histEntries, step.ID); verdict.Detected {
-		if st := stream.FromContext(ctx); st != nil {
+		if st := ProgressFromContext(ctx); st != nil {
 			st.EmitPhase("governance", "重复识别强制终止: "+verdict.Reason, "taskgov", 0, 0, 0)
 		}
 		appendGovAudit(wsDir, "loop_aborted", map[string]string{"step_id": step.ID, "kind": verdict.Kind, "reason": verdict.Reason})
@@ -551,9 +550,9 @@ func (d *WorkflowDriver) runOneStep(
 		Targets:      targets,
 		Timeout:      stepTimeout,
 	}
-	// 过程流桥：把成员执行的思考/工具调用增量翻译为 detail 事件（streamer 缺失时为 no-op）。
+	// 过程流桥：把成员执行的思考/工具调用增量翻译为 detail 事件（sink 缺失时为 no-op）。
 	// 注意 retry 循环内多次 Execute 共用一个 bridge 语义（同 step 聚合），Stop 只在首次调用。
-	stopProcessBridge := AttachProcessBridge(streamer, step, &req)
+	stopProcessBridge := AttachProcessBridge(sink, step, &req)
 
 	var (
 		res        *agent.Result
@@ -577,8 +576,8 @@ func (d *WorkflowDriver) runOneStep(
 			req.Prompt = basePrompt + "\n\n【纠偏重试】你上一轮没有调用 write 工具写 " + deliverableName +
 				"，本轮必须先用 write 工具把完整交付物写入当前目录 " + deliverableName + "，再结束。"
 		}
-		if attempt > 1 && streamer != nil {
+		if attempt > 1 && sink != nil {
-			streamer.EmitPhase(
+			sink.EmitPhase(
 				"step_retrying",
 				fmt.Sprintf("%s遇瞬时失败，发起全新步骤调用（%d/%d）", roleTitle, attempt, agent.MaxTransientStepInvocations),
 				step.Role, 0, 0, 0,
@@ -700,8 +699,8 @@ func (d *WorkflowDriver) runOneStep(
 	}
 
 	if !ok && execErr != nil {
-		if streamer != nil {
+		if sink != nil {
-			streamer.EmitPhase("step_failed", fmt.Sprintf("%s执行异常：%s", roleTitle, execErr.Error()), step.Role, 0, 0, 0)
+			sink.EmitPhase("step_failed", fmt.Sprintf("%s执行异常：%s", roleTitle, execErr.Error()), step.Role, 0, 0, 0)
 		}
 		if notify != nil {
 			notify(StepEvent{
@@ -719,8 +718,8 @@ func (d *WorkflowDriver) runOneStep(
 
 	if !ok && res != nil && !res.Success {
 		errMsg := res.ErrorReason
-		if streamer != nil {
+		if sink != nil {
-			streamer.EmitPhase("step_failed", fmt.Sprintf("%s执行未成功：%s", roleTitle, errMsg), step.Role, 0, 0, 0)
+			sink.EmitPhase("step_failed", fmt.Sprintf("%s执行未成功：%s", roleTitle, errMsg), step.Role, 0, 0, 0)
 		}
 		if notify != nil {
 			notify(StepEvent{
@@ -777,10 +776,17 @@ func (d *WorkflowDriver) runOneStep(
 	*producedFiles = append(*producedFiles, filepath.ToSlash(filepath.Join(step.SubWorkspace, deliverableName)))
 	mu.Unlock()
 
-	if streamer != nil {
+	if sink != nil {
-		streamer.EmitPhase("step_completed", fmt.Sprintf("%s完成任务：%s (耗时 %dms)", roleTitle, step.Title, duration), step.Role, 0, 0, 0)
+		sink.EmitPhase("step_completed", fmt.Sprintf("%s完成任务：%s (耗时 %dms)", roleTitle, step.Title, duration), step.Role, 0, 0, 0)
 		if stepData, err := os.ReadFile(targetPath); err == nil && len(stepData) > 0 {
-			streamer.EmitDetail("step_output", step.ID, step.Role, fmt.Sprintf("%s阶段性产出: %s", roleTitle, step.Title), string(stepData), true)
+			sink.EmitDetail(DetailEvent{
+				Category:    "step_output",
+				StepID:      step.ID,
+				Role:        step.Role,
+				Title:       fmt.Sprintf("%s阶段性产出: %s", roleTitle, step.Title),
+				Content:     string(stepData),
+				Collapsible: true,
+			})
 		}
 	}
 
@@ -976,9 +982,9 @@ func (d *WorkflowDriver) applyDynamicAdjustment(
 			newSteps = append(newSteps, wf.Steps[insertIdx:]...)
 			wf.Steps = newSteps
 
-			streamer := stream.FromContext(ctx)
+			sink := ProgressFromContext(ctx)
-			if streamer != nil {
+			if sink != nil {
-				streamer.EmitPhase("dynamic_adjustment", fmt.Sprintf("单步评估发现缺口，动态追加任务步骤：%s (%s)", addStep.Title, FriendlyRoleName(addStep.Role)), addStep.Role, 0, 0, 0)
+				sink.EmitPhase("dynamic_adjustment", fmt.Sprintf("单步评估发现缺口，动态追加任务步骤：%s (%s)", addStep.Title, FriendlyRoleName(addStep.Role)), addStep.Role, 0, 0, 0)
 			}
 		}
 		_ = d.SaveWorkflow(wsDir, wf)