Agent-Squad/squad/workflow.go

package squad

import (
	"bytes"
	"context"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"error-code-system/errcode"
)

// flowDriver 工作流驱动器：推进 RunState 直到终态或 HITL 挂起边界。
type flowDriver struct {
	opts engineDeps
}

// advance 执行到边界；返回 (结果, 挂起上下文)，二者互斥非 nil（infra 错误时都为 nil 且 err 非 nil）。
// 推进模型（Q1 WP1/WP2）：每轮从权威 Steps 现场编译调度图，按 superstep
// 分层推进——层内无写集冲突的步骤并行（runParallel 复用），有冲突的按声明序
// 串行（含动态调整）；动态调整改 Steps 后下一轮自然生效。
func (d *flowDriver) advance(ctx context.Context, st *RunState) (res *RunResult, gate *HitlGate, err error) {
	wf := st.Workflow()
	wf.Status = "in_progress"
	st.setWorkflow(wf)
	d.emit(st, Event{Type: EvPhase, Text: "开始执行工作流"})

	superNo := 0
	// 运行级截止（Q2 long_running）：superstep 边界检查，超期未完步骤置
	// skipped 并降级收口；零值/不可解析 = 不设限。
	var runDeadline time.Time
	if v := st.Meta.Metadata["deadline"]; v != "" {
		// RFC3339Nano 同时接受秒级与纳秒级两种写法。
		if t, terr := time.Parse(time.RFC3339Nano, v); terr == nil {
			runDeadline = t
		}
	}
	cpw := newCheckpointWriter(CheckpointMode(st.Meta.Metadata["checkpoint_mode"]), st.storeOr())
	// 推进退出前 flush：async 落完、on_exit 写最终快照（Q1 WP4 三档的出口语义）。
	var lastGraph *graph
	defer func() {
		if lastGraph == nil {
			if g, gerr := compileGraph(st.Plan.Workflow.Steps); gerr == nil {
				lastGraph = g
			}
		}
		if lastGraph != nil {
			_ = cpw.flush(st.Meta.WorkDir, snapshotOf(superNo, lastGraph, st.Plan.Workflow.Steps))
		}
	}()
	for {
		if !runDeadline.IsZero() && time.Now().After(runDeadline) {
			d.emit(st, Event{Type: EvPhase, Text: "运行截止时刻已到，未完步骤按 skipped 收口"})
			st.mu.Lock()
			for i := range st.Plan.Workflow.Steps {
				if st.Plan.Workflow.Steps[i].Status == "pending" || st.Plan.Workflow.Steps[i].Status == "running" {
					st.Plan.Workflow.Steps[i].Status = "skipped"
					st.Plan.Workflow.Steps[i].Error = "deadline exceeded"
				}
			}
			_ = st.save()
			st.mu.Unlock()
			return d.finalize(ctx, st)
		}
		g, gerr := compileGraph(st.Plan.Workflow.Steps)
		if gerr != nil {
			wf2 := st.Workflow()
			wf2.Status = "failed"
			st.setWorkflow(wf2)
			return &RunResult{RunID: st.Meta.ID, Status: RunFailed, Plan: planPtr(st),
				Error: "graph: " + gerr.Error(), WorkDir: st.Meta.WorkDir}, nil, nil
		}
		layer, ok := g.nextLayer(st.Plan.Workflow.Steps)
		if !ok {
			if n := pendingCount(st.Plan.Workflow.Steps); n > 0 {
				wf2 := st.Workflow()
				wf2.Status = "failed"
				st.setWorkflow(wf2)
				return &RunResult{RunID: st.Meta.ID, Status: RunFailed, Plan: planPtr(st),
					Error: fmt.Sprintf("graph: 依赖无法满足，仍有 %d 个步骤无法调度", n), WorkDir: st.Meta.WorkDir}, nil, nil
			}
			break
		}
		superNo++
		d.emit(st, Event{Type: EvPhase, Text: fmt.Sprintf("superstep %d 推进（并行 %d 步，串行 %d 步）", superNo, len(layer.par), len(layer.seq))})

		// 并行段：组内每一步都先过审批和权限，任一需要挂起则整组不执行。
		if len(layer.par) > 0 {
			for _, idx := range layer.par {
				if out, g2, err := d.gateBeforeRun(ctx, st, idx); out != nil || g2 != nil || err != nil {
					if out == nil && g2 != nil {
						out = heldResult(st, g2)
					}
					return out, g2, err
				}
			}
			if out, g2, err := d.runParallel(ctx, st, layer.par); out != nil || g2 != nil || err != nil {
				if out == nil && g2 != nil {
					out = heldResult(st, g2)
				}
				return out, g2, err
			}
			if g2 := d.checkBlockingGaps(st); g2 != nil {
				return heldResult(st, g2), g2, nil
			}
			// 单成员并行组实际是顺序执行：与旧线性扫描一致，其后可动态调整
			// （多成员组不做，避免并发竞态）。
			if len(layer.par) == 1 {
				idx := layer.par[0]
				if st.Plan.Workflow.Steps[idx].AllowDynamicAdjustment && d.opts.Evaluator != nil {
					out, g2, err := d.applyDynamicAdjustment(ctx, st, idx)
					if out != nil || g2 != nil || err != nil {
						if out == nil && g2 != nil {
							out = heldResult(st, g2)
						}
						return out, g2, err
					}
				}
			}
		}

		// 串行段（写集冲突步骤，按声明序逐个执行；顺序步骤后可动态调整）。
		for _, idx := range layer.seq {
			if out, g2, err := d.gateBeforeRun(ctx, st, idx); out != nil || g2 != nil || err != nil {
				if out == nil && g2 != nil {
					out = heldResult(st, g2)
				}
				return out, g2, err
			}
			out, g2, err := d.runOneStep(ctx, st, idx)
			if out != nil || g2 != nil || err != nil {
				if out == nil && g2 != nil {
					out = heldResult(st, g2)
				}
				return out, g2, err
			}
			if g2 := d.checkBlockingGaps(st); g2 != nil {
				return heldResult(st, g2), g2, nil
			}
			if st.Plan.Workflow.Steps[idx].AllowDynamicAdjustment && d.opts.Evaluator != nil {
				out, g2, err := d.applyDynamicAdjustment(ctx, st, idx)
				if out != nil || g2 != nil || err != nil {
					if out == nil && g2 != nil {
						out = heldResult(st, g2)
					}
					return out, g2, err
				}
			}
		}

		// superstep 边界快照（sync 立即 / async 入队；on_exit 由 defer 的 flush 写）。
		lastGraph = g
		if err := cpw.boundary(st.Meta.WorkDir, snapshotOf(superNo, g, st.Plan.Workflow.Steps)); err != nil {
			return nil, nil, err
		}
	}

	return d.finalize(ctx, st)
}

// snapshotOf 组装一个 superstep 边界快照：序号、图结构指纹与全量步骤状态。
func snapshotOf(superNo int, g *graph, steps []StepSpec) Checkpoint {
	cp := Checkpoint{Superstep: superNo, At: time.Now(), GraphHash: g.hash()}
	for _, s := range steps {
		cp.StepStates = append(cp.StepStates, StepSnapshot{ID: s.ID, Status: s.Status, AttemptID: s.AttemptID})
	}
	return cp
}

// gateBeforeRun 在步骤真正执行前检查权限升级和显式审批。
// 返回非空结果或挂起时，调用方不得执行该步骤，也不得执行它所在的并行组。
func (d *flowDriver) gateBeforeRun(ctx context.Context, st *RunState, i int) (*RunResult, *HitlGate, error) {
	step := st.Plan.Workflow.Steps[i]
	role, _ := findAgent(st.Plan.Squad.Agents, step.Role)
	if role.BashPermission == BashPermissionAsk && !st.Approved["role:"+role.Role] {
		switch d.policyRole(st, role).mode() {
		case HitlDisabled, HitlAuto:
			step.Error = "permission denied: role " + role.Role + " requires bash approval (hitl mode=" + string(d.policyRole(st, role).mode()) + ")"
			st.Plan.Workflow.Steps[i] = step
			return d.failStep(ctx, st, i)
		default:
			return d.hold(st, HitlPermission, step.ID, step.Role,
				fmt.Sprintf("角色 %s（%s）声明 bash_permission=ask，是否授权其在本次运行中使用更高权限？", role.Name, role.Role),
				[]HumanOption{
					{Value: "approve", Label: "授权本次运行"},
					{Value: "reject", Label: "拒绝，步骤失败"},
					{Value: "abort", Label: "中止运行"},
				},
				fmt.Sprintf("步骤：%s\n角色：%s\n禁忌：%v", step.Title, role.Role, role.ForbiddenActions), "", nil)
		}
	}
	if step.RequiresApproval && !st.Approved["step:"+step.ID] {
		switch d.policyRole(st, role).mode() {
		case HitlDisabled, HitlAuto:
			step.Error = "approval required but hitl mode is " + string(d.policy(st).mode())
			st.Plan.Workflow.Steps[i] = step
			return d.failStep(ctx, st, i)
		default:
			return d.hold(st, HitlConfirmation, step.ID, step.Role,
				fmt.Sprintf("步骤「%s」标记为需人工审批后才能执行，是否批准？", step.Title),
				[]HumanOption{
					{Value: "approve", Label: "批准执行"},
					{Value: "skip", Label: "跳过该步骤"},
					{Value: "abort", Label: "中止运行"},
				},
				fmt.Sprintf("角色：%s\n产出：%v", step.Role, step.Targets), "step_approval", nil)
		}
	}
	return nil, nil, nil
}

// runParallel 并发执行一个并行组；关键步骤失败即终止整个运行。
func (d *flowDriver) runParallel(ctx context.Context, st *RunState, idxs []int) (*RunResult, *HitlGate, error) {
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()
	var wg sync.WaitGroup
	var mu sync.Mutex
	failedCritical := -1
	var firstFail *RunResult
	var gate *HitlGate
	for _, idx := range idxs {
		wg.Add(1)
		go func(idx int) {
			defer wg.Done()
			out, g, err := d.runOneStep(ctx, st, idx)
			mu.Lock()
			defer mu.Unlock()
			if g != nil && gate == nil {
				gate = g
				cancel()
			}
			if (out != nil && out.Status == RunFailed) || err != nil {
				cancel()
				// 保留首个带具体错误的失败结果（如 router 无命中）；
				// failStep 的通用格式只在没有更具体结果时兜底。
				if out != nil && out.Status == RunFailed && firstFail == nil {
					firstFail = out
				}
				if st.Plan.Workflow.Steps[idx].IsCritical() && failedCritical < 0 {
					failedCritical = idx
				}
			}
		}(idx)
	}
	wg.Wait()
	if gate != nil {
		return heldResult(st, gate), gate, nil
	}
	if firstFail != nil {
		return firstFail, nil, nil
	}
	if failedCritical >= 0 {
		return d.failStep(ctx, st, failedCritical)
	}
	return nil, nil, nil
}

// runOneStep 执行单个步骤（含瞬态重试与交付物校验）。
func (d *flowDriver) runOneStep(ctx context.Context, st *RunState, idx int) (*RunResult, *HitlGate, error) {
	if err := ctx.Err(); err != nil {
		return &RunResult{RunID: st.Meta.ID, Status: RunCancelled, Error: err.Error()}, nil, err
	}
	step := st.Plan.Workflow.Steps[idx]
	role, ok := findAgent(st.Plan.Squad.Agents, step.Role)
	if !ok {
		if r, exists := d.opts.Registry.Get(step.Role); exists {
			role = r
			st.mu.Lock()
			st.Plan.Squad.Agents = append(st.Plan.Squad.Agents, r)
			st.mu.Unlock()
		} else if step.IntentID != "" {
			role = RoleSpec{Role: step.Role, Name: step.Role}
		} else {
			step.Status, step.Error = "failed", "role not found: "+step.Role
			st.setStep(idx, step)
			return d.failStep(ctx, st, idx)
		}
	}
	// handoff 原语 fail-closed（Q2）：目标角色必须可解析，启动前拒绝。
	if step.Handoff != "" {
		if _, ok := findAgent(st.Plan.Squad.Agents, step.Handoff); !ok {
			if _, exists := d.opts.Registry.Get(step.Handoff); !exists {
				step.Status, step.Error = "failed", "handoff: 未知目标角色 "+step.Handoff
				st.setStep(idx, step)
				return d.failStep(ctx, st, idx)
			}
		}
	}
	subDir := filepath.Join(st.Meta.WorkDir, filepath.FromSlash(orDefault(step.SubWorkspace, fmt.Sprintf("tasks/%02d_%s", idx+1, step.Role))))
	if err := os.MkdirAll(subDir, 0o755); err != nil {
		return nil, nil, err
	}
	if step.SubWorkspace == "" {
		step.SubWorkspace = filepath.ToSlash(filepath.Join("tasks", fmt.Sprintf("%02d_%s", idx+1, step.Role)))
	}

	d.emit(st, Event{Type: EvStepStarted, StepID: step.ID, Role: step.Role, Text: step.Title})
	step.Status = "running"
	step.Attempts++
	step.AttemptID = fmt.Sprintf("%s/%s/%d", st.Meta.ID, step.ID, step.Attempts)
	step.Error = ""
	// 与收口落盘同一把锁：并行步骤一边改 Steps、一边全量 marshal 会数据竞争。
	st.mu.Lock()
	st.Plan.Workflow.Steps[idx] = step
	saveErr := st.save()
	st.mu.Unlock()
	if saveErr != nil {
		return nil, nil, saveErr
	}
	started := time.Now()

	if d.opts.Workspaces != nil {
		if err := d.opts.Workspaces.writeTaskFile(subDir, &step, role, &st.BB); err != nil {
			return nil, nil, err
		}
	}
	prompt := readFileOr(filepath.Join(subDir, "TASK.md"), step.Title)
	prompt = st.injectSteering(prompt)
	// 邮箱投递（Q2）：本步骤启动时把发给该角色（含广播）的未读消息并入
	// 任务书；只走 prompt 面，永不触碰 HITL 门（验收三构造性保证）。
	st.mu.Lock()
	mail := takeMailFor(st, step.Role)
	handoff := takeHandoffFor(st, step.Role)
	if len(mail) > 0 || handoff != nil {
		_ = st.save()
	}
	st.mu.Unlock()
	prompt += formatMailSection(mail)
	// handoff 原语（Q2）：目标角色的步骤接收过滤历史（只带移交方主产出）。
	prompt += formatHandoffSection(handoff)
	instructions := readFileOr(filepath.Join(subDir, "AGENTS.md"), d.opts.Workspaces.ResolveInstructions(role))

	var backend Executor
	backendName := ""
	if step.IntentID != "" {
		backendName = "intent:" + step.IntentID
	} else {
		var rerr error
		backend, backendName, rerr = d.opts.Backends.Resolve(role, st.Meta.Backend)
		if rerr != nil {
			step.Status, step.Error = "failed", rerr.Error()
			st.setStep(idx, step)
			return d.failStep(ctx, st, idx)
		}
	}

	targets := step.Targets
	if len(targets) == 0 {
		targets = []string{"output.md"}
	}
	// 步骤级共享预算：重试与首次尝试共用同一截止时刻
	//（agent-execution 标准：重试层共享总预算，不随次数翻倍）。
	var stepBudget *ExecBudget
	if d.opts.StepTimeout > 0 {
		stepBudget = &ExecBudget{Deadline: time.Now().Add(d.opts.StepTimeout)}
	}
	var result *ExecResult
	var err error
	for attempt := 1; attempt <= MaxTransientStepInvocations; attempt++ {
		reqPrompt := prompt
		if attempt > 1 {
			prevReason := "交付物缺失"
			if err != nil {
				prevReason = err.Error()
			} else if result != nil && result.ErrorReason != "" {
				prevReason = result.ErrorReason
			}
			reqPrompt += "\n\n【重试提示】上一次执行未成功（" + prevReason + "）。请务必把结果写入：" + strings.Join(targets, ", ")
		}
		req := ExecRequest{
			WorkDir:      subDir,
			Prompt:       reqPrompt,
			Instructions: instructions,
			Targets:      targets,
			Timeout:      d.opts.StepTimeout,
			ToolProfile:  orDefault(role.ToolProfile, ToolProfileWorkspace),
			OnEvent: func(ev ProcessEvent) {
				d.emit(st, Event{Type: EvDetail, StepID: step.ID, Role: step.Role, Text: ev.Text,
					Data: marshalJSON(map[string]string{"process_type": ev.Type, "tool": ev.Tool, "detail": ev.Detail})})
			},
			// RoleSpec v2 贯通：工具白名单 / 技能 / 标识（原为死字段或缺失）。
			ToolAllowlist: role.Tools,
			Skills:        firstNonEmptySlice(step.Skills, role.Skills),
			SessionScope:  st.Meta.ID + "/" + step.ID,
			ExecutionID:   step.AttemptID,
			Budget:        stepBudget,
		}
		if step.IntentID != "" {
			// 系统功能步：不经智能体，由 IntentRunner 直接执行；缺席即 fail-closed。
			// Runner 只产出内容（patches），由引擎统一物化到子工作区。
			if d.opts.IntentRunner == nil {
				step.Status, step.Error = "failed", fmt.Sprintf("intent step %s: no IntentRunner configured", step.ID)
				st.setStep(idx, step)
				return d.failStep(ctx, st, idx)
			}
			result, err = d.opts.IntentRunner.Run(ctx, IntentStep{
				IntentID: step.IntentID, Slots: step.IntentSlots,
				Title: step.Title, WorkDir: subDir, Targets: targets,
			})
			if err == nil && result != nil && len(result.Patches) > 0 {
				_, _ = MaterializeTargets(subDir, targets, result.Patches)
			}
		} else {
			result, err = backend.Execute(ctx, req)
		}
		if err == nil && result.Success {
			break
		}
		// 交付物物化：后端返回 patches 但未写盘时补写。
		if err == nil && len(result.Patches) > 0 {
			if _, merr := MaterializeTargets(subDir, targets, result.Patches); merr == nil {
				if allTargetsPresent(subDir, targets) {
					result.Success = true
					break
				}
			}
		}
		assessment := AssessStepFailure(StepFailureSignals{
			Err: err, Result: result, WorkDir: subDir,
			ExtraText:     fmt.Sprintf("step=%s backend=%s", step.ID, backendName),
			TargetMissing: !allTargetsPresent(subDir, targets),
		})
		if assessment.Class != FailureTransient || attempt == MaxTransientStepInvocations {
			step.Status = "failed"
			if err != nil {
				step.Error = err.Error()
			} else {
				step.Error = orDefault(assessment.Reason, result.ErrorReason)
			}
			break
		}
	}

	if ctx.Err() != nil {
		step.Status = "pending" // 取消后可恢复
		st.setStep(idx, step)
		return &RunResult{RunID: st.Meta.ID, Status: RunCancelled, Error: "context cancelled",
			Plan: planPtr(st), WorkDir: st.Meta.WorkDir}, nil, nil
	}

	if result != nil {
		st.addUsage(step.Role, result.PromptTokens, result.CompletionTokens)
	}

	step.DurationMs = time.Since(started).Milliseconds()
	if step.Status == "failed" {
		d.emit(st, Event{Type: EvStepFailed, StepID: step.ID, Role: step.Role, Text: step.Error, Code: string(errcode.SquadStepFailed)})
		st.setStep(idx, step)
		return d.failStep(ctx, st, idx)
	}

	// 交付物校验。
	missing := missingTargets(subDir, targets)
	if len(missing) > 0 {
		step.Status, step.Error = "failed", fmt.Sprintf("missing deliverables: %v", missing)
		d.emit(st, Event{Type: EvStepFailed, StepID: step.ID, Role: step.Role, Text: step.Error, Code: string(errcode.AgentMissingDeliverable)})
		st.setStep(idx, step)
		return d.failStep(ctx, st, idx)
	}

	step.Status = "completed"
	appendLog(subDir, fmt.Sprintf("[%s] backend=%s duration=%dms targets=%v\n", time.Now().Format(time.RFC3339), backendName, step.DurationMs, targets))

	// 状态回填 + 黑板合并 + 落盘同一临界区：并行步骤的共享状态变更串行化，
	// 防止 marshal 与写回竞争、临时文件互相踩踏把保存失败误判为步骤失败。
	st.mu.Lock()
	st.Plan.Workflow.Steps[idx] = step
	_ = mergeStepArtifacts(&st.BB, subDir)
	err = st.save()
	st.mu.Unlock()
	if err != nil {
		return nil, nil, err
	}
	d.emit(st, Event{Type: EvStepCompleted, StepID: step.ID, Role: step.Role, Text: step.Title,
		Data: marshalJSON(map[string]any{"duration_ms": step.DurationMs, "targets": targets})})

	// handoff 原语（Q2）：完成即移交最终答案所有权——记录 + 主产出快照，
	// 供目标角色下一步骤（过滤历史）与 finalize（链尾产出）消费。
	if step.Handoff != "" {
		d.recordHandoff(st, step, subDir, targets)
	}

	// 循环步再武装（Q2 loop）：停机条件不满足且预算未尽 → 本步与 loop.body
	// 一起重置为 pending；触顶发 loop_exhausted 事件后按完成继续。
	if step.Loop != nil {
		if d.loopShouldRearm(st, step) {
			if out, g2, err2 := d.rearmLoop(ctx, st, idx, step); out != nil || g2 != nil || err2 != nil {
				return out, g2, err2
			}
			return nil, nil, nil
		}
		if step.Loop.UntilField != "" && step.LoopRuns+1 >= step.Loop.MaxIterations {
			d.emit(st, Event{Type: EvPhase, StepID: step.ID, Text: fmt.Sprintf("loop_exhausted：%s 达 %d 轮上限，停机条件未满足", step.ID, step.Loop.MaxIterations)})
		}
	}
	// 路由步分支裁决（Q2 router）：首个命中的分支保留，其余分支目标置
	// skipped；无分支命中 fail-closed 成失败运行。
	if len(step.Branches) > 0 {
		return d.applyBranches(ctx, st, idx, step)
	}
	return nil, nil, nil
}

// loopShouldRearm 判定循环步是否再武装：停机条件（黑板字段等值）满足或
// 预算用尽则停。字段判据读的是本轮 mergeStepArtifacts 之后的黑板。
func (d *flowDriver) loopShouldRearm(st *RunState, step StepSpec) bool {
	// MaxIterations 是总迭代上限（含首跑）：已跑 LoopRuns+1 次。
	if step.LoopRuns+1 >= step.Loop.MaxIterations {
		return false
	}
	if step.Loop.UntilField != "" {
		st.mu.Lock()
		v := st.BB.Fields[step.Loop.UntilField]
		st.mu.Unlock()
		if v == step.Loop.UntilEquals {
			return false
		}
	}
	return true
}

// rearmLoop 把循环步与 loop.body 一起重置为 pending（同一临界区内落盘）。
func (d *flowDriver) rearmLoop(_ context.Context, st *RunState, idx int, step StepSpec) (*RunResult, *HitlGate, error) {
	st.mu.Lock()
	step.LoopRuns++
	step.Status = "pending"
	step.Error = ""
	st.Plan.Workflow.Steps[idx] = step
	for _, bodyID := range step.Loop.Body {
		for j := range st.Plan.Workflow.Steps {
			if st.Plan.Workflow.Steps[j].ID == bodyID && st.Plan.Workflow.Steps[j].Status == "completed" {
				st.Plan.Workflow.Steps[j].Status = "pending"
				st.Plan.Workflow.Steps[j].Error = ""
			}
		}
	}
	err := st.save()
	round := step.LoopRuns
	st.mu.Unlock()
	if err != nil {
		return nil, nil, err
	}
	d.emit(st, Event{Type: EvPhase, StepID: step.ID, Text: fmt.Sprintf("loop 第 %d 轮再武装：%s", round+1, step.ID)})
	return nil, nil, nil
}

// applyBranches 裁决路由步分支：When 按声明序取首个命中；未选中分支的
// pending 目标置 skipped（skipped=依赖已满足，join 步照常推进）。
func (d *flowDriver) applyBranches(_ context.Context, st *RunState, _ int, step StepSpec) (*RunResult, *HitlGate, error) {
	st.mu.Lock()
	fields := map[string]string{}
	for k, v := range st.BB.Fields {
		fields[k] = v
	}
	st.mu.Unlock()
	var selected *Branch
	for i := range step.Branches {
		if branchMatches(fields, step.Branches[i].When) {
			selected = &step.Branches[i]
			break
		}
	}
	if selected == nil {
		wf := st.Workflow()
		wf.Status = "failed"
		st.setWorkflow(wf)
		_ = st.save()
		return &RunResult{RunID: st.Meta.ID, Status: RunFailed, Plan: planPtr(st),
			Error: fmt.Sprintf("router: 步骤 %s 无分支命中（fail-closed）", step.ID), WorkDir: st.Meta.WorkDir}, nil, nil
	}
	keep := map[string]bool{}
	for _, t := range selected.Then {
		keep[t] = true
	}
	skipped := 0
	st.mu.Lock()
	for j := range st.Plan.Workflow.Steps {
		sj := &st.Plan.Workflow.Steps[j]
		if sj.ID == step.ID || keep[sj.ID] || sj.Status != "pending" {
			continue
		}
		for _, br := range step.Branches {
			inBranch := false
			for _, t := range br.Then {
				if t == sj.ID {
					inBranch = true
					break
				}
			}
			if inBranch {
				sj.Status = "skipped"
				sj.Error = routerSkipReason
				skipped++
				break
			}
		}
	}
	err := st.save()
	st.mu.Unlock()
	if err != nil {
		return nil, nil, err
	}
	d.emit(st, Event{Type: EvPhase, StepID: step.ID,
		Text: fmt.Sprintf("router %s 选中分支 %v（跳过 %d 步）", step.ID, selected.Then, skipped)})
	return nil, nil, nil
}

// routerSkipReason 是分支落选的 skipped 标记（finalize 的降级豁免凭据）。
const routerSkipReason = "router: 分支未选中"

// branchMatches 判定分支条件：Equals 非空须相等，NotEquals 非空须不等；
// 字段未设时按零值 "" 参与比较（compileGraph 已保证两者至少其一非空）。
func branchMatches(fields map[string]string, c BranchCond) bool {
	v := fields[c.Field]
	if c.Equals != "" && v != c.Equals {
		return false
	}
	if c.NotEquals != "" && v == c.NotEquals {
		return false
	}
	return true
}

// failStep 按关键性处理失败步骤：关键 → 运行失败；非关键 → 标记后继续（返回 nil 让循环推进）。
func (d *flowDriver) failStep(ctx context.Context, st *RunState, idx int) (*RunResult, *HitlGate, error) {
	st.mu.Lock()
	step := st.Plan.Workflow.Steps[idx]
	critical := step.IsCritical()
	if critical {
		wf := st.Workflow()
		wf.Status = "failed"
		st.setWorkflow(wf)
	}
	_ = st.save()
	st.mu.Unlock()
	if critical {
		return &RunResult{
			RunID: st.Meta.ID, Status: RunFailed, Plan: planPtr(st), Error: fmt.Sprintf("step %s(%s) failed: %s", step.ID, step.Role, step.Error),
			WorkDir: st.Meta.WorkDir,
		}, nil, nil
	}
	// 非关键降级：标记 failed，工作流继续，收口时 degraded。
	d.emit(st, Event{Type: EvStepSkipped, StepID: step.ID, Role: step.Role,
		Text: "非关键步骤失败，降级继续: " + step.Error})
	return nil, nil, nil
}

// checkBlockingGaps 检查未解决阻塞缺口；interactive 挂起，auto 记录兜底假设并继续。
func (d *flowDriver) checkBlockingGaps(st *RunState) *HitlGate {
	gaps := st.BB.UnresolvedBlockingGaps()
	if len(gaps) == 0 {
		return nil
	}
	mode := d.policy(st).mode()
	if mode == HitlAuto {
		for _, g := range gaps {
			st.BB.Assumptions = append(st.BB.Assumptions, AssumptionItem{
				ID: fmt.Sprintf("asm_%d", len(st.BB.Assumptions)+1), Topic: g.ID,
				Statement: "无人值守模式：未获得人工输入，按现状继续推进（缺口保持未解决）",
				Rationale: g.Item, Source: "fallback",
			})
			for i := range st.BB.Gaps {
				if st.BB.Gaps[i].ID == g.ID {
					st.BB.Gaps[i].Blocking = false
				}
			}
		}
		_ = st.save()
		return nil
	}
	if mode == HitlDisabled {
		return nil // 交由步骤级失败语义处理；缺口不再阻塞
	}
	var lines []string
	for _, g := range gaps {
		lines = append(lines, fmt.Sprintf("- [%s] %s（owner=%s）", g.ID, g.Item, g.Owner))
	}
	waitCond := waitConditionFrom(gaps)
	req := &HumanRequest{
		ID: newRequestID(), RunID: st.Meta.ID, Kind: HitlGap, Severity: SeverityActionRequired,
		WaitCondition: waitCond,
		Question:      fmt.Sprintf("发现 %d 个阻塞缺口，需要人工补充信息后才能继续：", len(gaps)),
		Options: []HumanOption{
			{Value: "resolved", Label: "已在回答中补充信息，继续执行"},
			{Value: "skip", Label: "跳过缺口，按现状继续"},
			{Value: "abort", Label: "中止运行"},
		},
		Context:   strings.Join(lines, "\n"),
		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()
		d.emitHuman(st, req)
	}
	return gate
}

// applyDynamicAdjustment 评估步骤产出并应用/审批动态调整。
func (d *flowDriver) applyDynamicAdjustment(ctx context.Context, st *RunState, idx int) (*RunResult, *HitlGate, error) {
	step := st.Plan.Workflow.Steps[idx]
	subDir := filepath.Join(st.Meta.WorkDir, filepath.FromSlash(step.SubWorkspace))
	output := readFileOr(filepath.Join(subDir, firstTarget(step.Targets)), "")
	evalRes, err := d.opts.Evaluator.Evaluate(ctx, EvaluateRequest{
		RunID: st.Meta.ID, Step: step, Workspace: st.Meta.WorkDir, Output: output, Blackboard: &st.BB,
	})
	if err != nil || evalRes == nil {
		return nil, nil, err
	}
	for _, na := range evalRes.NewAssumptions {
		st.BB.Assumptions = append(st.BB.Assumptions, na)
	}
	for _, gapID := range evalRes.ResolvedGaps {
		st.BB.ResolveGapByID(gapID, "")
	}
	hasChanges := len(evalRes.DynamicStepsToAdd) > 0 || len(evalRes.StepsToSkip) > 0
	if !hasChanges {
		return nil, nil, nil
	}
	if evalRes.NeedsApproval && d.policy(st).mode() == HitlInteractive {
		var addLines, skipLines []string
		for _, s := range evalRes.DynamicStepsToAdd {
			addLines = append(addLines, fmt.Sprintf("- %s（%s）%s", s.ID, s.Role, s.Title))
		}
		for _, id := range evalRes.StepsToSkip {
			skipLines = append(skipLines, "- "+id)
		}
		req := &HumanRequest{
			ID: newRequestID(), RunID: st.Meta.ID, StepID: step.ID, Kind: HitlAdjustment, Severity: SeverityAttention,
			Question: "评估器建议动态调整工作流，是否应用？",
			Options: []HumanOption{
				{Value: "approve", Label: "应用调整"},
				{Value: "reject", Label: "忽略调整"},
				{Value: "abort", Label: "中止运行"},
			},
			Context: fmt.Sprintf("评估理由：%s\n追加步骤：\n%s\n跳过步骤：\n%s",
				evalRes.Reason, strings.Join(addLines, "\n"), strings.Join(skipLines, "\n")),
			CreatedAt: time.Now(), Status: "pending",
		}
		gate := &HitlGate{Request: *req, Steps: evalRes.DynamicStepsToAdd, SkipIDs: evalRes.StepsToSkip, PrevStatus: RunExecuting}
		if err := st.saveGate(gate); err == nil {
			st.Meta.Status = string(RunWaitingHuman)
			_ = st.save()
			d.emitHuman(st, req)
		}
		return heldResult(st, gate), gate, nil
	}
	applyAdjustment(st, evalRes.DynamicStepsToAdd, evalRes.StepsToSkip)
	return nil, nil, nil
}

// applyAdjustment 应用动态调整：新步骤插到最后一步之前，跳过步骤标记 skipped。
func applyAdjustment(st *RunState, add []StepSpec, skipIDs []string) {
	steps := st.Plan.Workflow.Steps
	for _, id := range skipIDs {
		for i := range steps {
			if steps[i].ID == id && steps[i].Status == "pending" {
				steps[i].Status = "skipped"
			}
		}
	}
	if len(add) > 0 {
		insertAt := len(steps)
		if insertAt > 0 {
			insertAt-- // 插在最后（成文/终审）步骤之前
		}
		for i := range add {
			add[i].Status = "pending"
			if add[i].Targets == nil {
				add[i].Targets = []string{"notes.md", "fields.json"}
			}
		}
		steps = append(steps[:insertAt], append(append([]StepSpec{}, add...), steps[insertAt:]...)...)
	}
	st.Plan.Workflow.Steps = steps
	_ = st.save()
}

// finalize 收口：汇总交付物 → 质检 → 自动修复 → （必要时）人工裁决 → 清单。
func (d *flowDriver) finalize(ctx context.Context, st *RunState) (*RunResult, *HitlGate, error) {
	wf := st.Workflow()
	degraded := false
	for _, s := range wf.Steps {
		if s.Status == "failed" {
			degraded = true
		}
		// router 分支落选是预期路径，不算降级（Q2）。
		if s.Status == "skipped" && s.Error != routerSkipReason {
			degraded = true
		}
	}
	// 汇总交付物：最后一个 completed 步骤的主产出 → output.md。
	// handoff 链尾优先（Q2）：有所有权移交时，交付物取链尾角色的最后一个
	// completed 步骤产出（最终答案所有权在链尾），无则该角色无完成步骤时回落。
	var srcStep *StepSpec
	for i := range wf.Steps {
		if wf.Steps[i].Status == "completed" {
			srcStep = &wf.Steps[i]
		}
	}
	if tail := handoffChainTail(st); tail != "" {
		for i := range wf.Steps {
			if wf.Steps[i].Status == "completed" && wf.Steps[i].Role == tail {
				srcStep = &wf.Steps[i]
			}
		}
	}
	if srcStep == nil {
		wf.Status = "failed"
		st.setWorkflow(wf)
		_ = st.save()
		return &RunResult{RunID: st.Meta.ID, Status: RunFailed, Plan: planPtr(st),
			Error: "no completed step to assemble deliverable", WorkDir: st.Meta.WorkDir}, nil, nil
	}
	src := filepath.Join(st.Meta.WorkDir, filepath.FromSlash(srcStep.SubWorkspace), firstTarget(srcStep.Targets))
	data, err := os.ReadFile(src)
	if err != nil {
		wf.Status = "failed"
		st.setWorkflow(wf)
		_ = st.save()
		return &RunResult{RunID: st.Meta.ID, Status: RunFailed, Plan: planPtr(st),
			Error: "deliverable source missing: " + err.Error(), WorkDir: st.Meta.WorkDir}, nil, nil
	}
	deliverable := "output.md"
	if err := os.WriteFile(filepath.Join(st.Meta.WorkDir, deliverable), data, 0o644); err != nil {
		return nil, nil, err
	}
	d.emit(st, Event{Type: EvDeliverable, Text: deliverable})

	// 质检 + 有界自动修复。
	report := d.opts.Checker.Check(CheckRequest{
		RunID: st.Meta.ID, Workspace: st.Meta.WorkDir, Deliverable: deliverable,
		Plan: planPtr(st), Squad: &st.Plan.Squad, Template: st.Template,
	})
	if srcStep.MinScore > 0 && report.Score < float64(srcStep.MinScore) {
		report.Passed = false
		report.Violations = append(report.Violations, fmt.Sprintf("总分 %.0f 低于步骤门槛 %d", report.Score, srcStep.MinScore))
	}
	if !report.Passed && st.RepairBudgetRemaining > 0 {
		st.RepairBudgetRemaining--
		_ = st.save()
		// 重跑交付物源步骤一次，再检一次。
		if out, g, err := d.rerunForRepair(ctx, st, srcStep.ID); out != nil || g != nil || err != nil {
			return out, g, err
		}
		data2, _ := os.ReadFile(src)
		_ = os.WriteFile(filepath.Join(st.Meta.WorkDir, deliverable), data2, 0o644)
		report = d.opts.Checker.Check(CheckRequest{
			RunID: st.Meta.ID, Workspace: st.Meta.WorkDir, Deliverable: deliverable,
			Plan: planPtr(st), Squad: &st.Plan.Squad, Template: st.Template,
		})
	}

	if !report.Passed {
		switch d.policy(st).mode() {
		case HitlDisabled, HitlAuto:
			wf.Status = "failed"
			st.setWorkflow(wf)
			_ = st.save()
			return &RunResult{RunID: st.Meta.ID, Status: RunFailed, Plan: planPtr(st),
				Quality: report, Error: "quality check failed (hitl mode=" + string(d.policy(st).mode()) + "): " + strings.Join(report.Violations, "; "),
				WorkDir: st.Meta.WorkDir}, nil, nil
		default:
			req := &HumanRequest{
				ID: newRequestID(), RunID: st.Meta.ID, Kind: HitlQualityReview, Severity: SeverityActionRequired,
				Question: "交付物质检未通过且自动修复预算已用尽，请人工裁决：",
				Options: []HumanOption{
					{Value: "approve", Label: "人工放行（记录人工通过）"},
					{Value: "retry", Label: "再修复一次（重跑交付步骤）"},
					{Value: "abort", Label: "判定失败"},
				},
				Context:   string(marshalJSON(report)),
				CreatedAt: time.Now(), Status: "pending",
			}
			gate := &HitlGate{Request: *req, PrevStatus: RunExecuting}
			if err := st.saveGate(gate); err == nil {
				st.Meta.Status = string(RunQualityReview)
				_ = st.save()
				d.emitHuman(st, req)
			}
			return heldResult(st, gate), gate, nil
		}
	}

	status := RunCompleted
	if degraded {
		status = RunDegraded
	}
	now := time.Now()
	wf.Status = string(status)
	st.setWorkflow(wf)
	st.Meta.Status = string(status)
	st.Meta.CompletedAt = &now
	_ = st.save()

	manifest := buildManifest(st, deliverable, *report)
	_ = writeJSON(filepath.Join(st.Meta.WorkDir, "manifest.json"), manifest)

	d.emit(st, Event{Type: EvRunCompleted, Text: string(status),
		Data: marshalJSON(map[string]any{"deliverable": deliverable, "score": report.Score})})
	return &RunResult{
		RunID: st.Meta.ID, Status: status, Plan: planPtr(st), Deliverable: deliverable,
		Quality: report, WorkDir: st.Meta.WorkDir, Usage: st.usageSnapshot(),
		Summary: summarize(st, *report, status),
	}, nil, nil
}

// rerunForRepair 重置交付物源步骤并重新执行到收口。
func (d *flowDriver) rerunForRepair(ctx context.Context, st *RunState, stepID string) (*RunResult, *HitlGate, error) {
	for i := range st.Plan.Workflow.Steps {
		if st.Plan.Workflow.Steps[i].ID == stepID {
			st.Plan.Workflow.Steps[i].Status = "pending"
			st.Plan.Workflow.Steps[i].Error = ""
			note, err := os.ReadFile(filepath.Join(st.Meta.WorkDir, "quality-feedback.md"))
			if err == nil && len(bytes.TrimSpace(note)) > 0 {
				taskPath := filepath.Join(st.Meta.WorkDir, filepath.FromSlash(st.Plan.Workflow.Steps[i].SubWorkspace), "TASK.md")
				existing, _ := os.ReadFile(taskPath)
				_ = os.WriteFile(taskPath, append(existing, append([]byte("\n\n"), note...)...), 0o644)
			}
		}
	}
	d.emit(st, Event{Type: EvPhase, Text: "质检未通过，自动修复：重跑交付步骤 " + stepID})
	return d.advance(ctx, st)
}

// heldResult 由挂起上下文构造对外结果（挂起路径统一返回非 nil 结果）。
func heldResult(st *RunState, gate *HitlGate) *RunResult {
	status := RunWaitingHuman
	switch gate.Request.Kind {
	case HitlClarification:
		status = RunAwaitingClarification
	case HitlConfirmation:
		status = RunAwaitingConfirmation
	case HitlQualityReview:
		status = RunQualityReview
	}
	return &RunResult{
		RunID: st.Meta.ID, Status: status, Plan: planPtr(st),
		Pending: &gate.Request, WorkDir: st.Meta.WorkDir, Usage: st.usageSnapshot(),
	}
}

// waitConditionFrom 从阻塞缺口提取等待条件（首个声明者胜出）。
func waitConditionFrom(gaps []GapItem) *WaitCondition {
	for _, g := range gaps {
		if g.WaitRunID != "" {
			return &WaitCondition{Type: "run_done", RunID: g.WaitRunID}
		}
		if g.WaitField != "" {
			return &WaitCondition{Type: "field_set", Field: g.WaitField}
		}
	}
	return nil
}

// hold 生成挂起请求并返回结果。
func (d *flowDriver) hold(st *RunState, kind HumanRequestKind, stepID, role, question string, options []HumanOption, context, reason string, extra *HitlGate) (*RunResult, *HitlGate, error) {
	req := &HumanRequest{
		ID: newRequestID(), RunID: st.Meta.ID, StepID: stepID, Kind: kind,
		Severity: SeverityActionRequired,
		Question: question, Options: options, Context: context, Reason: reason,
		CreatedAt: time.Now(), Status: "pending",
	}
	status := RunWaitingHuman
	switch kind {
	case HitlClarification:
		status = RunAwaitingClarification
	case HitlConfirmation:
		status = RunAwaitingConfirmation
	}
	gate := extra
	if gate == nil {
		gate = &HitlGate{}
	}
	gate.Request = *req
	gate.PrevStatus = RunExecuting
	if err := st.saveGate(gate); err != nil {
		return nil, nil, err
	}
	st.Meta.Status = string(status)
	_ = st.save()
	d.emitHuman(st, req)
	return &RunResult{RunID: st.Meta.ID, Status: status, Plan: planPtr(st),
		Pending: req, WorkDir: st.Meta.WorkDir}, nil, nil
}

// policy 合并引擎全局策略与 Goal 覆盖（窄者胜）。
// policyRole 在引擎全局/Goal 链之上叠加角色级 HitlPolicy
// （窄者胜；Goal 仍最高）。
func (d *flowDriver) policyRole(st *RunState, role RoleSpec) HitlPolicy {
	p := d.policy(st)
	if role.HitlPolicy != nil {
		p = mergeHitlPolicy(p, role.HitlPolicy)
	}
	return p
}

func (d *flowDriver) policy(st *RunState) HitlPolicy {
	return mergeHitlPolicy(d.opts.HitlPolicy, st.Meta.HitlPolicyOf())
}

func (d *flowDriver) emit(st *RunState, ev Event) {
	if d.opts.Sink == nil {
		return
	}
	ev.RunID = st.Meta.ID
	if ev.At.IsZero() {
		ev.At = time.Now()
	}
	d.opts.Sink.Emit(ev)
}

func (d *flowDriver) emitHuman(st *RunState, req *HumanRequest) {
	d.emit(st, Event{Type: EvHumanInputRequired, Text: req.Question, Data: marshalJSON(req)})
}

func summarize(st *RunState, report QualityReport, status RunStatus) string {
	var b strings.Builder
	fmt.Fprintf(&b, "运行 %s：%s。步骤 %d 个（完成 %d/跳过 %d/失败 %d）。质检得分 %.0f。",
		st.Meta.ID, status, len(st.Plan.Workflow.Steps),
		countStatus(st, "completed"), countStatus(st, "skipped"), countStatus(st, "failed"), report.Score)
	if gaps := st.BB.UnresolvedBlockingGaps(); len(gaps) > 0 {
		fmt.Fprintf(&b, "未解决缺口 %d 个。", len(gaps))
	}
	return b.String()
}

func countStatus(st *RunState, status string) int {
	n := 0
	for _, s := range st.Plan.Workflow.Steps {
		if s.Status == status {
			n++
		}
	}
	return n
}

func buildManifest(st *RunState, deliverable string, report QualityReport) Manifest {
	files := map[string]string{}
	_ = filepath.Walk(st.Meta.WorkDir, func(path string, info os.FileInfo, err error) error {
		if err != nil || info.IsDir() {
			return nil
		}
		rel, _ := filepath.Rel(st.Meta.WorkDir, path)
		files[filepath.ToSlash(rel)] = sha256File(path)
		return nil
	})
	start := st.Meta.CreatedAt
	return Manifest{
		ManifestVersion: "1", WorkspaceID: st.Meta.ID, GeneratedAt: time.Now(),
		Deliverable: deliverable, Files: files, QualityReport: report,
		DurationMs: time.Since(start).Milliseconds(),
	}
}

func firstTarget(targets []string) string {
	if len(targets) > 0 {
		return targets[0]
	}
	return "output.md"
}

func allTargetsPresent(dir string, targets []string) bool {
	return len(missingTargets(dir, targets)) == 0
}

func missingTargets(dir string, targets []string) []string {
	var missing []string
	for _, t := range targets {
		if strings.TrimSpace(t) == "" {
			continue
		}
		info, err := os.Stat(filepath.Join(dir, filepath.FromSlash(t)))
		if err != nil || info.Size() == 0 {
			missing = append(missing, t)
		}
	}
	return missing
}

func readFileOr(path, fallback string) string {
	if data, err := os.ReadFile(path); err == nil && len(strings.TrimSpace(string(data))) > 0 {
		return string(data)
	}
	return fallback
}

func appendLog(dir, line string) {
	f, err := os.OpenFile(filepath.Join(dir, "execution.log"), os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
	if err != nil {
		return
	}
	defer f.Close()
	_, _ = f.WriteString(line)
}

func planPtr(st *RunState) *WorkflowPlan {
	wf := st.Workflow()
	return &wf
}

func orDefault(v, def string) string {
	if v == "" {
		return def
	}
	return v
}