package squad import ( "context" "fmt" "strings" "time" ) // taskboard.go — task_board 拓扑驱动(Q2 WP4)。 // 形态:计划步骤 = 板上任务(Needs = 依赖门);执行语义是队列化的 // claim/lease/complete 三原语——工作者认领任务(带租约时限)、执行、完成; // 进程死亡留下的孤儿租约(claimed 且租约过期)在驱动重入时再武装回 open // (Attempts 保留,超限由步骤级重试护栏兜底)。全部任务收口后走标准 // finalize(交付物/质检/降级判定与图拓扑同一收口面)。 // 挂起(HITL)安全:claimed 且步骤未完的任务在重入时续跑,不重复认领。 func init() { _ = RegisterTopologyDriver(TopologyTaskBoard) } // DefaultTaskLeaseSeconds 任务租约缺省时限(秒)。 const DefaultTaskLeaseSeconds = 600 // boardSkipReason 依赖门无法满足(上游 failed/skipped)时的跳过原因。 const boardSkipReason = "task_board: 依赖未满足(上游失败或跳过)" // BoardTask 是任务板上一条任务的队列状态。 type BoardTask struct { ID string `json:"id"` Status string `json:"status"` // open / claimed / done / failed / skipped ClaimedBy string `json:"claimed_by,omitempty"` // LeaseExpires 认领租约到期时刻;过期未 complete = 孤儿租约,重入再武装。 LeaseExpires time.Time `json:"lease_expires,omitempty"` // Attempts 认领次数(孤儿再武装累加;执行重试由步骤级护栏负责)。 Attempts int `json:"attempts,omitempty"` } // runTaskBoardDriver task_board 拓扑入口(幂等可恢复)。 func (e *SquadEngine) runTaskBoardDriver(ctx context.Context, st *RunState) (*RunResult, error) { drv := &flowDriver{opts: e.deps} e.initBoard(st) e.rearmOrphanLeases(st) leaseTTL := DefaultTaskLeaseSeconds if v := st.Meta.Metadata["task_board_lease_seconds"]; v != "" { if n := parseIntDefault(v, DefaultTaskLeaseSeconds); n > 0 { leaseTTL = n } } for { idx, held := e.boardHeldTask(st) if !held { var ok bool idx, ok = e.boardClaimNext(st, leaseTTL) if !ok { // 无可认领:把依赖已死(上游 failed/skipped)的 open 任务级联 // 置 skipped;再无变化则收口。 if !e.boardSkipDeadlocked(st) { break } continue } } out, gate, err := e.boardRunTask(ctx, drv, st, idx) if err != nil { return nil, err } if gate != nil { if out == nil { out = heldResult(st, gate) } return out, nil } if out != nil { // 关键任务失败:fail-closed 传播(failStep 已置运行失败) return out, nil } e.boardSettleTask(st, idx) } return e.boardFinish(ctx, drv, st) } // initBoard 首次进入时从计划步骤派生任务板(恢复重入直读持久化状态)。 func (e *SquadEngine) initBoard(st *RunState) { st.mu.Lock() defer st.mu.Unlock() if len(st.Board) > 0 { return } for _, s := range st.Plan.Workflow.Steps { status := "open" switch s.Status { case "completed": status = "done" case "failed": status = "failed" case "skipped": status = "skipped" } st.Board = append(st.Board, BoardTask{ID: s.ID, Status: status}) } _ = st.save() } // rearmOrphanLeases 孤儿租约再武装:claimed 且租约过期 → open(进程死亡 // 恢复路径;Attempts 保留作观测)。 func (e *SquadEngine) rearmOrphanLeases(st *RunState) { now := time.Now() st.mu.Lock() var rearmed []string for i := range st.Board { t := &st.Board[i] if t.Status == "claimed" && !t.LeaseExpires.IsZero() && t.LeaseExpires.Before(now) { t.Status = "open" t.ClaimedBy = "" rearmed = append(rearmed, t.ID) } } changed := len(rearmed) > 0 if changed { _ = st.save() } st.mu.Unlock() if changed { (&flowDriver{opts: e.deps}).emit(st, Event{Type: EvPhase, Text: fmt.Sprintf("task_board:孤儿租约再武装 %s", strings.Join(rearmed, ","))}) } } // boardHeldTask 返回 claimed 且步骤未完的任务(HITL 挂起/中断重入的续跑点)。 func (e *SquadEngine) boardHeldTask(st *RunState) (int, bool) { for ti := range st.Board { t := &st.Board[ti] if t.Status != "claimed" { continue } if idx, s := e.boardStep(st, t.ID); idx >= 0 && (s.Status == "pending" || s.Status == "running") { return idx, true } } return -1, false } // boardClaimNext 认领下一个可执行任务(open 且依赖门全 done),返回步骤下标。 func (e *SquadEngine) boardClaimNext(st *RunState, leaseTTL int) (int, bool) { st.mu.Lock() defer st.mu.Unlock() for ti := range st.Board { t := &st.Board[ti] if t.Status != "open" { continue } if !e.boardDepsDoneLocked(st, t.ID) { continue } idx, s := e.boardStepLocked(st, t.ID) if idx < 0 { continue } t.Status = "claimed" t.ClaimedBy = s.Role t.Attempts++ t.LeaseExpires = time.Now().Add(time.Duration(leaseTTL) * time.Second) _ = st.save() (&flowDriver{opts: e.deps}).emit(st, Event{Type: EvPhase, StepID: t.ID, Role: s.Role, Text: fmt.Sprintf("task_board:%s 认领 %s(租约 %ds,第 %d 次)", s.Role, t.ID, leaseTTL, t.Attempts)}) return idx, true } return -1, false } // boardSkipDeadlocked 把依赖已死(上游 failed/skipped)的 open 任务级联置 // skipped;返回是否有变化(有则继续循环级联)。 func (e *SquadEngine) boardSkipDeadlocked(st *RunState) bool { st.mu.Lock() changed := false for ti := range st.Board { t := &st.Board[ti] if t.Status != "open" { continue } if e.boardDepsDeadLocked(st, t.ID) { t.Status = "skipped" if idx, s := e.boardStepLocked(st, t.ID); idx >= 0 { s.Status = "skipped" s.Error = boardSkipReason st.Plan.Workflow.Steps[idx] = *s } changed = true } } if changed { _ = st.save() } st.mu.Unlock() return changed } // boardRunTask 执行已认领任务:过门 → runOneStep。 func (e *SquadEngine) boardRunTask(ctx context.Context, drv *flowDriver, st *RunState, idx int) (*RunResult, *HitlGate, error) { out, gate, err := drv.gateBeforeRun(ctx, st, idx) if err != nil || gate != nil || out != nil { return out, gate, err } return drv.runOneStep(ctx, st, idx) } // boardSettleTask 任务结算:步骤 completed → done;failed(非关键降级)→ failed。 func (e *SquadEngine) boardSettleTask(st *RunState, idx int) { step := st.Plan.Workflow.Steps[idx] status := "done" if step.Status == "failed" { status = "failed" } st.mu.Lock() for i := range st.Board { if st.Board[i].ID == step.ID { st.Board[i].Status = status st.Board[i].LeaseExpires = time.Time{} } } _ = st.save() st.mu.Unlock() (&flowDriver{opts: e.deps}).emit(st, Event{Type: EvPhase, StepID: step.ID, Role: step.Role, Text: fmt.Sprintf("task_board:%s → %s", step.ID, status)}) } // boardFinish 全部任务收口后走标准 finalize(交付物/质检/降级与图拓扑同面)。 // 仍有 open/claimed 任务 = 依赖不可满足,fail-closed 不静默收口。 func (e *SquadEngine) boardFinish(ctx context.Context, drv *flowDriver, st *RunState) (*RunResult, error) { done, total := 0, len(st.Board) for _, t := range st.Board { switch t.Status { case "done": done++ case "open", "claimed": return e.driverFail(st, fmt.Sprintf("task_board: 任务 %s 无法收口(依赖不可满足)", t.ID)), nil } } drv.emit(st, Event{Type: EvPhase, Text: fmt.Sprintf("task_board:收口(done %d/%d)", done, total)}) out, gate, err := drv.finalize(ctx, st) if err != nil { return nil, err } if gate != nil { return heldResult(st, gate), nil } return out, nil } // boardStep 按任务 ID 找步骤(只读副本)。 func (e *SquadEngine) boardStep(st *RunState, id string) (int, *StepSpec) { st.mu.Lock() defer st.mu.Unlock() return e.boardStepLocked(st, id) } // boardStepLocked 按任务 ID 找步骤(调用方持锁)。返回下标与副本指针。 func (e *SquadEngine) boardStepLocked(st *RunState, id string) (int, *StepSpec) { for i := range st.Plan.Workflow.Steps { if st.Plan.Workflow.Steps[i].ID == id { s := st.Plan.Workflow.Steps[i] return i, &s } } return -1, nil } // boardDepsDoneLocked 依赖门:Needs 全部 done(调用方持锁)。 func (e *SquadEngine) boardDepsDoneLocked(st *RunState, id string) bool { needs := e.boardNeedsLocked(st, id) for _, dep := range needs { found := false for i := range st.Board { if st.Board[i].ID == dep { found = true if st.Board[i].Status != "done" { return false } } } if !found { return false // 依赖不在板上 = fail-closed 不认领 } } return true } // boardDepsDeadLocked 依赖已死:任一 Need 为 failed/skipped(级联跳过依据)。 func (e *SquadEngine) boardDepsDeadLocked(st *RunState, id string) bool { for _, dep := range e.boardNeedsLocked(st, id) { for i := range st.Board { if st.Board[i].ID == dep && (st.Board[i].Status == "failed" || st.Board[i].Status == "skipped") { return true } } } return false } func (e *SquadEngine) boardNeedsLocked(st *RunState, id string) []string { for i := range st.Plan.Workflow.Steps { if st.Plan.Workflow.Steps[i].ID == id { return st.Plan.Workflow.Steps[i].Needs } } return nil }