From 2cd52da85dfd8229b468fccdc65ca73737177769 Mon Sep 17 00:00:00 2001 From: agent-bot Date: Tue, 29 Sep 2026 05:51:32 +0000 Subject: [PATCH] refactor(squad): typed ProgressSink port replaces direct stream dependency KTD1 second step (behavior preserved): - squad defines ProgressSink (EmitPhase/Detail/Deliverable/Summary/Briefing) with its own DetailEvent/ToolEvent types; zero internal/stream imports left - bizdev assembly maps port to SSE streamer (squad_progress_adapter) and injects via ctx at the four squad entry points - process_bridge tests rewritten as port-contract tests (captureSink) - verified: build+vet clean, squad/bizdev/archtest green --- internal/bizdev/service.go | 4 +- internal/bizdev/service_task_run.go | 2 +- internal/bizdev/squad_progress_adapter.go | 65 ++++++++++++ internal/squad/process_bridge.go | 45 ++++----- internal/squad/process_bridge_test.go | 118 ++++++++++++---------- internal/squad/progress.go | 66 ++++++++++++ internal/squad/quality_gate.go | 9 +- internal/squad/service.go | 35 ++++--- internal/squad/workflow.go | 52 +++++----- 9 files changed, 268 insertions(+), 128 deletions(-) create mode 100644 internal/bizdev/squad_progress_adapter.go create mode 100644 internal/squad/progress.go diff --git a/internal/bizdev/service.go b/internal/bizdev/service.go index baf8a216..93c1c14b 100644 --- a/internal/bizdev/service.go +++ b/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) } diff --git a/internal/bizdev/service_task_run.go b/internal/bizdev/service_task_run.go index 21bbc563..c225423e 100644 --- a/internal/bizdev/service_task_run.go +++ b/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{ diff --git a/internal/bizdev/squad_progress_adapter.go b/internal/bizdev/squad_progress_adapter.go new file mode 100644 index 00000000..6a017353 --- /dev/null +++ b/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}) +} diff --git a/internal/squad/process_bridge.go b/internal/squad/process_bridge.go index 9945b78c..f03cfbfa 100644 --- a/internal/squad/process_bridge.go +++ b/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 { - if streamer == nil { +func newProcessBridge(sink ProgressSink, step StepSpec, friendlyRole string) *processBridge { + 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) { - if b == nil || b.streamer == nil { +func (b *processBridge) emitDetail(p *DetailEvent) { + 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()) { - if streamer == nil || req == nil { +func AttachProcessBridge(sink ProgressSink, step StepSpec, req *agent.Request) (stop func()) { + 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() {} } diff --git a/internal/squad/process_bridge_test.go b/internal/squad/process_bridge_test.go index e0c1fdcf..f693a729 100644 --- a/internal/squad/process_bridge_test.go +++ b/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 - var events []stream.StreamEvent - streamer := stream.NewStreamer(true, func(ev stream.StreamEvent) { - mu.Lock() - events = append(events, ev) - mu.Unlock() - }) + sink, events, mu := newCaptureSink() 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 { - if ev.Type != stream.EventDetail || ev.Detail == nil { - continue + for _, ev := range *events { + if ev.Category != detailCategoryAgentThought { + t.Fatalf("unexpected category %s", ev.Category) } - if ev.Detail.Category != detailCategoryAgentThought { - t.Fatalf("unexpected category %s", ev.Detail.Category) - } - if ev.Detail.Delta != "" { + if ev.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 { - t.Fatalf("step identity missing: %+v", ev.Detail) + if ev.StepID != step.ID || ev.Role != step.Role { + 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 - var events []stream.StreamEvent - streamer := stream.NewStreamer(true, func(ev stream.StreamEvent) { - mu.Lock() - events = append(events, ev) - mu.Unlock() - }) + sink, events, mu := newCaptureSink() 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 { - if ev.Type != stream.EventDetail || ev.Detail == nil || ev.Detail.Category != detailCategoryToolCall { + for _, ev := range *events { + 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 - 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"}, "测试") + sink, events, mu := newCaptureSink() + b := newProcessBridge(sink, 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 { - if ev.Detail != nil && ev.Detail.Tool != nil && ev.Detail.Tool.Status == "failed" { + for _, ev := range *events { + 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) { - 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"}, "市场专员") + sink := &captureSink{mu: &mu} + sink.events = &[]DetailEvent{} + capture := func(p DetailEvent) { + sink.mu.Lock() + *sink.events = append(*sink.events, p) + sink.mu.Unlock() + } + _ = capture + b1 := newProcessBridge(sink, StepSpec{ID: "p1", Role: "regulatory"}, "法规专员") + 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) diff --git a/internal/squad/progress.go b/internal/squad/progress.go new file mode 100644 index 00000000..1e8f7fa7 --- /dev/null +++ b/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 +} diff --git a/internal/squad/quality_gate.go b/internal/squad/quality_gate.go index 0858143e..a8cb1f2e 100644 --- a/internal/squad/quality_gate.go +++ b/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)}) diff --git a/internal/squad/service.go b/internal/squad/service.go index 8b18de70..d70b7bc4 100644 --- a/internal/squad/service.go +++ b/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) - if streamer != nil { - streamer.EmitPhase("squad_assembly", fmt.Sprintf("正在组建并启动【%s】智能体小队（后端: %s）...", TaskArtifactTitle(req.TaskKey), backendName), "leader", 0, 0, 5) + sink := ProgressFromContext(ctx) + if sink != nil { + 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 { - streamer.EmitPhase("task_failed", fmt.Sprintf("任务执行异常: %v", err), "leader", 0, 0, 0) + if sink != nil { + 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 { - streamer.EmitPhase("quality_gate", "审核主管正在核验交付物规范并执行质量门禁把关...", "leader", 0, 0, 85) + if sink != nil { + 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 { - streamer.EmitPhase("quality_gate", fmt.Sprintf("质量门禁审查告警: %v", err), "leader", 0, 0, 90) + if sink != nil { + 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) - streamer.EmitSummary(TaskArtifactTitle(req.TaskKey)+"已生成", advisorBrief.MaturityHeadline, advisorBrief.MaturityScore, advisorBrief.MaturityLevel) - streamer.EmitBriefing(advisorBrief) + sink.EmitPhase("advisor_briefing", "贴身业务参谋正在提炼关键决策摘要与风险雷达...", "coordinator", 0, 0, 98) + sink.EmitSummary(TaskArtifactTitle(req.TaskKey)+"已生成", advisorBrief.MaturityHeadline, advisorBrief.MaturityScore, advisorBrief.MaturityLevel) + sink.EmitBriefing(advisorBrief) } } diff --git a/internal/squad/workflow.go b/internal/squad/workflow.go index 4e12d6cf..b6f17852 100644 --- a/internal/squad/workflow.go +++ b/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 { - streamer.EmitPhase("step_executing", fmt.Sprintf("%s正在进行：%s...", roleTitle, step.Title), step.Role, 0, 0, 0) + if sink != nil { + 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 { - streamer.EmitPhase( + if attempt > 1 && sink != nil { + 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 { - streamer.EmitPhase("step_failed", fmt.Sprintf("%s执行异常：%s", roleTitle, execErr.Error()), step.Role, 0, 0, 0) + if sink != nil { + 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 { - streamer.EmitPhase("step_failed", fmt.Sprintf("%s执行未成功：%s", roleTitle, errMsg), step.Role, 0, 0, 0) + if sink != nil { + 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 { - streamer.EmitPhase("step_completed", fmt.Sprintf("%s完成任务：%s (耗时 %dms)", roleTitle, step.Title, duration), step.Role, 0, 0, 0) + if sink != nil { + 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) - if streamer != nil { - streamer.EmitPhase("dynamic_adjustment", fmt.Sprintf("单步评估发现缺口，动态追加任务步骤：%s (%s)", addStep.Title, FriendlyRoleName(addStep.Role)), addStep.Role, 0, 0, 0) + sink := ProgressFromContext(ctx) + if sink != nil { + sink.EmitPhase("dynamic_adjustment", fmt.Sprintf("单步评估发现缺口，动态追加任务步骤：%s (%s)", addStep.Title, FriendlyRoleName(addStep.Role)), addStep.Role, 0, 0, 0) } } _ = d.SaveWorkflow(wsDir, wf) -- 2.54.0