You've already forked agentic-coding-workflow
308 lines
11 KiB
Go
308 lines
11 KiB
Go
package orchestrator
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
|
|
"github.com/yan1h/agent-coding-workflow/internal/acp"
|
|
"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"
|
|
)
|
|
|
|
type stepHandlerHarness struct {
|
|
repo *fakeRepo
|
|
pm *fakePM
|
|
turns *fakeTurnRunner
|
|
sess *fakeACPSessions
|
|
lookup *fakeLookup
|
|
jobs *fakeJobs
|
|
audit *fakeAudit
|
|
handler *StepHandler
|
|
run *Run
|
|
}
|
|
|
|
func newStepHandlerHarness(t *testing.T, gate PhaseGate, startPhase project.Phase, maxAttempts int) *stepHandlerHarness {
|
|
t.Helper()
|
|
repo := newFakeRepo()
|
|
wsID := uuid.New()
|
|
akID := uuid.New()
|
|
pm := &fakePM{
|
|
proj: &project.Project{ID: uuid.New(), Slug: "demo"},
|
|
req: &project.Requirement{ID: uuid.New(), Number: 3, Title: "T", WorkspaceID: &wsID, Phase: startPhase, Status: project.StatusOpen},
|
|
}
|
|
run := &Run{
|
|
ID: uuid.New(), ProjectID: pm.proj.ID, RequirementID: pm.req.ID,
|
|
WorkspaceID: wsID, OwnerID: uuid.New(), Status: RunRunning,
|
|
CurrentPhase: startPhase,
|
|
Config: RunConfig{MaxAttemptsPerPhase: maxAttempts, PhaseAgentKinds: allPhaseKinds(akID)},
|
|
}
|
|
repo.runs[run.ID] = run
|
|
|
|
turns := &fakeTurnRunner{stopReason: "end_turn"}
|
|
sess := &fakeACPSessions{}
|
|
lookup := newFakeLookup()
|
|
jq := &fakeJobs{}
|
|
audit := &fakeAudit{}
|
|
|
|
svc := NewService(ServiceDeps{
|
|
Repo: repo, PM: pm, ACL: fakeACL{allow: true}, Jobs: jq,
|
|
Defaults: allPhaseKinds(akID), Config: Config{MaxAttemptsPerPhase: maxAttempts}, Audit: audit,
|
|
})
|
|
|
|
h := NewStepHandler(StepHandlerDeps{
|
|
Repo: repo,
|
|
Runner: NewRunner(sess, turns, lookup, time.Minute),
|
|
Gate: gate,
|
|
PM: pm,
|
|
Enqueue: svc.(StepEnqueuer),
|
|
DefaultAgentKind: allPhaseKinds(akID),
|
|
Config: Config{MaxAttemptsPerPhase: maxAttempts},
|
|
Audit: audit,
|
|
})
|
|
|
|
return &stepHandlerHarness{repo: repo, pm: pm, turns: turns, sess: sess, lookup: lookup, jobs: jq, audit: audit, handler: h, run: run}
|
|
}
|
|
|
|
func allPhaseKinds(akID uuid.UUID) map[project.Phase]uuid.UUID {
|
|
m := map[project.Phase]uuid.UUID{}
|
|
for _, p := range project.AllPhases {
|
|
m[p] = akID
|
|
}
|
|
return m
|
|
}
|
|
|
|
func (h *stepHandlerHarness) seedStep(phase project.Phase) *Step {
|
|
step := &Step{
|
|
ID: uuid.New(), RunID: h.run.ID, Phase: phase, Attempt: 1,
|
|
Status: StepPending, AgentKindID: h.run.Config.PhaseAgentKinds[phase], Prompt: "do it",
|
|
}
|
|
h.repo.steps[step.ID] = step
|
|
return step
|
|
}
|
|
|
|
func jobFor(step *Step) jobs.Job {
|
|
payload, _ := json.Marshal(stepPayload{RunID: step.RunID, StepID: step.ID})
|
|
return jobs.Job{ID: uuid.New(), Type: JobTypeOrchestratorStep, Payload: payload}
|
|
}
|
|
|
|
func TestStepHandler_HappyAdvancesAndEnqueuesNext(t *testing.T) {
|
|
h := newStepHandlerHarness(t, fakeGate{result: GateResult{Pass: true}}, project.PhasePlanning, 3)
|
|
step := h.seedStep(project.PhasePlanning)
|
|
|
|
if err := h.handler.Handle(context.Background(), jobFor(step)); err != nil {
|
|
t.Fatalf("Handle: %v", err)
|
|
}
|
|
got, _ := h.repo.GetStep(context.Background(), step.ID)
|
|
if got.Status != StepSucceeded {
|
|
t.Errorf("step status = %s; want succeeded", got.Status)
|
|
}
|
|
if h.jobs.count() != 1 {
|
|
t.Errorf("expected 1 next-step enqueued; got %d", h.jobs.count())
|
|
}
|
|
if len(h.pm.phaseWrites) != 1 || h.pm.phaseWrites[0] != project.PhasePrototyping {
|
|
t.Errorf("expected phase advance to prototyping; got %v", h.pm.phaseWrites)
|
|
}
|
|
if !h.audit.has("orchestrator.step.advance") {
|
|
t.Error("expected step.advance audit")
|
|
}
|
|
}
|
|
|
|
func TestStepHandler_PhaseGateFailurePausesWithoutAdvancing(t *testing.T) {
|
|
h := newStepHandlerHarness(t, fakeGate{result: GateResult{Pass: true}}, project.PhaseAuditing, 3)
|
|
h.pm.phaseErr = errs.New(errs.CodePhaseGateFailed, "audit artifact not approved")
|
|
step := h.seedStep(project.PhaseAuditing)
|
|
|
|
if err := h.handler.Handle(context.Background(), jobFor(step)); err != nil {
|
|
t.Fatalf("Handle should complete job and pause run on external phase gate; got %v", err)
|
|
}
|
|
got, _ := h.repo.GetStep(context.Background(), step.ID)
|
|
if got.Status != StepAwaitingGate {
|
|
t.Errorf("step status = %s; want awaiting_gate", got.Status)
|
|
}
|
|
if got.Attempt != 1 {
|
|
t.Errorf("phase gate pause must not consume agent attempt; got %d", got.Attempt)
|
|
}
|
|
gotRun, _ := h.repo.GetRun(context.Background(), h.run.ID)
|
|
if gotRun.Status != RunPaused {
|
|
t.Errorf("run status = %s; want paused", gotRun.Status)
|
|
}
|
|
if gotRun.CurrentPhase != project.PhaseAuditing {
|
|
t.Errorf("run phase = %s; want auditing", gotRun.CurrentPhase)
|
|
}
|
|
if h.pm.req.Phase != project.PhaseAuditing {
|
|
t.Errorf("requirement phase = %s; want auditing", h.pm.req.Phase)
|
|
}
|
|
if h.jobs.count() != 0 {
|
|
t.Errorf("next step must not be enqueued when project gate blocks; got %d", h.jobs.count())
|
|
}
|
|
if !h.audit.has("orchestrator.step.phase_blocked") {
|
|
t.Error("expected phase_blocked audit")
|
|
}
|
|
}
|
|
|
|
func TestStepHandler_PhaseWriteFailureRetriesWithoutAdvancing(t *testing.T) {
|
|
h := newStepHandlerHarness(t, fakeGate{result: GateResult{Pass: true}}, project.PhasePlanning, 3)
|
|
h.pm.phaseErr = errors.New("db temporarily unavailable")
|
|
step := h.seedStep(project.PhasePlanning)
|
|
|
|
err := h.handler.Handle(context.Background(), jobFor(step))
|
|
if err == nil {
|
|
t.Fatal("expected retryable error")
|
|
}
|
|
if jobs.IsPermanent(err) {
|
|
t.Fatalf("expected retryable error; got permanent: %v", err)
|
|
}
|
|
got, _ := h.repo.GetStep(context.Background(), step.ID)
|
|
if got.Status != StepAwaitingGate {
|
|
t.Errorf("step status = %s; want awaiting_gate", got.Status)
|
|
}
|
|
if got.Attempt != 2 {
|
|
t.Errorf("attempt = %d; want 2", got.Attempt)
|
|
}
|
|
gotRun, _ := h.repo.GetRun(context.Background(), h.run.ID)
|
|
if gotRun.CurrentPhase != project.PhasePlanning {
|
|
t.Errorf("run phase = %s; want planning", gotRun.CurrentPhase)
|
|
}
|
|
if h.jobs.count() != 0 {
|
|
t.Errorf("next step must not be enqueued after failed phase write; got %d", h.jobs.count())
|
|
}
|
|
}
|
|
|
|
func TestStepHandler_AwaitingGateResumeDoesNotRerunAgent(t *testing.T) {
|
|
h := newStepHandlerHarness(t, fakeGate{result: GateResult{Pass: true}}, project.PhasePlanning, 3)
|
|
step := h.seedStep(project.PhasePlanning)
|
|
gate := GateResult{Pass: true, Reason: "ready"}
|
|
step.Status = StepAwaitingGate
|
|
step.GateResult = &gate
|
|
h.repo.steps[step.ID] = step
|
|
|
|
if err := h.handler.Handle(context.Background(), jobFor(step)); err != nil {
|
|
t.Fatalf("Handle: %v", err)
|
|
}
|
|
if h.turns.calls != 0 {
|
|
t.Errorf("awaiting gate resume should not rerun agent; calls=%d", h.turns.calls)
|
|
}
|
|
if len(h.sess.created) != 0 {
|
|
t.Errorf("awaiting gate resume should not create session; got %d", len(h.sess.created))
|
|
}
|
|
if h.jobs.count() != 1 {
|
|
t.Errorf("expected next step enqueued after phase write succeeds; got %d", h.jobs.count())
|
|
}
|
|
}
|
|
|
|
func TestStepHandler_GateRetryReturnsRetryableError(t *testing.T) {
|
|
h := newStepHandlerHarness(t, fakeGate{result: GateResult{Pass: false, Retry: true, Reason: "no artifact"}}, project.PhasePlanning, 3)
|
|
step := h.seedStep(project.PhasePlanning)
|
|
|
|
err := h.handler.Handle(context.Background(), jobFor(step))
|
|
if err == nil {
|
|
t.Fatal("expected retryable error")
|
|
}
|
|
if jobs.IsPermanent(err) {
|
|
t.Errorf("expected non-permanent error; got permanent: %v", err)
|
|
}
|
|
got, _ := h.repo.GetStep(context.Background(), step.ID)
|
|
if got.Attempt != 2 {
|
|
t.Errorf("attempt = %d; want 2", got.Attempt)
|
|
}
|
|
if h.jobs.count() != 0 {
|
|
t.Errorf("no next step should be enqueued on retry; got %d", h.jobs.count())
|
|
}
|
|
}
|
|
|
|
func TestStepHandler_AttemptsExhaustedDeadLetters(t *testing.T) {
|
|
h := newStepHandlerHarness(t, fakeGate{result: GateResult{Pass: false, Retry: true}}, project.PhasePlanning, 1)
|
|
step := h.seedStep(project.PhasePlanning) // attempt already 1 == max
|
|
|
|
err := h.handler.Handle(context.Background(), jobFor(step))
|
|
if !jobs.IsPermanent(err) {
|
|
t.Fatalf("expected permanent error; got %v", err)
|
|
}
|
|
got, _ := h.repo.GetStep(context.Background(), step.ID)
|
|
if got.Status != StepDead {
|
|
t.Errorf("step status = %s; want dead", got.Status)
|
|
}
|
|
gotRun, _ := h.repo.GetRun(context.Background(), h.run.ID)
|
|
if gotRun.Status != RunFailed {
|
|
t.Errorf("run status = %s; want failed", gotRun.Status)
|
|
}
|
|
if !h.audit.has("orchestrator.step.dead_letter") {
|
|
t.Error("expected dead_letter audit")
|
|
}
|
|
}
|
|
|
|
func TestStepHandler_GateNoRetryImmediateDeadLetter(t *testing.T) {
|
|
h := newStepHandlerHarness(t, fakeGate{result: GateResult{Pass: false, Retry: false, Reason: "refusal"}}, project.PhasePlanning, 5)
|
|
step := h.seedStep(project.PhasePlanning)
|
|
|
|
err := h.handler.Handle(context.Background(), jobFor(step))
|
|
if !jobs.IsPermanent(err) {
|
|
t.Fatalf("expected permanent error on no-retry gate; got %v", err)
|
|
}
|
|
got, _ := h.repo.GetStep(context.Background(), step.ID)
|
|
if got.Status != StepDead {
|
|
t.Errorf("step status = %s; want dead", got.Status)
|
|
}
|
|
}
|
|
|
|
func TestStepHandler_PausedRunNoOp(t *testing.T) {
|
|
h := newStepHandlerHarness(t, fakeGate{result: GateResult{Pass: true}}, project.PhasePlanning, 3)
|
|
h.run.Status = RunPaused
|
|
h.repo.runs[h.run.ID].Status = RunPaused
|
|
step := h.seedStep(project.PhasePlanning)
|
|
|
|
if err := h.handler.Handle(context.Background(), jobFor(step)); err != nil {
|
|
t.Fatalf("Handle should no-op complete; got %v", err)
|
|
}
|
|
if h.turns.calls != 0 {
|
|
t.Errorf("paused run should not drive a turn; calls=%d", h.turns.calls)
|
|
}
|
|
if len(h.sess.created) != 0 {
|
|
t.Errorf("paused run should not spawn a session; got %d", len(h.sess.created))
|
|
}
|
|
}
|
|
|
|
func TestStepHandler_DoneMarksRunSucceeded(t *testing.T) {
|
|
h := newStepHandlerHarness(t, fakeGate{result: GateResult{Pass: true}}, project.PhaseDone, 3)
|
|
step := h.seedStep(project.PhaseDone)
|
|
|
|
if err := h.handler.Handle(context.Background(), jobFor(step)); err != nil {
|
|
t.Fatalf("Handle: %v", err)
|
|
}
|
|
gotRun, _ := h.repo.GetRun(context.Background(), h.run.ID)
|
|
if gotRun.Status != RunSucceeded {
|
|
t.Errorf("run status = %s; want succeeded", gotRun.Status)
|
|
}
|
|
if h.jobs.count() != 0 {
|
|
t.Errorf("done should enqueue nothing; got %d", h.jobs.count())
|
|
}
|
|
if !h.audit.has("orchestrator.run.succeeded") {
|
|
t.Error("expected run.succeeded audit")
|
|
}
|
|
}
|
|
|
|
func TestStepHandler_ResumesCrashedSession(t *testing.T) {
|
|
h := newStepHandlerHarness(t, fakeGate{result: GateResult{Pass: true}}, project.PhasePlanning, 3)
|
|
step := h.seedStep(project.PhasePlanning)
|
|
// 该 step 已有崩溃 session：driveTurn 应走 resume，而非 Create。
|
|
crashedID := uuid.New()
|
|
h.lookup.crashed[step.ID] = &acp.Session{ID: crashedID, Status: acp.SessionCrashed}
|
|
|
|
if err := h.handler.Handle(context.Background(), jobFor(step)); err != nil {
|
|
t.Fatalf("Handle: %v", err)
|
|
}
|
|
if len(h.sess.created) != 0 {
|
|
t.Errorf("should resume not create; created=%d", len(h.sess.created))
|
|
}
|
|
got, _ := h.repo.GetStep(context.Background(), step.ID)
|
|
if got.ACPSessionID == nil || *got.ACPSessionID != crashedID {
|
|
t.Errorf("step should reference resumed session %s; got %v", crashedID, got.ACPSessionID)
|
|
}
|
|
}
|