You've already forked agentic-coding-workflow
108 lines
4.0 KiB
Go
108 lines
4.0 KiB
Go
// runner.go 是回合驱动器：给定一个 Step，要么经 acp.SessionService.Create 以特权
|
|
// owner 身份创建新 session，要么在崩溃 session 的同一 worktree 上 ResumeSession，然后
|
|
// 经 acp.TurnRunner.SendPromptAndWait 阻塞跑完一个回合。封装重试/恢复策略。
|
|
package orchestrator
|
|
|
|
import (
|
|
"context"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
|
|
"github.com/yan1h/agent-coding-workflow/internal/acp"
|
|
"github.com/yan1h/agent-coding-workflow/internal/infra/errs"
|
|
)
|
|
|
|
// ACPSessions 是 runner 创建 session 所需的窄接口（acp.SessionService 满足）。
|
|
type ACPSessions interface {
|
|
Create(ctx context.Context, c acp.Caller, in acp.CreateSessionInput) (*acp.Session, error)
|
|
}
|
|
|
|
// ACPSessionLookup 是 runner 查找 step 已有 session 所需的窄接口（acp.Repository 满足）。
|
|
type ACPSessionLookup interface {
|
|
GetSessionByStepID(ctx context.Context, stepID uuid.UUID) (*acp.Session, error)
|
|
GetCrashedSessionForResume(ctx context.Context, stepID uuid.UUID) (*acp.Session, error)
|
|
}
|
|
|
|
// turnResult 是一次回合驱动的结果。
|
|
type turnResult struct {
|
|
SessionID uuid.UUID
|
|
StopReason string
|
|
}
|
|
|
|
// runner 封装 spawn-or-resume + SendPromptAndWait。
|
|
type runner struct {
|
|
sessions ACPSessions
|
|
turns acp.TurnRunner
|
|
lookup ACPSessionLookup
|
|
turnTimeout time.Duration
|
|
}
|
|
|
|
// NewRunner 构造回合驱动器（由 app.go 装配 StepHandler 时调用）。turnTimeout<=0 时
|
|
// 退回 10 分钟默认值。
|
|
func NewRunner(sessions ACPSessions, turns acp.TurnRunner, lookup ACPSessionLookup, turnTimeout time.Duration) *runner {
|
|
if turnTimeout <= 0 {
|
|
turnTimeout = 10 * time.Minute
|
|
}
|
|
return &runner{sessions: sessions, turns: turns, lookup: lookup, turnTimeout: turnTimeout}
|
|
}
|
|
|
|
// driveTurn 为 step 驱动一个回合：
|
|
//
|
|
// 1. 若该 step 已有崩溃/退出 session → ResumeSession（同 worktree）。
|
|
// 2. 否则若该 step 已有活跃 session → 复用（幂等：避免重试重复 spawn）。
|
|
// 3. 否则 → Create 一个新 session（特权 owner 身份；OrchestratorStepID 反向关联）。
|
|
//
|
|
// requirementNumber 用于把新 session 绑定到 req/<slug>-N worktree，使同一 run 的各
|
|
// 阶段 step 共享同一 worktree（resume 也据此回到原 worktree）。
|
|
//
|
|
// 然后 SendPromptAndWait（受 turnTimeout 约束）。返回 sessionID + 原始 stopReason。
|
|
func (r *runner) driveTurn(ctx context.Context, run *Run, step *Step, requirementNumber int) (turnResult, error) {
|
|
owner := acp.Caller{UserID: run.OwnerID, IsAdmin: false}
|
|
|
|
var sessionID uuid.UUID
|
|
|
|
// 1. 崩溃 session → resume（同一 worktree）。
|
|
if crashed, err := r.lookup.GetCrashedSessionForResume(ctx, step.ID); err == nil && crashed != nil {
|
|
resumed, rerr := r.turns.ResumeSession(ctx, owner, crashed.ID)
|
|
if rerr != nil {
|
|
return turnResult{}, rerr
|
|
}
|
|
sessionID = resumed.ID
|
|
} else if live, lerr := r.lookup.GetSessionByStepID(ctx, step.ID); lerr == nil && live != nil && live.Status.IsActive() {
|
|
// 2. 已有活跃 session → 复用。
|
|
sessionID = live.ID
|
|
} else {
|
|
// 3. 创建新 session（特权 owner；反向关联到 step）。
|
|
stepID := step.ID
|
|
in := acp.CreateSessionInput{
|
|
WorkspaceID: run.WorkspaceID,
|
|
AgentKindID: step.AgentKindID,
|
|
InitialPrompt: nil, // 由 SendPromptAndWait 单独发送，确保阻塞等待回合
|
|
OrchestratorStepID: &stepID,
|
|
}
|
|
if requirementNumber > 0 {
|
|
// 绑定到 req 分支 worktree，使同一 run 的各阶段共享同一 worktree。
|
|
rn := requirementNumber
|
|
in.RequirementNumber = &rn
|
|
}
|
|
created, cerr := r.sessions.Create(ctx, owner, in)
|
|
if cerr != nil {
|
|
return turnResult{}, cerr
|
|
}
|
|
sessionID = created.ID
|
|
}
|
|
|
|
if sessionID == uuid.Nil {
|
|
return turnResult{}, errs.New(errs.CodeInternal, "orchestrator: no session id after spawn/resume")
|
|
}
|
|
|
|
turnCtx, cancel := context.WithTimeout(ctx, r.turnTimeout)
|
|
defer cancel()
|
|
stop, err := r.turns.SendPromptAndWait(turnCtx, sessionID, step.Prompt)
|
|
if err != nil {
|
|
return turnResult{SessionID: sessionID}, err
|
|
}
|
|
return turnResult{SessionID: sessionID, StopReason: stop}, nil
|
|
}
|