package squad import ( "context" "encoding/json" "fmt" "os" "path/filepath" "strings" "time" ) // leaderExecRequest 构造 leader 轮次的执行请求:与标准步骤同等装配 // (过程事件桥、ToolProfile/Skills/工具白名单、执行标识与预算)—— // 修平此前 leader 轮丢弃 OnEvent 与 ToolProfile 的缺口。 func leaderExecRequest(drv *flowDriver, st *RunState, leader RoleSpec, workDir, prompt, instructions string, targets []string, timeout time.Duration) ExecRequest { stepID := fmt.Sprintf("leader_turn_%d", st.Leader.Turns) var budget *ExecBudget if timeout > 0 { budget = &ExecBudget{Deadline: time.Now().Add(timeout)} } return ExecRequest{ WorkDir: workDir, Prompt: prompt, Instructions: instructions, Targets: targets, Timeout: timeout, ToolProfile: orDefault(leader.ToolProfile, ToolProfileWorkspace), ToolAllowlist: leader.Tools, Skills: leader.Skills, SessionScope: st.Meta.ID + "/" + stepID, ExecutionID: st.Meta.ID + "-" + stepID, Budget: budget, OnEvent: func(ev ProcessEvent) { drv.emit(st, Event{Type: EvDetail, StepID: stepID, Role: leader.Role, Text: ev.Text, Data: marshalJSON(map[string]string{"process_type": ev.Type, "tool": ev.Tool, "detail": ev.Detail})}) }, } } // leader-delegation 模式(机制参考 Multica Squad Leader 协议,原创实现): // leader 是注册表中的普通角色,被引擎在每轮激活——读目标、按成员 skills // 挑人、产出一份结构化委派决策(delegation.json),引擎按决策执行成员步骤、 // 把成员产出并回黑板、再唤醒 leader 评估下一步。leader 只协调不实现; // 委派不了就升级问人;轮次有上限防失控。 // DefaultMaxLeaderTurns leader 模式每运行的 leader 轮次上限。 const DefaultMaxLeaderTurns = 8 // DefaultMaxReplans leader 卡顿触发的重规划上限(Q2;与轮次上限独立)。 const DefaultMaxReplans = 3 // DefaultStuckThreshold 连续无进展轮数达到该值触发一次重规划(Q2)。 const DefaultStuckThreshold = 2 // LeaderState leader 循环状态(workspace leader.json 持久化,重启可恢复)。 type LeaderState struct { Turns int `json:"turns"` MaxTurns int `json:"max_turns"` LeaderRole string `json:"leader_role"` Delegations []DelegationRecord `json:"delegations"` // 进度账本(Q2):每轮委派后记一条进展评估;卡顿检测与重规划的凭据。 Progress []ProgressEntry `json:"progress,omitempty"` // StuckTurns 连续无进展轮数(有进展即清零)。 StuckTurns int `json:"stuck_turns,omitempty"` // Replans 已触发的重规划次数;MaxReplans 触顶降级收口。 Replans int `json:"replans,omitempty"` MaxReplans int `json:"max_replans,omitempty"` // ReplanNote 一次性重规划指令(并入下一轮 leader 任务书后清空)。 ReplanNote string `json:"replan_note,omitempty"` } // ProgressEntry 是进度账本的一条评估记录。 type ProgressEntry struct { Turn int `json:"turn"` Signal string `json:"signal"` // progress | stuck Detail string `json:"detail,omitempty"` } // DelegationRecord 一次已执行的委派(历史注入下一轮 leader 上下文)。 type DelegationRecord struct { Turn int `json:"turn"` Member string `json:"member"` Title string `json:"title"` Targets []string `json:"targets,omitempty"` Status string `json:"status"` // completed / failed / degraded Error string `json:"error,omitempty"` } // delegation leader 单轮产出的结构化决策。 type delegation struct { Action string `json:"action"` // delegate | escalate | close | failed Member string `json:"member,omitempty"` Title string `json:"title,omitempty"` Targets []string `json:"targets,omitempty"` Skills []string `json:"skills,omitempty"` MinScore int `json:"min_score,omitempty"` Briefing string `json:"briefing,omitempty"` // 给成员的补充指令 Question string `json:"question,omitempty"` // escalate 时的问题 Reason string `json:"reason,omitempty"` } const leaderSubWorkspace = "tasks/00_leader" func (st *RunState) loadLeader() { if st.Leader != nil { return } st.Leader = &LeaderState{MaxTurns: DefaultMaxLeaderTurns} data, err := st.storeOr().LoadLeader(st.Meta.WorkDir) if err == nil && data != nil { _ = json.Unmarshal(data, st.Leader) } } func (st *RunState) saveLeader() { if st.Leader == nil { return } if data, err := json.Marshal(st.Leader); err == nil { _ = st.storeOr().SaveLeader(st.Meta.WorkDir, data) } } // leaderPrompt 组装 leader 的系统指令(协调契约 + 通用守则)。 func leaderPrompt(members []RoleSpec, catalogMd string) string { var b strings.Builder b.WriteString(`你是本小队的 LEADER。你被激活的唯一职责是协调,不是自己实现。 每轮你必须做且只做一件事:评估当前进展,然后输出一份委派决策文件 delegation.json 到本子目录(用工具写入,不要输出到对话里): {"action":"delegate","member":"成员role","title":"给成员的子任务标题","targets":["output.md"],"skills":["可选,覆盖该成员本步技能"],"min_score":0,"briefing":"给成员的补充指令/验收标准","reason":"为什么派给这个人"} {"action":"escalate","question":"要向人提出的问题","reason":"为什么需要人拍板"} {"action":"close","reason":"目标已达成,收口"} {"action":"failed","reason":"无法达成的原因"} 硬规则: 1. 按成员的能力(skills/职责)挑人,不按头衔;没有合适成员就 escalate 说明缺口,而不是自己动手或乱派。 2. 一轮只委派一个子任务;子任务要小到可以一次完成并产出明确文件。 3. 委派的子任务标题里写清验收标准;不要复述目标全文。 4. 交付受阻时如实 escalate/failed,严禁假装完成或声称"基本完成"。 5. 所有成员子任务完成且交付物满足目标才 close;close 不等于人工终审。 6. 缺失的信息若会实质改变结果、执行路径、权限或安全,就 escalate 问人;否则自己决定并在 reason 里说明你决定了什么。 成员交付核实:每个成员完成后,其交付文件会自动快照到本目录 inbox/<步骤id>/ 下——请用工具查看该目录核实交付,不要凭空假设文件是否存在。 可用成员(role — 名称 — 技能 — 职责): `) for _, m := range members { fmt.Fprintf(&b, "- %s — %s — skills=%v — %v\n", m.Role, m.Name, m.Skills, m.Responsibilities) } if catalogMd != "" { b.WriteString("\n系统能力目录(供你判断哪些子任务不必派成员、以及该向用户收集什么信息;缺失必填信息时 escalate 并使用提问建议):\n") b.WriteString(catalogMd) } return b.String() } // leaderTaskBrief 组装 leader 本轮的 TASK.md(目标 + 黑板 + 委派历史)。 func leaderTaskBrief(st *RunState, mail []MailMessage) string { var b strings.Builder b.WriteString("# Leader 轮次:评估并委派\n\n") fmt.Fprintf(&b, "## 运行目标\n%s\n\n", st.Meta.Goal.Text) if len(st.Clarifications) > 0 { b.WriteString("## 已澄清问答\n") for _, c := range st.Clarifications { fmt.Fprintf(&b, "- %s\n", c) } b.WriteString("\n") } if len(st.BB.Fields) > 0 { b.WriteString("## 黑板字段\n") for k, v := range st.BB.Fields { fmt.Fprintf(&b, "- %s = %s\n", k, v) } b.WriteString("\n") } if gaps := st.BB.UnresolvedBlockingGaps(); len(gaps) > 0 { b.WriteString("## 未解决缺口\n") for _, g := range gaps { fmt.Fprintf(&b, "- [%s] %s\n", g.ID, g.Item) } b.WriteString("\n") } if len(st.Leader.Delegations) > 0 { b.WriteString("## 委派历史\n") for _, d := range st.Leader.Delegations { fmt.Fprintf(&b, "- 第%d轮 → %s:%s(%s)\n", d.Turn, d.Member, d.Title, d.Status) if d.Error != "" { fmt.Fprintf(&b, " - 错误:%s\n", d.Error) } } b.WriteString("\n") } else { b.WriteString("(尚无委派,这是第一轮。)\n\n") } if len(st.Leader.Progress) > 0 { b.WriteString("## 进度账本\n") for _, p := range st.Leader.Progress { fmt.Fprintf(&b, "- 第%d轮 [%s] %s\n", p.Turn, p.Signal, p.Detail) } fmt.Fprintf(&b, "- 已重规划 %d/%d 次\n\n", st.Leader.Replans, st.Leader.MaxReplans) } if st.Leader.ReplanNote != "" { fmt.Fprintf(&b, "%s\n\n", st.Leader.ReplanNote) } if section := formatMailSection(mail); section != "" { b.WriteString(strings.TrimPrefix(section, "\n\n")) b.WriteString("\n") } b.WriteString("## 本轮产出要求\n\n写 delegation.json(结构见你的指令)到本子目录。\n") return b.String() } // runLeaderLoop leader-delegation 主循环:委派→执行→唤醒,直到 // close/failed/上限/挂起(escalation/gap/权限等由 HITL 处理后重入本函数)。 func (e *SquadEngine) runLeaderLoop(ctx context.Context, st *RunState) (*RunResult, *HitlGate, error) { st.loadLeader() drv := &flowDriver{opts: e.deps} leader, ok := findAgent(st.Plan.Squad.Agents, st.Leader.LeaderRole) if !ok { if spec, exists := e.opts.Registry.Get(st.Leader.LeaderRole); exists { leader = spec } else { return &RunResult{RunID: st.Meta.ID, Status: RunFailed, Plan: planPtr(st), Error: "leader role not found: " + st.Leader.LeaderRole, WorkDir: st.Meta.WorkDir}, nil, nil } } var members []RoleSpec for _, a := range st.Plan.Squad.Agents { if a.Role != leader.Role { members = append(members, a) } } if len(members) == 0 { return &RunResult{RunID: st.Meta.ID, Status: RunFailed, Plan: planPtr(st), Error: "leader mode requires at least one member role besides the leader", WorkDir: st.Meta.WorkDir}, nil, nil } subDir := filepath.Join(st.Meta.WorkDir, filepath.FromSlash(leaderSubWorkspace)) if err := os.MkdirAll(subDir, 0o755); err != nil { return nil, nil, err } _ = snapshotContextInto(st.Meta.WorkDir, subDir) term := e.terminationOf(st) for st.Leader.Turns < term.MaxLeaderTurns { // HITL:leader 角色权限升级(与标准模式同语义)。 if leader.BashPermission == "ask" && !st.Approved["role:"+leader.Role] { if drv.policy(st).mode() != HitlInteractive { return &RunResult{RunID: st.Meta.ID, Status: RunFailed, Plan: planPtr(st), Error: "leader role requires bash approval but hitl mode is " + string(drv.policy(st).mode()), WorkDir: st.Meta.WorkDir}, nil, nil } req := &HumanRequest{ ID: newRequestID(), RunID: st.Meta.ID, Kind: HitlPermission, Severity: SeverityActionRequired, Question: fmt.Sprintf("leader 角色 %s 声明 bash_permission=ask,是否授权其在本次运行中使用更高权限?", leader.Name), Options: []HumanOption{ {Value: "approve", Label: "授权本次运行"}, {Value: "reject", Label: "拒绝"}, {Value: "abort", Label: "中止"}, }, CreatedAt: time.Now(), Status: "pending", } gate := &HitlGate{Request: *req, PrevStatus: RunExecuting} if err := st.saveGate(gate); err == nil { st.Meta.Status = string(RunWaitingHuman) _ = st.save() drv.emitHuman(st, req) } return heldResult(st, gate), gate, nil } // 每轮重写 leader 上下文(黑板/委派历史已变化)。 // 邮箱消息与重规划指令是一次性的:并入本轮任务书后清空并落盘。 st.mu.Lock() leaderMail := takeMailFor(st, st.Leader.LeaderRole) st.mu.Unlock() if err := os.WriteFile(filepath.Join(subDir, "TASK.md"), []byte(leaderTaskBrief(st, leaderMail)), 0o644); err != nil { return nil, nil, err } if len(leaderMail) > 0 { _ = st.save() } if st.Leader.ReplanNote != "" { st.Leader.ReplanNote = "" st.saveLeader() } if err := os.WriteFile(filepath.Join(subDir, "AGENTS.md"), []byte(leaderPrompt(members, e.opts.IntentCatalog.intentCatalogMd())), 0o644); err != nil { return nil, nil, err } st.Leader.Turns++ st.saveLeader() drv.emit(st, Event{Type: EvStepStarted, StepID: fmt.Sprintf("leader_turn_%d", st.Leader.Turns), Role: leader.Role, Text: "leader 评估并委派"}) backend, backendName, err := e.deps.Backends.Resolve(leader, st.Meta.Backend) if err != nil { return e.leaderFail(st, err.Error()), nil, nil } prompt := readFileOr(filepath.Join(subDir, "TASK.md"), leaderTaskBrief(st, nil)) prompt = st.injectSteering(prompt) instructions := readFileOr(filepath.Join(subDir, "AGENTS.md"), leaderPrompt(members, e.opts.IntentCatalog.intentCatalogMd())) result, execErr := backend.Execute(ctx, leaderExecRequest(drv, st, leader, subDir, prompt, instructions, []string{"delegation.json"}, e.deps.StepTimeout)) if execErr != nil || result == nil || !result.Success { reason := "leader execution failed" if execErr != nil { reason = execErr.Error() } else if result != nil && result.ErrorReason != "" { reason = result.ErrorReason } _ = os.Remove(filepath.Join(subDir, "delegation.json")) return e.leaderFail(st, fmt.Sprintf("leader turn %d: %s (backend=%s)", st.Leader.Turns, reason, backendName)), nil, nil } dec, perr := parseDelegation(subDir) if perr != nil { return e.leaderFail(st, fmt.Sprintf("leader turn %d: %v", st.Leader.Turns, perr)), nil, nil } _ = os.Remove(filepath.Join(subDir, "delegation.json")) // 每轮只消费一次 switch dec.Action { case "close": drv.emit(st, Event{Type: EvStepCompleted, StepID: "leader_close", Role: leader.Role, Text: dec.Reason}) st.saveLeader() out, gate, err := drv.finalize(ctx, st) if err != nil { return nil, nil, err } e.finalizeEvent(out) return out, gate, nil case "failed": return e.leaderFail(st, "leader 报告无法达成:"+dec.Reason), nil, nil case "escalate": if drv.policy(st).mode() != HitlInteractive { return e.leaderFail(st, "leader 请求升级问人,但 hitl mode="+string(drv.policy(st).mode())), nil, nil } req := &HumanRequest{ ID: newRequestID(), RunID: st.Meta.ID, Kind: HitlEscalation, Severity: SeverityActionRequired, StepID: fmt.Sprintf("leader_turn_%d", st.Leader.Turns), Question: firstNonEmpty(dec.Question, "leader 需要人工拍板:") + "(原因:" + dec.Reason + ")", Options: []HumanOption{ {Value: "continue", Label: "已在回答中给出决定,leader 继续"}, {Value: "abort", Label: "中止运行"}, }, CreatedAt: time.Now(), Status: "pending", } gate := &HitlGate{Request: *req, PrevStatus: RunExecuting} if err := st.saveGate(gate); err == nil { st.Meta.Status = string(RunWaitingHuman) _ = st.save() drv.emitHuman(st, req) } return heldResult(st, gate), gate, nil case "delegate": member, ok := findAgent(members, dec.Member) if !ok { return e.leaderFail(st, fmt.Sprintf("leader 委派了编制外成员 %q", dec.Member)), nil, nil } step := StepSpec{ ID: fmt.Sprintf("m_%02d_%s", st.Leader.Turns, member.Role), Role: member.Role, Title: firstNonEmpty(dec.Title, "子任务"), SubWorkspace: fmt.Sprintf("tasks/%02d_%s", st.Leader.Turns, member.Role), Status: "pending", } if len(dec.Targets) > 0 { step.Targets = dec.Targets } else { step.Targets = []string{"output.md"} } if len(dec.Skills) > 0 { step.Skills = dec.Skills } if dec.MinScore > 0 { step.MinScore = dec.MinScore } st.Plan.Workflow.Steps = append(st.Plan.Workflow.Steps, step) _ = st.save() // 构建成员子工作区并执行(复用标准驱动器单步执行)。 idx := len(st.Plan.Workflow.Steps) - 1 if err := e.ws.BuildSubWorkspaces(st.Meta.WorkDir, &st.Plan, &st.BB); err != nil { return nil, nil, err } memberBrief := readFileOr(filepath.Join(st.Meta.WorkDir, filepath.FromSlash(step.SubWorkspace), "TASK.md"), step.Title) if strings.TrimSpace(dec.Briefing) != "" { memberBrief += "\n\n## Leader 补充指令\n" + dec.Briefing _ = os.WriteFile(filepath.Join(st.Meta.WorkDir, filepath.FromSlash(step.SubWorkspace), "TASK.md"), []byte(memberBrief), 0o644) } factsBefore := len(st.BB.Facts) out, gate, err := drv.runOneStep(ctx, st, idx) if err != nil { return nil, nil, err } rec := DelegationRecord{Turn: st.Leader.Turns, Member: member.Role, Title: step.Title, Targets: step.Targets, Status: "completed"} if out != nil { // 关键失败或挂起 rec.Status = "failed" rec.Error = firstNonEmpty(out.Error, string(out.Status)) st.Leader.Delegations = append(st.Leader.Delegations, rec) st.saveLeader() if out.Status == RunFailed { return out, nil, nil } return out, gate, nil // 挂起(gap/权限等),Respond 后重入循环 } st.Leader.Delegations = append(st.Leader.Delegations, rec) // 进度账本(Q2):委派完成且有新事实、或不是与上轮同题重复 → 有进展。 factsGrew := len(st.BB.Facts) > factsBefore sameAsPrev := false if n := len(st.Leader.Delegations); n >= 2 { prev := st.Leader.Delegations[n-2] sameAsPrev = prev.Member == rec.Member && prev.Title == rec.Title } if factsGrew || !sameAsPrev { st.Leader.StuckTurns = 0 st.Leader.Progress = append(st.Leader.Progress, ProgressEntry{ Turn: rec.Turn, Signal: "progress", Detail: fmt.Sprintf("%s:%s(新事实 %v)", rec.Member, rec.Title, factsGrew), }) } else { st.Leader.StuckTurns++ st.Leader.Progress = append(st.Leader.Progress, ProgressEntry{ Turn: rec.Turn, Signal: "stuck", Detail: fmt.Sprintf("与上轮同题重复且无新事实(连续 %d 轮)", st.Leader.StuckTurns), }) } // 卡顿 → 重规划;重规划触顶 → 降级收口(与轮次上限同款 Q23 语义)。 if st.Leader.StuckTurns >= DefaultStuckThreshold { st.Leader.StuckTurns = 0 st.Leader.Replans++ if st.Leader.Replans >= st.Leader.MaxReplans { st.saveLeader() drv.emit(st, Event{Type: EvPhase, Text: fmt.Sprintf( "重规划达上限 %d,按现有产出降级收口(Q2)", st.Leader.MaxReplans)}) out2, gate2, err2 := drv.finalize(ctx, st) if err2 != nil { return nil, nil, err2 } if gate2 != nil { return heldResult(st, gate2), gate2, nil } return out2, nil, nil } st.Leader.ReplanNote = fmt.Sprintf( "【重规划指令】连续 %d 轮无进展(同题重复且无新事实),已触发第 %d/%d 次重规划。"+ "本轮必须改变策略:换成员、把任务拆小、补充缺失输入,或 escalate 问人;不得原样重发上一次委派。", DefaultStuckThreshold, st.Leader.Replans, st.Leader.MaxReplans) drv.emit(st, Event{Type: EvPhase, Text: fmt.Sprintf( "卡顿检测:触发重规划 %d/%d", st.Leader.Replans, st.Leader.MaxReplans)}) } st.saveLeader() // 成员交付快照进 leader inbox(leader 工具沙箱看不到成员目录, // 须物化到 leader 工作区才能自查交付——InputLayoutSnapshot 的对称应用)。 if err := snapshotDeliverablesToLeader(st, step); err != nil { return nil, nil, err } // 成员产出后检查阻塞缺口(等待条件挂起在标准 gap 通道上)。 if g2 := drv.checkBlockingGaps(st); g2 != nil { return heldResult(st, g2), g2, nil } // 唤醒 leader:进入下一轮。 continue default: return e.leaderFail(st, fmt.Sprintf("leader 输出了未知 action %q", dec.Action)), nil, nil } } // Q23:达上限不再裸失败——按现有产出走 finalize 降级收口(质检/汇总路径与正常收口一致)。 drv2 := &flowDriver{opts: e.deps} drv2.emit(st, Event{Type: EvPhase, Text: fmt.Sprintf("leader 轮次达上限 %d,按现有产出降级收口(Q23)", term.MaxLeaderTurns)}) out, gate2, err := drv2.finalize(ctx, st) if err != nil { return nil, nil, err } if gate2 != nil { return heldResult(st, gate2), gate2, nil } return out, nil, nil } // leaderFail 统一的 leader 模式失败收口。 func (e *SquadEngine) leaderFail(st *RunState, reason string) *RunResult { now := time.Now() st.Meta.Status = string(RunFailed) st.Meta.CompletedAt = &now wf := st.Workflow() wf.Status = "failed" st.setWorkflow(wf) _ = st.save() e.deps.Sink.Emit(Event{Type: EvRunFailed, RunID: st.Meta.ID, Text: reason, At: time.Now()}) return &RunResult{RunID: st.Meta.ID, Status: RunFailed, Plan: planPtr(st), Error: reason, WorkDir: st.Meta.WorkDir, Usage: st.usageSnapshot()} } // snapshotDeliverablesToLeader 把成员交付文件复制进 leader 工作区 inbox/<步骤id>/。 func snapshotDeliverablesToLeader(st *RunState, step StepSpec) error { srcDir := filepath.Join(st.Meta.WorkDir, filepath.FromSlash(step.SubWorkspace)) dstDir := filepath.Join(st.Meta.WorkDir, filepath.FromSlash(leaderSubWorkspace), "inbox", step.ID) for _, t := range step.Targets { data, err := os.ReadFile(filepath.Join(srcDir, filepath.FromSlash(t))) if err != nil || len(data) == 0 { continue } dst := filepath.Join(dstDir, filepath.Base(filepath.FromSlash(t))) if err := AtomicWrite(dst, data); err != nil { return err } } return nil } // parseDelegation 解析 leader 产出的 delegation.json。 func parseDelegation(subDir string) (*delegation, error) { data, err := os.ReadFile(filepath.Join(subDir, "delegation.json")) if err != nil { return nil, fmt.Errorf("leader 未产出 delegation.json: %w", err) } dec := &delegation{} if err := json.Unmarshal(data, dec); err != nil { return nil, fmt.Errorf("delegation.json 解析失败: %w", err) } if dec.Action == "" { return nil, fmt.Errorf("delegation.json 缺少 action") } return dec, nil }