Agent-Squad/squad/taskboard.go

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
}