You've already forked agentic-coding-workflow
501 lines
15 KiB
Go
501 lines
15 KiB
Go
// service.go 实现 orchestrator.Service：StartRun / GetRun / ListRuns / Pause /
|
|
// Resume / Cancel / RequestSubtask / EnqueueStepRedrive。step 的实际驱动在
|
|
// step_handler.go（jobs worker 内执行）。
|
|
package orchestrator
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"log/slog"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
"github.com/jackc/pgx/v5"
|
|
|
|
"github.com/yan1h/agent-coding-workflow/internal/infra/errs"
|
|
"github.com/yan1h/agent-coding-workflow/internal/jobs"
|
|
"github.com/yan1h/agent-coding-workflow/internal/project"
|
|
)
|
|
|
|
// JobEnqueuer 是 service 入队 orchestrator.step job 所需的窄接口（jobs.Repository 满足）。
|
|
type JobEnqueuer interface {
|
|
Enqueue(ctx context.Context, typ jobs.JobType, payload json.RawMessage, scheduledAt time.Time, maxAttempts int32) (*jobs.Job, error)
|
|
}
|
|
|
|
// ProjectACL 是 StartRun 校验 owner 对项目写权限的窄接口（app 层用 project 的
|
|
// owner+admin ACL 适配）。
|
|
type ProjectACL interface {
|
|
CanRead(ctx context.Context, userID uuid.UUID, isAdmin bool, projectID uuid.UUID) (bool, error)
|
|
CanWrite(ctx context.Context, userID uuid.UUID, isAdmin bool, projectID uuid.UUID) (bool, error)
|
|
}
|
|
|
|
// ServiceDeps 是 service 的依赖集合。
|
|
type ServiceDeps struct {
|
|
Repo Repository
|
|
PM PMAccess
|
|
ACL ProjectACL
|
|
Jobs JobEnqueuer
|
|
Sessions sessionTerminator
|
|
Defaults map[project.Phase]uuid.UUID
|
|
Config Config
|
|
Audit AuditRecorder
|
|
Log *slog.Logger
|
|
}
|
|
|
|
// sessionTerminator 终止某 acp session（Pause/Cancel 用）。acp.SessionService 经 adapter 适配。
|
|
type sessionTerminator interface {
|
|
TerminateSession(ctx context.Context, ownerID uuid.UUID, isAdmin bool, sessionID uuid.UUID) error
|
|
}
|
|
|
|
type service struct {
|
|
repo Repository
|
|
pm PMAccess
|
|
acl ProjectACL
|
|
jobs JobEnqueuer
|
|
sessions sessionTerminator
|
|
defaults map[project.Phase]uuid.UUID
|
|
cfg Config
|
|
audit AuditRecorder
|
|
log *slog.Logger
|
|
}
|
|
|
|
// NewService 构造编排器 Service。
|
|
func NewService(d ServiceDeps) Service {
|
|
log := d.Log
|
|
if log == nil {
|
|
log = slog.Default()
|
|
}
|
|
return &service{
|
|
repo: d.Repo, pm: d.PM, acl: d.ACL, jobs: d.Jobs, sessions: d.Sessions,
|
|
defaults: d.Defaults, cfg: d.Config, audit: d.Audit, log: log,
|
|
}
|
|
}
|
|
|
|
var _ Service = (*service)(nil)
|
|
|
|
// EnqueueStep 实现 stepEnqueuer：入队一个 orchestrator.step job。
|
|
func (s *service) EnqueueStep(ctx context.Context, runID, stepID uuid.UUID, scheduledAt time.Time) error {
|
|
payload, err := json.Marshal(stepPayload{RunID: runID, StepID: stepID})
|
|
if err != nil {
|
|
return errs.Wrap(err, errs.CodeInternal, "marshal step payload")
|
|
}
|
|
// maxAttempts: jobs 层的尝试计数仅用于 reaper/worker 调度；编排器自身以 step.attempt
|
|
// + MaxAttemptsPerPhase 控制重试上限。给一个较大的 jobs maxAttempts，让 worker 在
|
|
// 编排器返回非 Permanent error 时继续重试，直到编排器返回 Permanent。
|
|
maxAttempts := int32(s.effectiveMaxAttempts() + 2)
|
|
if _, err := s.jobs.Enqueue(ctx, JobTypeOrchestratorStep, payload, scheduledAt, maxAttempts); err != nil {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *service) effectiveMaxAttempts() int {
|
|
if s.cfg.MaxAttemptsPerPhase > 0 {
|
|
return s.cfg.MaxAttemptsPerPhase
|
|
}
|
|
return 3
|
|
}
|
|
|
|
// StartRun 启动一次编排运行：解析项目/需求 + ACL 校验 + 在一个事务内插入 run + 第一个
|
|
// step，事务提交后入队第一个 orchestrator.step job。每需求只允许一个活跃 run（DB 唯一
|
|
// 部分索引兜底，命中翻译为 409）。
|
|
func (s *service) StartRun(ctx context.Context, c Caller, in StartRunInput) (*Run, error) {
|
|
proj, err := s.pm.GetProjectBySlug(ctx, in.ProjectSlug)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
// 写权限校验：owner 须对项目可写（不依赖 admin 放行）。
|
|
if s.acl != nil {
|
|
ok, aerr := s.acl.CanWrite(ctx, c.UserID, c.IsAdmin, proj.ID)
|
|
if aerr != nil {
|
|
return nil, aerr
|
|
}
|
|
if !ok {
|
|
return nil, errs.New(errs.CodeForbidden, "no write access to project")
|
|
}
|
|
}
|
|
|
|
req, err := s.pm.GetRequirementByNumber(ctx, proj.ID, in.RequirementNumber)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if req.WorkspaceID == nil {
|
|
return nil, errs.New(errs.CodeInvalidInput,
|
|
"requirement has no workspace; link a workspace before starting a run")
|
|
}
|
|
if req.Status == project.StatusClosed {
|
|
return nil, errs.New(errs.CodeRequirementClosed, "requirement is closed")
|
|
}
|
|
|
|
startPhase := in.StartPhase
|
|
if startPhase == "" {
|
|
startPhase = req.Phase
|
|
}
|
|
if !startPhase.IsValid() {
|
|
return nil, errs.New(errs.CodeInvalidInput, "invalid start phase")
|
|
}
|
|
|
|
// 规范化 config 默认值。
|
|
cfg := in.Config
|
|
if cfg.MaxAttemptsPerPhase <= 0 {
|
|
cfg.MaxAttemptsPerPhase = s.effectiveMaxAttempts()
|
|
}
|
|
if cfg.MaxDepth <= 0 {
|
|
cfg.MaxDepth = s.cfg.MaxDepth
|
|
}
|
|
if cfg.FanOutLimit <= 0 {
|
|
cfg.FanOutLimit = s.cfg.FanOutLimit
|
|
}
|
|
|
|
run := &Run{
|
|
ID: uuid.New(),
|
|
ProjectID: proj.ID,
|
|
RequirementID: req.ID,
|
|
WorkspaceID: *req.WorkspaceID,
|
|
OwnerID: proj.OwnerID,
|
|
Status: RunPending,
|
|
CurrentPhase: startPhase,
|
|
Config: cfg,
|
|
}
|
|
|
|
var (
|
|
outRun *Run
|
|
outStep *Step
|
|
)
|
|
txErr := s.repo.InTx(ctx, func(txCtx context.Context, tx pgx.Tx) error {
|
|
txRepo := s.repo.WithTx(tx)
|
|
inserted, ierr := txRepo.InsertRun(txCtx, run)
|
|
if ierr != nil {
|
|
return ierr
|
|
}
|
|
outRun = inserted
|
|
step, serr := buildPhaseStep(txCtx, txRepo, s.pm, inserted, startPhase, nil, 0, s.defaults)
|
|
if serr != nil {
|
|
return serr
|
|
}
|
|
outStep = step
|
|
return nil
|
|
})
|
|
if txErr != nil {
|
|
if IsUniqueViolation(txErr) {
|
|
return nil, errs.New(errs.CodeConflict, "an active orchestrator run already exists for this requirement")
|
|
}
|
|
return nil, txErr
|
|
}
|
|
|
|
// 事务已提交：入队第一个 step job。
|
|
if eerr := s.EnqueueStep(ctx, outRun.ID, outStep.ID, time.Now().UTC()); eerr != nil {
|
|
// 入队失败：标 run 失败（step 已落库但永不会被驱动）。
|
|
msg := eerr.Error()
|
|
_, _ = s.repo.UpdateRunStatus(ctx, outRun.ID, RunFailed, &msg)
|
|
return nil, eerr
|
|
}
|
|
|
|
s.recordAudit(ctx, c.UserID, "orchestrator.run.start", outRun.ID.String(), map[string]any{
|
|
"project_slug": in.ProjectSlug,
|
|
"requirement": in.RequirementNumber,
|
|
"start_phase": string(startPhase),
|
|
"first_step_id": outStep.ID.String(),
|
|
"execution_owner_id": outRun.OwnerID.String(),
|
|
})
|
|
return outRun, nil
|
|
}
|
|
|
|
func (s *service) GetRun(ctx context.Context, c Caller, id uuid.UUID) (*Run, error) {
|
|
run, err := s.repo.GetRun(ctx, id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if !s.canReadRun(ctx, c, run) {
|
|
return nil, errs.New(errs.CodeNotFound, "orchestrator run not found")
|
|
}
|
|
return run, nil
|
|
}
|
|
|
|
func (s *service) ListRuns(ctx context.Context, c Caller, f RunFilter) ([]*Run, error) {
|
|
runs, err := s.repo.ListRuns(ctx, f)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if c.IsAdmin {
|
|
return runs, nil
|
|
}
|
|
out := make([]*Run, 0, len(runs))
|
|
for _, r := range runs {
|
|
if s.canReadRun(ctx, c, r) {
|
|
out = append(out, r)
|
|
}
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// Pause 把活跃 run 置 paused 并终止其当前活跃 session（停止驱动；已入队 step job 在
|
|
// Handle 时见 run 非活跃即 no-op）。
|
|
func (s *service) Pause(ctx context.Context, c Caller, id uuid.UUID) error {
|
|
run, err := s.GetRun(ctx, c, id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !run.Status.IsActive() {
|
|
return errs.New(errs.CodeConflict, "run is not active")
|
|
}
|
|
if _, uerr := s.repo.UpdateRunStatus(ctx, run.ID, RunPaused, nil); uerr != nil {
|
|
return uerr
|
|
}
|
|
s.terminateActiveSessions(ctx, run)
|
|
s.recordAudit(ctx, c.UserID, "orchestrator.run.pause", run.ID.String(), nil)
|
|
return nil
|
|
}
|
|
|
|
// Resume 把 paused run 置回 running 并重新入队当前阶段的一个 step（让驱动继续）。
|
|
func (s *service) Resume(ctx context.Context, c Caller, id uuid.UUID) error {
|
|
run, err := s.GetRun(ctx, c, id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if run.Status != RunPaused {
|
|
return errs.New(errs.CodeConflict, "run is not paused")
|
|
}
|
|
if _, uerr := s.repo.UpdateRunStatus(ctx, run.ID, RunRunning, nil); uerr != nil {
|
|
return uerr
|
|
}
|
|
// 找当前阶段最近一个非终态 step 重新入队；若没有则新建一个。
|
|
step, serr := s.latestResumableStep(ctx, run)
|
|
if serr != nil {
|
|
return serr
|
|
}
|
|
if step == nil {
|
|
built, berr := buildPhaseStep(ctx, s.repo, s.pm, run, run.CurrentPhase, nil, 0, s.defaults)
|
|
if berr != nil {
|
|
return berr
|
|
}
|
|
step = built
|
|
}
|
|
if eerr := s.EnqueueStep(ctx, run.ID, step.ID, time.Now().UTC()); eerr != nil {
|
|
return eerr
|
|
}
|
|
s.recordAudit(ctx, c.UserID, "orchestrator.run.resume", run.ID.String(), nil)
|
|
return nil
|
|
}
|
|
|
|
// Cancel 把 run 置 canceled 并终止其活跃 session。
|
|
func (s *service) Cancel(ctx context.Context, c Caller, id uuid.UUID) error {
|
|
run, err := s.GetRun(ctx, c, id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if run.Status.IsTerminal() {
|
|
return errs.New(errs.CodeConflict, "run already terminal")
|
|
}
|
|
if _, uerr := s.repo.UpdateRunStatus(ctx, run.ID, RunCanceled, nil); uerr != nil {
|
|
return uerr
|
|
}
|
|
s.terminateActiveSessions(ctx, run)
|
|
s.recordAudit(ctx, c.UserID, "orchestrator.run.cancel", run.ID.String(), nil)
|
|
return nil
|
|
}
|
|
|
|
// RequestSubtask 让运行中的 agent 为其当前 step 派生子 step。在一个事务内校验
|
|
// depth<MaxDepth 与 active children<FanOutLimit（避免并发竞态突破上限），插入子 step，
|
|
// 事务提交后入队其 job。
|
|
func (s *service) RequestSubtask(ctx context.Context, parentACPSessionID uuid.UUID, in SubtaskInput) (*Step, error) {
|
|
parent, err := s.repo.GetStepBySession(ctx, parentACPSessionID)
|
|
if err != nil {
|
|
return nil, errs.New(errs.CodeForbidden, "calling session is not an orchestrator step")
|
|
}
|
|
run, err := s.repo.GetRun(ctx, parent.RunID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if !run.Status.IsActive() {
|
|
return nil, errs.New(errs.CodeConflict, "run is not active")
|
|
}
|
|
|
|
phase := in.Phase
|
|
if phase == "" {
|
|
phase = parent.Phase
|
|
}
|
|
if !phase.IsValid() {
|
|
return nil, errs.New(errs.CodeInvalidInput, "invalid subtask phase")
|
|
}
|
|
|
|
maxDepth := run.Config.MaxDepth
|
|
if maxDepth <= 0 {
|
|
maxDepth = s.cfg.MaxDepth
|
|
}
|
|
fanOut := run.Config.FanOutLimit
|
|
if fanOut <= 0 {
|
|
fanOut = s.cfg.FanOutLimit
|
|
}
|
|
childDepth := parent.Depth + 1
|
|
if maxDepth > 0 && childDepth > maxDepth {
|
|
return nil, errs.New(errs.CodeForbidden, "subtask depth limit reached")
|
|
}
|
|
|
|
akID := uuid.Nil
|
|
if in.AgentKindID != nil {
|
|
akID = *in.AgentKindID
|
|
}
|
|
|
|
var outStep *Step
|
|
txErr := s.repo.InTx(ctx, func(txCtx context.Context, tx pgx.Tx) error {
|
|
txRepo := s.repo.WithTx(tx)
|
|
// 在同一事务内计活跃子 step，避免并发突破 fan-out。
|
|
n, cerr := txRepo.CountActiveChildSteps(txCtx, parent.ID)
|
|
if cerr != nil {
|
|
return cerr
|
|
}
|
|
if fanOut > 0 && int(n) >= fanOut {
|
|
return errs.New(errs.CodeForbidden, "subtask fan-out limit reached")
|
|
}
|
|
|
|
// 解析 agent-kind：显式给定优先，否则走 planner（config/defaults）。
|
|
resolvedAK := akID
|
|
if resolvedAK == uuid.Nil {
|
|
id, perr := PhaseAgentKind(run.Config, phase, s.defaults)
|
|
if perr != nil {
|
|
return perr
|
|
}
|
|
resolvedAK = id
|
|
}
|
|
|
|
req, rerr := s.pm.GetRequirementByID(txCtx, run.RequirementID)
|
|
if rerr != nil {
|
|
return rerr
|
|
}
|
|
artifacts, _ := s.pm.ListLatestArtifacts(txCtx, run.RequirementID)
|
|
prompt := in.Prompt
|
|
if prompt == "" {
|
|
prompt = BuildPrompt(phase, req, artifacts, run.Config.PromptOverrides[phase])
|
|
}
|
|
parentID := parent.ID
|
|
step := &Step{
|
|
ID: uuid.New(),
|
|
RunID: run.ID,
|
|
Phase: phase,
|
|
Attempt: 1,
|
|
ParentStepID: &parentID,
|
|
Depth: childDepth,
|
|
Status: StepPending,
|
|
AgentKindID: resolvedAK,
|
|
Prompt: prompt,
|
|
}
|
|
inserted, ierr := txRepo.InsertStep(txCtx, step)
|
|
if ierr != nil {
|
|
return ierr
|
|
}
|
|
outStep = inserted
|
|
return nil
|
|
})
|
|
if txErr != nil {
|
|
return nil, txErr
|
|
}
|
|
|
|
if eerr := s.EnqueueStep(ctx, run.ID, outStep.ID, time.Now().UTC()); eerr != nil {
|
|
return nil, eerr
|
|
}
|
|
s.recordAudit(ctx, run.OwnerID, "orchestrator.subtask.request", outStep.ID.String(), map[string]any{
|
|
"parent_step_id": parent.ID.String(),
|
|
"phase": string(phase),
|
|
"depth": childDepth,
|
|
})
|
|
return outStep, nil
|
|
}
|
|
|
|
// EnqueueStepRedrive 由 acp supervisor onExit 的注入回调间接调用：把崩溃 session 的
|
|
// step 重新入队（短延迟，给 worktree 释放留窗口）。
|
|
func (s *service) EnqueueStepRedrive(ctx context.Context, stepID uuid.UUID, reason string) {
|
|
step, err := s.repo.GetStep(ctx, stepID)
|
|
if err != nil {
|
|
s.log.Warn("orchestrator.redrive.step_missing", "step_id", stepID, "err", err.Error())
|
|
return
|
|
}
|
|
switch step.Status {
|
|
case StepSucceeded, StepFailed, StepDead:
|
|
return // 已终态：无需再驱动
|
|
case StepSpawning, StepRunning, StepAwaitingGate:
|
|
// 仍有活跃 job 在驱动本 step：其 driveTurn 会因 session 崩溃触发 relayCtx 取消而
|
|
// 返回错误 → 走 worker 重试（driveTurn 是 spawn-or-resume，会在同一 worktree 续跑）；
|
|
// worker 进程整体崩溃的情形由 job_reaper 重新租约兜底。此处若再入队会让同一 step
|
|
// 产生两条 job、双重驱动（两个 agent 会话）。故跳过，恢复只走活跃 job 这一条路径。
|
|
s.log.Info("orchestrator.redrive.skip_active",
|
|
"step_id", stepID, "status", step.Status, "reason", reason)
|
|
return
|
|
}
|
|
run, err := s.repo.GetRun(ctx, step.RunID)
|
|
if err != nil || !run.Status.IsActive() {
|
|
return // run 不活跃：不再驱动
|
|
}
|
|
// 短延迟入队，让 onExit 的 worktree 释放先落定。
|
|
if eerr := s.EnqueueStep(ctx, run.ID, step.ID, time.Now().UTC().Add(2*time.Second)); eerr != nil {
|
|
s.log.Warn("orchestrator.redrive.enqueue_failed", "step_id", stepID, "err", eerr.Error())
|
|
return
|
|
}
|
|
s.recordAudit(ctx, run.OwnerID, "orchestrator.step.redrive", step.ID.String(), map[string]any{
|
|
"reason": reason,
|
|
})
|
|
}
|
|
|
|
// ===== helpers =====
|
|
|
|
// latestResumableStep 返回当前阶段最近一个非终态 step（用于 Resume 重新入队）。
|
|
func (s *service) latestResumableStep(ctx context.Context, run *Run) (*Step, error) {
|
|
steps, err := s.repo.ListStepsByRun(ctx, run.ID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var found *Step
|
|
for _, st := range steps {
|
|
if st.Phase != run.CurrentPhase {
|
|
continue
|
|
}
|
|
switch st.Status {
|
|
case StepSucceeded, StepFailed, StepDead:
|
|
continue
|
|
}
|
|
found = st // ListStepsByRun 按 created_at 升序：取最后一个匹配
|
|
}
|
|
return found, nil
|
|
}
|
|
|
|
// terminateActiveSessions 终止 run 下当前活跃 step 的 session（best-effort）。
|
|
func (s *service) terminateActiveSessions(ctx context.Context, run *Run) {
|
|
if s.sessions == nil {
|
|
return
|
|
}
|
|
steps, err := s.repo.ListStepsByRun(ctx, run.ID)
|
|
if err != nil {
|
|
return
|
|
}
|
|
for _, st := range steps {
|
|
if st.ACPSessionID == nil {
|
|
continue
|
|
}
|
|
switch st.Status {
|
|
case StepSpawning, StepRunning, StepAwaitingGate:
|
|
if terr := s.sessions.TerminateSession(ctx, run.OwnerID, false, *st.ACPSessionID); terr != nil {
|
|
s.log.Warn("orchestrator.terminate_session_failed",
|
|
"session_id", *st.ACPSessionID, "err", terr.Error())
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *service) recordAudit(ctx context.Context, userID uuid.UUID, action, targetID string, meta map[string]any) {
|
|
if s.audit == nil {
|
|
return
|
|
}
|
|
s.audit.Record(context.WithoutCancel(ctx), userID, action, targetID, meta)
|
|
}
|
|
|
|
func (s *service) canReadRun(ctx context.Context, c Caller, run *Run) bool {
|
|
if c.IsAdmin || run.OwnerID == c.UserID {
|
|
return true
|
|
}
|
|
if s.acl == nil {
|
|
return false
|
|
}
|
|
ok, err := s.acl.CanRead(ctx, c.UserID, c.IsAdmin, run.ProjectID)
|
|
return err == nil && ok
|
|
}
|