// plan_executor.go -- stepped plan execution (drives PlanProgress + projects to queue).
//
// What this is: the missing consumer that both PlanProgress and PlanQueue were
// built for but never had. PlanProgress (plan_progress.go) is a full 5-state
// step FSM with Kahn-order ReadySteps; PlanQueue (plan_queue.go) is an async
// file-backed queue. Neither shipped a driver that actually steps a plan
// forward -- StartStep/FinishStep had zero non-test callers, and the queue's
// execFunc merged every step into one prompt and fired a single e.Run (so a
// poller saw only pending -> done, never running, and skipped was inexpressible).
// This file is that driver: it walks an approved plan step by step in dependency
// order, runs one e.Run per step, and reports every transition so both the
// in-memory PlanProgress observer stream AND the on-disk queue StepStatuses
// reflect pending -> running -> done/failed/skipped in real time.
//
// 这是什么: PlanProgress 和 PlanQueue 都为之而建却一直没有的那个消费者.
// PlanProgress (plan_progress.go) 是完整 5 态步骤状态机 + Kahn 序 ReadySteps;
// PlanQueue (plan_queue.go) 是异步文件队列. 两者都没出过一个真正推进计划的驱动器
// -- StartStep/FinishStep 零非测试调用者, 队列 execFunc 把每步合并成一个 prompt
// 发一次 e.Run (轮询只见 pending -> done, 永远看不到 running, 也表达不了 skipped).
// 本文件就是那个驱动器: 按依赖顺序逐步走过已批准的计划, 每步一次 e.Run, 并报告
// 每次状态变更, 使内存 PlanProgress observer 流 + 落盘队列 StepStatuses 都实时反映
// pending -> running -> done/failed/skipped.

package engine

import (
	"context"
	"fmt"
	"strings"
	"sync"

	"git.flytoex.net/yuanwei/flyto-agent/core/pkg/query"
)

// ErrPlanStepsStuck reports that steps remain pending but none are ready -- a
// cyclic dependency or a dep referencing an unknown step. Surfaced as an error
// rather than spinning forever.
//
// ErrPlanStepsStuck 表示仍有步骤 pending 但无一 ready -- 循环依赖, 或依赖指向
// 未知步骤. 显式上报为错误而非空转死循环.
var ErrPlanStepsStuck = fmt.Errorf("plan_executor: steps pending but none ready")

// runPlanSteps drives progress through its steps in dependency (topological)
// order, one session.Send per step. It is the production entry: the queue
// execFunc calls it after building a PlanProgress from the approved plan. It
// delegates to runPlanStepsWith with a per-step runner closure, keeping the
// topological loop independent of *Engine so it stays unit-testable with a stub
// runner (no live LLM).
//
// Cross-step context: all steps of one plan share a single throwaway Session, so
// a later step sees earlier steps' output (e.g. "step-2 calls the function
// step-1 wrote"). newSession (not e.Session) keeps it OUT of
// e.sessionState.sessions -- that map has no eviction, so registering one
// session per plan would slow-leak. The session is unreferenced once the plan
// finishes and GCs naturally; Close() is best-effort lifecycle hygiene (wakes any
// pending permission waiter, though the queue path is non-interactive).
//
// runPlanSteps 按依赖 (拓扑) 顺序驱动 progress 走过其步骤, 每步一次 session.Send.
// 这是生产入口: 队列 execFunc 从已批准计划构造 PlanProgress 后调它. 它委托给
// runPlanStepsWith, 以 per-step 执行器闭包作单步执行器, 使拓扑循环独立于 *Engine,
// 从而可用 stub 执行器单测 (不跑真 LLM).
//
// 跨步上下文: 一个 plan 的所有步骤共享单个 throwaway Session, 使后续步骤能看到
// 前面步骤的产出 (例 "step-2 调用 step-1 写的函数"). 用 newSession (非 e.Session)
// 使其不进 e.sessionState.sessions -- 那个 map 无驱逐, 每 plan 注册一个会慢性
// 泄漏. 本 session 在 plan 结束后无引用自然 GC; Close() 是 best-effort 生命周期
// 收尾 (唤醒任何 pending 权限等待者, 虽然队列路径是非交互的).
func (e *Engine) runPlanSteps(ctx context.Context, planID string, parallel bool, progress *PlanProgress, onStep func(stepID string, status StepExecStatus, errMsg string)) error {
	sess := newSession(planID, e)
	defer sess.Close()

	// resolveModel returns the runtime run model so a mid-plan provider hot-swap
	// (ADR-0017) is picked up by later steps/waves -- matches the prior per-step
	// e.Run behavior. Empty when no resolver is wired (Config.DefaultRunModelFunc):
	// Send/fork fall back to the frozen Config.Model.
	//
	// resolveModel 返回运行时 run model, 使 plan 执行途中的 provider 热换 (ADR-0017)
	// 被后续步骤/wave 接住 -- 对齐原 per-step e.Run 行为. 未接 resolver
	// (Config.DefaultRunModelFunc) 时为空: Send/fork 回落冻结的 Config.Model.
	resolveModel := func() string {
		if e.defaultRunModel != nil {
			return e.defaultRunModel()
		}
		return ""
	}

	if parallel {
		// Parallel mode (opt-in per plan): each topological wave of ready steps
		// fans out into isolated forked sub-agents. A sub-agent's runLoop does
		// NOT call SetMessageID (only the main loop does), so concurrent workers
		// avoid the shared-tool data race that concurrent main Runs would hit --
		// proven race-free by plan_parallel_spike_test.go. Workers are seeded with
		// the shared session's history snapshot so they still thread prior-wave
		// context, and each wave's outputs merge back in deterministic order so
		// later waves see them. File-conflict safety is the plan author's
		// contract: concurrently-ready steps must not write the same files (the
		// DAG encodes declared deps, not file-level conflicts) -- hence default off.
		//
		// 并行模式 (per-plan opt-in): 每个拓扑 wave 的 ready 步骤扇出成隔离的 fork
		// sub-agent. sub-agent 的 runLoop 不调 SetMessageID (只主循环调), 故并发
		// worker 避开了并发主 Run 会撞的共享工具 data race -- plan_parallel_spike_test.go
		// 证 race-free. worker 用共享 session 的历史快照 seed 故仍串联前面 wave 的
		// 上下文, 每个 wave 的产出按确定顺序 merge 回去使后续 wave 看到. 文件冲突
		// 安全是 plan 作者的契约: 同时 ready 的步骤不得写同一文件 (DAG 编码声明的依赖,
		// 不编码文件级冲突) -- 故默认关.
		runWave := func(ctx context.Context, steps []PlanStep) []planStepResult {
			return e.runPlanWave(ctx, sess, resolveModel(), steps)
		}
		return runPlanStepsParallelWith(ctx, progress, onStep, runWave)
	}

	// Serial mode (default, #1): one step at a time on the shared session.
	// 串行模式 (默认, #1): 共享会话上每次一步.
	runStep := func(ctx context.Context, step PlanStep) error {
		return runOnePlanStep(ctx, sess, step, resolveModel())
	}
	return runPlanStepsWith(ctx, progress, onStep, runStep)
}

// runPlanStepsWith is the *Engine-independent topological driver. runStep
// executes a single step (production: e.runOnePlanStep; tests: a stub). Every
// state change is pushed to onStep (may be nil) so a queue/UI can project it;
// PlanProgress also emits its own observer events internally. Independent
// branches keep running after a failure -- a failed step's dependents are marked
// skipped (soft), but steps not depending on it still execute. Reports failed
// iff any step ended in the failed state.
//
// CLEVER: ReadySteps() is the Kahn-algorithm consumer side -- snapshot, take a
// ready step, run it, repeat. No bespoke topo-sort lives here; the ordering is
// owned (and unit-tested) inside PlanProgress.
//
// runPlanStepsWith 是独立于 *Engine 的拓扑驱动器. runStep 执行单步 (生产:
// e.runOnePlanStep; 测试: stub). 每次状态变更推给 onStep (可为 nil) 供队列/UI
// 投影; PlanProgress 内部也发自己的 observer 事件. 失败后独立分支继续跑 -- 失败
// 步骤的依赖者标 skipped (软变体), 但不依赖它的步骤仍执行. 当且仅当有步骤以 failed
// 收尾时报失败.
//
// CLEVER: ReadySteps() 是 Kahn 算法的消费者端 -- 快照, 取一个 ready 步骤, 跑它,
// 重复. 这里没有定制拓扑排序; 排序逻辑住在 PlanProgress 里且有单测.
func runPlanStepsWith(ctx context.Context, progress *PlanProgress, onStep func(stepID string, status StepExecStatus, errMsg string), runStep func(context.Context, PlanStep) error) error {
	report := func(stepID string, status StepExecStatus, errMsg string) {
		if onStep != nil {
			onStep(stepID, status, errMsg)
		}
	}

	// firstFailure captures the first failing step's error verbatim so the
	// plan-level ErrorMsg can name the real cause at a glance -- a single summary
	// pointing at the first failure, so a client need not scan per-step StepErrors
	// (the queue now persists those separately via the failed onStep callback).
	//
	// firstFailure 原样捕获首个失败步骤的错误, 使 plan 级 ErrorMsg 能一眼点出真因 --
	// 单一摘要指向首个失败, 客户端不必遍历 per-step StepErrors (队列现在经 failed
	// onStep 回调单独持久化后者).
	var firstFailure string

	for {
		if ctx.Err() != nil {
			return ctx.Err()
		}

		snap := progress.Snapshot()
		// Exit when no step is left to run. Note: not snap.IsComplete() -- that
		// requires TotalCount > 0, which would misclassify an empty plan (zero
		// steps) as "not complete" and fall through to the stuck check below.
		//
		// 无待跑步骤即退出. 注意不用 snap.IsComplete() -- 它要求 TotalCount > 0, 会把
		// 空计划 (零步骤) 误判为 "未完成" 而落到下面的 stuck 检查.
		if snap.PendingCount == 0 && snap.RunningCount == 0 {
			break
		}

		ready := snap.ReadySteps()
		if len(ready) == 0 {
			// Pending steps remain but none are ready: cyclic or dangling deps.
			// (Serial executor never leaves a step Running across iterations, so
			// "no ready + not complete" can only be a dependency-graph defect.)
			//
			// 仍有 pending 但无一 ready: 循环依赖或悬空依赖. (串行执行器不会跨轮留下
			// Running 步骤, 故 "无 ready 且未完成" 只能是依赖图缺陷.)
			return fmt.Errorf("%w (%d pending)", ErrPlanStepsStuck, snap.PendingCount)
		}

		// Serial: take one ready step per iteration. Parallel fan-out of all
		// currently-ready steps lives in runPlanStepsParallelWith (opt-in per
		// plan); serial stays the safe default (plan_queue.go runLoop CLEVER note:
		// avoids steps clobbering the same files).
		//
		// 串行: 每轮取一个 ready 步骤. 并行扇出所有当前 ready 步骤在
		// runPlanStepsParallelWith (per-plan opt-in); 串行仍是安全默认
		// (plan_queue.go runLoop CLEVER: 避免多步互踩同一批文件).
		step := ready[0].Step

		_ = progress.StartStep(step.ID, "")
		report(step.ID, StepExecRunning, "")

		stepErr := runStep(ctx, step)
		if stepErr != nil {
			if firstFailure == "" {
				firstFailure = fmt.Sprintf("%s: %v", step.ID, stepErr)
			}
			_ = progress.FinishStep(step.ID, StepStatusFailed, stepErr.Error())
			report(step.ID, StepExecFailed, stepErr.Error())

			// Soft-fail: mark dependents skipped, let independent branches run on.
			// 软失败: 标依赖者 skipped, 让独立分支继续跑.
			for _, id := range progress.SkipDependents(step.ID) {
				report(id, StepExecSkipped, "")
			}
			continue
		}

		_ = progress.FinishStep(step.ID, StepStatusDone, "")
		report(step.ID, StepExecDone, "")
	}

	final := progress.Snapshot()
	if final.HasFailed() {
		return fmt.Errorf("plan_executor: %d of %d steps failed (first failure -- %s)", final.FailedCount, final.TotalCount, firstFailure)
	}
	return nil
}

// planStepResult is one wave step's outcome, returned by a runWaveFunc in the
// SAME order as the input steps so the driver can finalize each step's progress.
// reply is the step's assistant output (already merged into shared history by
// the wave runner); err is non-nil iff the step failed.
//
// planStepResult 是一个 wave 步骤的结果, runWaveFunc 按输入步骤的相同顺序返回,
// 使驱动器能逐步收尾 progress. reply 是步骤的助手输出 (已由 wave runner merge 进
// 共享历史); err 非 nil 当且仅当步骤失败.
type planStepResult struct {
	reply string
	err   error
}

// runWaveFunc executes ALL steps of one topological wave and returns their
// results in input order. Production (e.runPlanWave) forks one isolated
// sub-agent per step and runs them concurrently; tests inject a stub. The wave
// runner owns history seeding + merge-back (session mutation); the driver below
// owns only progress bookkeeping, keeping it *Engine/Session-independent and
// unit-testable with a stub wave runner.
//
// runWaveFunc 执行一个拓扑 wave 的所有步骤, 按输入顺序返回结果. 生产
// (e.runPlanWave) 每步 fork 一个隔离 sub-agent 并发跑; 测试注入 stub. wave runner
// 负责历史 seed + merge 回写 (session 变更); 下面的驱动器只负责 progress 记账, 从而
// 独立于 *Engine/Session 且可用 stub wave runner 单测.
type runWaveFunc func(ctx context.Context, steps []PlanStep) []planStepResult

// runPlanStepsParallelWith is the parallel topological driver: each iteration
// takes the WHOLE ready set (a wave) and hands it to runWave for concurrent
// execution, instead of the serial runPlanStepsWith taking ready[0]. The wave's
// steps are mutually independent (all currently-ready, so none depends on
// another in the same wave), which is exactly what makes concurrent execution
// safe at the dependency-graph level. Marks all wave steps running before
// fan-out; after the barrier, finalizes each (done/failed) in deterministic wave
// order and skips a failed step's dependents (soft-fail, like the serial driver).
//
// runPlanStepsParallelWith 是并行拓扑驱动器: 每轮取整个 ready 集合 (一个 wave) 交给
// runWave 并发执行, 而非串行 runPlanStepsWith 取 ready[0]. wave 的步骤互相独立
// (都当前 ready, 同 wave 内无一依赖另一个), 这正是依赖图层面并发安全的依据. 扇出前
// 标所有 wave 步骤 running; barrier 后按确定的 wave 顺序逐个收尾 (done/failed) 并跳过
// 失败步骤的依赖者 (软失败, 同串行驱动器).
func runPlanStepsParallelWith(ctx context.Context, progress *PlanProgress, onStep func(stepID string, status StepExecStatus, errMsg string), runWave runWaveFunc) error {
	report := func(stepID string, status StepExecStatus, errMsg string) {
		if onStep != nil {
			onStep(stepID, status, errMsg)
		}
	}

	var firstFailure string

	for {
		if ctx.Err() != nil {
			return ctx.Err()
		}

		snap := progress.Snapshot()
		if snap.PendingCount == 0 && snap.RunningCount == 0 {
			break
		}

		ready := snap.ReadySteps()
		if len(ready) == 0 {
			return fmt.Errorf("%w (%d pending)", ErrPlanStepsStuck, snap.PendingCount)
		}

		// The whole ready set is this wave.
		// 整个 ready 集合即本 wave.
		wave := make([]PlanStep, len(ready))
		for i, rs := range ready {
			wave[i] = rs.Step
			_ = progress.StartStep(wave[i].ID, "")
			report(wave[i].ID, StepExecRunning, "")
		}

		results := runWave(ctx, wave)

		// Defensive: a misbehaving runWave returning the wrong count would
		// desync the loop. Fail loud rather than index out of range.
		// 防御: runWave 返回数量不符会使循环失同步. 响亮失败而非越界.
		if len(results) != len(wave) {
			return fmt.Errorf("plan_executor: parallel wave returned %d results for %d steps", len(results), len(wave))
		}

		for i, step := range wave {
			res := results[i]
			if res.err != nil {
				if firstFailure == "" {
					firstFailure = fmt.Sprintf("%s: %v", step.ID, res.err)
				}
				_ = progress.FinishStep(step.ID, StepStatusFailed, res.err.Error())
				report(step.ID, StepExecFailed, res.err.Error())
				for _, id := range progress.SkipDependents(step.ID) {
					report(id, StepExecSkipped, "")
				}
				continue
			}
			_ = progress.FinishStep(step.ID, StepStatusDone, "")
			report(step.ID, StepExecDone, "")
		}
	}

	final := progress.Snapshot()
	if final.HasFailed() {
		return fmt.Errorf("plan_executor: %d of %d steps failed (first failure -- %s)", final.FailedCount, final.TotalCount, firstFailure)
	}
	return nil
}

// runPlanWave is the production runWaveFunc: it runs every step of a wave in its
// own forked sub-agent CONCURRENTLY (WaitGroup barrier, mirroring
// Team.RunWorkersSync), then merges each step's (prompt, reply) back into the
// shared session in deterministic input order so later waves thread the work.
//
// CLEVER: the history snapshot is taken ONCE, before fan-out, and every worker
// is seeded with the SAME snapshot -- correct because the wave's steps are
// mutually independent (they build on prior context, not on each other). Workers
// never touch the shared session during the goroutines (only read their seeded
// copy); the merge-back happens here, after the barrier, in a single goroutine,
// so the session history stays deterministic regardless of worker finish order.
//
// runPlanWave 是生产 runWaveFunc: 每步在自己的 fork sub-agent 里并发跑 (WaitGroup
// barrier, 镜像 Team.RunWorkersSync), 然后按确定的输入顺序把每步的 (prompt, reply)
// merge 回共享 session 使后续 wave 串联到这些工作.
//
// CLEVER: 历史快照在扇出前只取一次, 每个 worker 用同一快照 seed -- 正确, 因为 wave
// 的步骤互相独立 (它们基于前面的上下文, 不基于彼此). worker 在 goroutine 期间从不
// 碰共享 session (只读自己 seed 的副本); merge 回写在这里 barrier 之后单 goroutine
// 做, 故 session 历史与 worker 完成顺序无关, 保持确定.
func (e *Engine) runPlanWave(ctx context.Context, sess *Session, model string, steps []PlanStep) []planStepResult {
	// Shared prior-context snapshot for every worker in this wave.
	// 本 wave 每个 worker 共享的前置上下文快照.
	history := sess.Messages()

	type forkOutcome struct {
		prompt string
		reply  string
		err    error
	}
	outcomes := make([]forkOutcome, len(steps))

	var wg sync.WaitGroup
	for i := range steps {
		wg.Add(1)
		go func(idx int) {
			defer wg.Done()
			prompt, reply, err := e.runOnePlanStepForked(ctx, history, steps[idx], model)
			outcomes[idx] = forkOutcome{prompt: prompt, reply: reply, err: err}
		}(i)
	}
	wg.Wait()

	results := make([]planStepResult, len(steps))
	for i := range steps {
		oc := outcomes[i]
		// Thread this step's turn into the shared session (deterministic order)
		// so later waves see it. Mirrors the serial path, where Session.Send
		// auto-appends each step's turn. applyTurn with zero stats: the forked
		// sub-agent owns its own token/cost accounting.
		//
		// 把本步的轮次 merge 进共享 session (确定顺序) 使后续 wave 看到. 对齐串行
		// 路径 (那里 Session.Send 自动追加每步轮次). applyTurn 传零统计: fork 的
		// sub-agent 自己记 token/cost.
		sess.applyTurn(oc.prompt, oc.reply, 0, 0, 0)
		results[i] = planStepResult{reply: oc.reply, err: oc.err}
	}
	return results
}

// runOnePlanStep executes a single plan step through the plan's shared session
// (sess.Send, not a bare engine.Run), threading conversation history so this
// step sees prior steps' output. Returns the last ErrorEvent seen (nil =
// success). The step Description becomes the prompt; Tools is surfaced as a hint
// only (permission gating is enforced in the engine dispatch path, not here).
// model (when non-empty) targets the runtime run model via WithModel; the caller
// resolves it per step (see runPlanSteps).
//
// *Engine-independent (a free function taking *Session) so the cross-step
// context-threading path is unit-testable with a fake-EngineRef-backed session,
// mirroring runPlanStepsWith's stub-runner testability. Session.Send appends
// WithMessages(history) last, so the model opt here never clobbers the threaded
// history (see Session.Send's opts-order invariant).
//
// runOnePlanStep 把单个计划步骤经该 plan 的共享会话执行 (sess.Send, 不是裸
// engine.Run), 串联对话历史使本步能看到前面步骤的产出. 返回最后一个 ErrorEvent
// (nil = 成功). 步骤 Description 成为 prompt; Tools 仅作提示透出 (权限闸在引擎
// dispatch 路径强制, 不在这里). model 非空时经 WithModel 锁定运行时 run model;
// 由调用方每步解析 (见 runPlanSteps).
//
// 独立于 *Engine (接收 *Session 的自由函数), 使跨步上下文串联路径可用
// fake-EngineRef 撑起的 session 单测, 对齐 runPlanStepsWith 的 stub-runner 可测性.
// Session.Send 把 WithMessages(history) append 在最后, 故这里的 model opt 永不
// 覆盖串联的历史 (见 Session.Send 的 opts 顺序 invariant).
func runOnePlanStep(ctx context.Context, sess *Session, step PlanStep, model string) error {
	var opts []RunOption
	if model != "" {
		opts = append(opts, WithModel(model))
	}

	var lastErr error
	for evt := range sess.Send(ctx, buildPlanStepPrompt(step), opts...) {
		if errEvt, ok := evt.(*ErrorEvent); ok {
			lastErr = errEvt.Err
		}
	}
	return lastErr
}

// buildPlanStepPrompt renders a single step into the per-step prompt. Shared by
// the serial (runOnePlanStep) and parallel (runOnePlanStepForked) paths so both
// drive the model with identical step framing. The step Description is the task;
// Tools is a non-binding hint (real permission gating is in the dispatch path).
//
// buildPlanStepPrompt 把单个步骤渲染成 per-step prompt. 串行 (runOnePlanStep) 与
// 并行 (runOnePlanStepForked) 路径共用, 使两者以相同的步骤框架驱动模型. 步骤
// Description 是任务; Tools 是非约束提示 (真权限闸在 dispatch 路径).
func buildPlanStepPrompt(step PlanStep) string {
	var sb strings.Builder
	sb.WriteString("执行以下计划步骤, 完成后简述结果.\n\n")
	sb.WriteString(step.Description)
	if len(step.Tools) > 0 {
		sb.WriteString("\n\n建议使用工具: ")
		sb.WriteString(strings.Join(step.Tools, ", "))
	}
	return sb.String()
}

// runOnePlanStepForked executes a single step in its OWN forked sub-agent
// (isolated runLoop), seeded with the wave's shared history snapshot so it
// threads prior-wave context. Returns (prompt, reply, err): prompt + reply feed
// the deterministic merge-back into the shared session (see runPlanWave); err
// captures the last ErrorEvent exactly like the serial runOnePlanStep -- NOT
// SubAgent.RunSync, which swallows an error when any text was produced and would
// desync parallel failure semantics from serial.
//
// Concurrency: forked sub-agents share the parent's tool registry but their
// runLoop never calls SetMessageID (only the main loop does), so concurrent
// workers are race-free on the shared tools (proven by
// plan_parallel_spike_test.go). model (when non-empty) targets the runtime run
// model via SubAgentConfig.Model.
//
// runOnePlanStepForked 在自己 fork 的 sub-agent (隔离 runLoop) 里执行单步, 用 wave
// 的共享历史快照 seed 故串联前面 wave 的上下文. 返回 (prompt, reply, err): prompt +
// reply 喂给回共享 session 的确定 merge (见 runPlanWave); err 与串行 runOnePlanStep
// 一样捕获最后一个 ErrorEvent -- 不用 SubAgent.RunSync (它在产生了文本时吞错, 会让
// 并行的失败语义与串行失同步).
//
// 并发: fork 的 sub-agent 共享父工具注册表, 但其 runLoop 从不调 SetMessageID
// (只主循环调), 故并发 worker 在共享工具上 race-free (plan_parallel_spike_test.go
// 证). model 非空时经 SubAgentConfig.Model 锁定运行时 run model.
func (e *Engine) runOnePlanStepForked(ctx context.Context, history []query.Message, step PlanStep, model string) (prompt string, reply string, err error) {
	prompt = buildPlanStepPrompt(step)
	sa := SpawnSubAgent(e, &SubAgentConfig{
		Description:     step.ID,
		Model:           model,
		HistoryMessages: history,
	})

	var sb strings.Builder
	for evt := range sa.Run(ctx, prompt) {
		switch ev := evt.(type) {
		case *TextEvent:
			sb.WriteString(ev.Text)
		case *ErrorEvent:
			err = ev.Err
		}
	}
	return prompt, sb.String(), err
}

// normalizePlanStepIDs guarantees every step carries a unique non-empty ID.
// Steps with a blank ID get a generated "step-N"; a (rare) duplicate non-empty
// ID gets an "x" suffix until unique. Without this, PlanProgress's ID-keyed
// index and the queue's StepStatuses map silently collapse colliding keys --
// two blank-ID steps would map to one slot and progress tracking would break.
//
// Note on Deps: renaming only happens for blank IDs, where the model also
// supplied no Deps referencing them (a model that writes Deps necessarily names
// the steps it references). So normalization does not silently break a declared
// dependency graph.
//
// normalizePlanStepIDs 确保每步有唯一非空 ID. 空 ID 的步骤补生成 "step-N"; (罕见)
// 重复非空 ID 加 "x" 后缀直到唯一. 没有它, PlanProgress 的 ID 索引和队列
// StepStatuses map 会静默折叠相撞的 key -- 两个空 ID 步骤映射到同一槽, 进度追踪
// 就坏了.
//
// Deps 说明: 仅在 ID 为空时重命名, 而那种情况模型也没写引用它们的 Deps (会写 Deps
// 的模型必然给所引用步骤起了名). 故规范化不会静默破坏已声明的依赖图.
func normalizePlanStepIDs(steps []PlanStep) []PlanStep {
	seen := make(map[string]bool, len(steps))
	out := make([]PlanStep, len(steps))
	for i, s := range steps {
		id := strings.TrimSpace(s.ID)
		if id == "" {
			id = fmt.Sprintf("step-%d", i+1)
		}
		for seen[id] {
			id += "x"
		}
		seen[id] = true
		s.ID = id
		out[i] = s
	}
	return out
}