From d7aafc47d3628f65a37f51dd7fea75946fe44213 Mon Sep 17 00:00:00 2001 From: toki Date: Mon, 1 Jun 2026 10:23:08 +0900 Subject: [PATCH] =?UTF-8?q?refactor(workflow):=20task=20lifecycle=EB=A5=BC?= =?UTF-8?q?=20=EB=B3=84=EB=8F=84=20=EB=A0=88=EC=9D=B4=EC=96=B4=EB=A1=9C=20?= =?UTF-8?q?=EB=B6=84=EB=A6=AC=ED=95=98=EA=B3=A0=20interface=20=EC=B6=94?= =?UTF-8?q?=EC=83=81=ED=99=94=ED=95=9C=EB=8B=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Lifecycle에 taskStore interface를 도입해 storage 의존성 추상화 - validTaskStatus, terminalTaskStatus, canEnqueueStatus helper 함수로 상태 검사 로직 분리 - QueueTask 메서드를 Lifecycle로 이동해 상태 전이 통합 관리 - Service에 lifecycle 필드 추가하고 EnqueueTask에서 lifecycle 활용 - lifecycle 전이 검증 및 enqueue 흐름 테스트 추가 --- services/core/internal/workflow/lifecycle.go | 56 ++- services/core/internal/workflow/service.go | 33 +- .../core/internal/workflow/service_test.go | 403 ++++++++++++++++++ 3 files changed, 468 insertions(+), 24 deletions(-) diff --git a/services/core/internal/workflow/lifecycle.go b/services/core/internal/workflow/lifecycle.go index a93b95e..db31ec4 100644 --- a/services/core/internal/workflow/lifecycle.go +++ b/services/core/internal/workflow/lifecycle.go @@ -10,8 +10,18 @@ import ( "github.com/nomadcode/nomadcode-core/internal/storage" ) +type taskStore interface { + CreateTask(context.Context, storage.CreateTaskInput) (storage.Task, error) + GetTask(context.Context, string) (storage.Task, error) + ListTasks(context.Context, int32) ([]storage.Task, error) + UpdateMetadata(context.Context, string, json.RawMessage) (storage.Task, error) + UpdateStatus(context.Context, string, string) (storage.Task, error) + CompleteTask(context.Context, string, json.RawMessage) (storage.Task, error) + FailTask(context.Context, string, string) (storage.Task, error) +} + type Lifecycle struct { - store *storage.Store + store taskStore logger *slog.Logger } @@ -22,10 +32,39 @@ func NewLifecycle(store *storage.Store, logger *slog.Logger) *Lifecycle { } } +func validTaskStatus(status TaskStatus) bool { + switch status { + case StatusPending, StatusQueued, StatusRunning, StatusCompleted, StatusFailed, StatusCanceled: + return true + default: + return false + } +} + +func terminalTaskStatus(status TaskStatus) bool { + switch status { + case StatusCompleted, StatusFailed, StatusCanceled: + return true + default: + return false + } +} + +func canEnqueueStatus(status TaskStatus) bool { + switch status { + case StatusPending, StatusFailed: + return true + default: + return false + } +} + func canTransition(from, to TaskStatus) bool { switch from { - case StatusPending, StatusQueued, StatusFailed: + case StatusPending, StatusFailed: return to == StatusRunning + case StatusQueued: + return to == StatusRunning || to == StatusFailed || to == StatusCanceled case StatusRunning: return to == StatusCompleted || to == StatusFailed || to == StatusCanceled default: @@ -78,6 +117,19 @@ func mergeTaskMetadata(current json.RawMessage, updates map[string]any) (json.Ra return json.Marshal(currentMap) } +func (l *Lifecycle) QueueTask(ctx context.Context, id string) (storage.Task, error) { + task, err := l.store.GetTask(ctx, id) + if err != nil { + return storage.Task{}, err + } + + if !canEnqueueStatus(TaskStatus(task.Status)) { + return storage.Task{}, ErrTaskCannotBeEnqueued + } + + return l.store.UpdateStatus(ctx, id, string(StatusQueued)) +} + func (l *Lifecycle) StartTask(ctx context.Context, id string) (storage.Task, error) { task, err := l.store.GetTask(ctx, id) if err != nil { diff --git a/services/core/internal/workflow/service.go b/services/core/internal/workflow/service.go index ca42a7d..1617c1f 100644 --- a/services/core/internal/workflow/service.go +++ b/services/core/internal/workflow/service.go @@ -21,16 +21,18 @@ type TaskEnqueuer interface { } type Service struct { - store *storage.Store - enqueuer TaskEnqueuer - logger *slog.Logger + store *storage.Store + enqueuer TaskEnqueuer + logger *slog.Logger + lifecycle *Lifecycle } func NewService(store *storage.Store, enqueuer TaskEnqueuer, logger *slog.Logger) *Service { return &Service{ - store: store, - enqueuer: enqueuer, - logger: logger, + store: store, + enqueuer: enqueuer, + logger: logger, + lifecycle: NewLifecycle(store, logger), } } @@ -146,15 +148,7 @@ func (s *Service) UpdateTaskMetadata(ctx context.Context, id string, metadata js } func (s *Service) EnqueueTask(ctx context.Context, id string) (storage.Task, error) { - task, err := s.store.GetTask(ctx, id) - if err != nil { - return storage.Task{}, err - } - if !canEnqueue(task.Status) { - return storage.Task{}, ErrTaskCannotBeEnqueued - } - - queuedTask, err := s.store.UpdateStatus(ctx, id, string(StatusQueued)) + queuedTask, err := s.lifecycle.QueueTask(ctx, id) if err != nil { return storage.Task{}, err } @@ -163,7 +157,7 @@ func (s *Service) EnqueueTask(ctx context.Context, id string) (storage.Task, err return queuedTask, errors.New("task enqueuer is not configured") } if err := s.enqueuer.EnqueueTask(ctx, id); err != nil { - if _, failErr := s.store.FailTask(ctx, id, err.Error()); failErr != nil && s.logger != nil { + if _, failErr := s.lifecycle.FailTask(ctx, id, err.Error()); failErr != nil && s.logger != nil { s.logger.Error("failed to mark task enqueue error", "task_id", id, "error", failErr) } return queuedTask, err @@ -173,10 +167,5 @@ func (s *Service) EnqueueTask(ctx context.Context, id string) (storage.Task, err } func canEnqueue(status string) bool { - switch TaskStatus(status) { - case StatusPending, StatusFailed: - return true - default: - return false - } + return canEnqueueStatus(TaskStatus(status)) } diff --git a/services/core/internal/workflow/service_test.go b/services/core/internal/workflow/service_test.go index 58daa21..21fc89c 100644 --- a/services/core/internal/workflow/service_test.go +++ b/services/core/internal/workflow/service_test.go @@ -1,8 +1,12 @@ package workflow import ( + "context" "encoding/json" + "errors" "testing" + + "github.com/nomadcode/nomadcode-core/internal/storage" ) func TestNormalizeExternalRefDefaultsMetadata(t *testing.T) { @@ -173,3 +177,402 @@ func TestCanTransitionAllowsRetryFromFailedToRunning(t *testing.T) { t.Error("expected canTransition(StatusFailed, StatusRunning) to be true, got false") } } + +func TestTaskStatusHelpersClassifyLifecycleStates(t *testing.T) { + // 1. validTaskStatus + validStatuses := []TaskStatus{StatusPending, StatusQueued, StatusRunning, StatusCompleted, StatusFailed, StatusCanceled} + for _, status := range validStatuses { + if !validTaskStatus(status) { + t.Errorf("expected %s to be valid", status) + } + } + if validTaskStatus(TaskStatus("invalid")) { + t.Errorf("expected 'invalid' status to be invalid") + } + + // 2. terminalTaskStatus + terminalStatuses := map[TaskStatus]bool{ + StatusPending: false, + StatusQueued: false, + StatusRunning: false, + StatusCompleted: true, + StatusFailed: true, + StatusCanceled: true, + } + for status, expected := range terminalStatuses { + got := terminalTaskStatus(status) + if got != expected { + t.Errorf("terminalTaskStatus(%s) = %v; want %v", status, got, expected) + } + } + + // 3. canEnqueueStatus + enqueueStatuses := map[TaskStatus]bool{ + StatusPending: true, + StatusQueued: false, + StatusRunning: false, + StatusCompleted: false, + StatusFailed: true, + StatusCanceled: false, + } + for status, expected := range enqueueStatuses { + got := canEnqueueStatus(status) + if got != expected { + t.Errorf("canEnqueueStatus(%s) = %v; want %v", status, got, expected) + } + } +} + +type fakeTaskStore struct { + tasks map[string]storage.Task + operations []string + metadataCalls []metadataCall + statusCalls []statusCall + completeCalls []completeCall + failCalls []failCall +} + +type metadataCall struct { + id string + metadata json.RawMessage +} + +type statusCall struct { + id string + status string +} + +type completeCall struct { + id string + result json.RawMessage +} + +type failCall struct { + id string + message string +} + +func newFakeTaskStore() *fakeTaskStore { + return &fakeTaskStore{ + tasks: make(map[string]storage.Task), + } +} + +func (f *fakeTaskStore) CreateTask(ctx context.Context, input storage.CreateTaskInput) (storage.Task, error) { + return storage.Task{}, nil +} + +func (f *fakeTaskStore) GetTask(ctx context.Context, id string) (storage.Task, error) { + t, ok := f.tasks[id] + if !ok { + return storage.Task{}, errors.New("task not found") + } + return t, nil +} + +func (f *fakeTaskStore) ListTasks(ctx context.Context, limit int32) ([]storage.Task, error) { + return nil, nil +} + +func (f *fakeTaskStore) UpdateMetadata(ctx context.Context, id string, metadata json.RawMessage) (storage.Task, error) { + f.operations = append(f.operations, "metadata:"+id) + f.metadataCalls = append(f.metadataCalls, metadataCall{id: id, metadata: metadata}) + t, ok := f.tasks[id] + if !ok { + return storage.Task{}, errors.New("task not found") + } + t.Metadata = metadata + f.tasks[id] = t + return t, nil +} + +func (f *fakeTaskStore) UpdateStatus(ctx context.Context, id, status string) (storage.Task, error) { + f.operations = append(f.operations, "status:"+id+":"+status) + f.statusCalls = append(f.statusCalls, statusCall{id: id, status: status}) + t, ok := f.tasks[id] + if !ok { + return storage.Task{}, errors.New("task not found") + } + t.Status = status + f.tasks[id] = t + return t, nil +} + +func (f *fakeTaskStore) CompleteTask(ctx context.Context, id string, result json.RawMessage) (storage.Task, error) { + f.operations = append(f.operations, "complete:"+id) + f.completeCalls = append(f.completeCalls, completeCall{id: id, result: result}) + t, ok := f.tasks[id] + if !ok { + return storage.Task{}, errors.New("task not found") + } + t.Status = string(StatusCompleted) + t.Result = result + f.tasks[id] = t + return t, nil +} + +func (f *fakeTaskStore) FailTask(ctx context.Context, id, message string) (storage.Task, error) { + f.operations = append(f.operations, "fail:"+id) + f.failCalls = append(f.failCalls, failCall{id: id, message: message}) + t, ok := f.tasks[id] + if !ok { + return storage.Task{}, errors.New("task not found") + } + t.Status = string(StatusFailed) + t.Error = &message + f.tasks[id] = t + return t, nil +} + +func TestLifecycleStartCompleteFailTransitionsWithStore(t *testing.T) { + ctx := context.Background() + + // 1. Pending or Queued task can start and returns running. + t.Run("start task from pending", func(t *testing.T) { + store := newFakeTaskStore() + store.tasks["task-1"] = storage.Task{ + ID: "task-1", + Status: string(StatusPending), + Metadata: json.RawMessage(`{}`), + } + l := &Lifecycle{store: store} + + task, err := l.StartTask(ctx, "task-1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if task.Status != string(StatusRunning) { + t.Errorf("expected status running, got: %s", task.Status) + } + + // Verify metadata gets agent_run_state, last_heartbeat_at, and attempt=1 on start + var meta map[string]any + if err := json.Unmarshal(task.Metadata, &meta); err != nil { + t.Fatalf("failed to unmarshal metadata: %v", err) + } + if meta[MetadataKeyAgentRunState] != "running" { + t.Errorf("expected agent_run_state running, got: %v", meta[MetadataKeyAgentRunState]) + } + if meta[MetadataKeyAttempt] != float64(1) { + t.Errorf("expected attempt 1, got: %v", meta[MetadataKeyAttempt]) + } + if _, ok := meta[MetadataKeyLastHeartbeat]; !ok { + t.Errorf("expected last_heartbeat_at to be set") + } + + // Verify store calls in order: UpdateMetadata, then UpdateStatus using global operations log + if len(store.operations) != 2 { + t.Fatalf("expected exactly 2 store operations, got: %d", len(store.operations)) + } + if store.operations[0] != "metadata:task-1" { + t.Errorf("expected first operation to be metadata:task-1, got: %s", store.operations[0]) + } + if store.operations[1] != "status:task-1:running" { + t.Errorf("expected second operation to be status:task-1:running, got: %s", store.operations[1]) + } + }) + + t.Run("start task from queued (increments attempt)", func(t *testing.T) { + store := newFakeTaskStore() + store.tasks["task-2"] = storage.Task{ + ID: "task-2", + Status: string(StatusQueued), + Metadata: json.RawMessage(`{"attempt":1}`), + } + l := &Lifecycle{store: store} + + task, err := l.StartTask(ctx, "task-2") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if task.Status != string(StatusRunning) { + t.Errorf("expected status running, got: %s", task.Status) + } + + var meta map[string]any + if err := json.Unmarshal(task.Metadata, &meta); err != nil { + t.Fatalf("failed to unmarshal metadata: %v", err) + } + if meta[MetadataKeyAttempt] != float64(2) { + t.Errorf("expected attempt 2, got: %v", meta[MetadataKeyAttempt]) + } + }) + + // 2. Running task can complete and returns completed. + t.Run("complete running task", func(t *testing.T) { + store := newFakeTaskStore() + store.tasks["task-3"] = storage.Task{ + ID: "task-3", + Status: string(StatusRunning), + Metadata: json.RawMessage(`{"attempt":1,"agent_run_state":"running"}`), + } + l := &Lifecycle{store: store} + + result := json.RawMessage(`{"summary":"success detail"}`) + task, err := l.CompleteTask(ctx, "task-3", result) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if task.Status != string(StatusCompleted) { + t.Errorf("expected status completed, got: %s", task.Status) + } + + var meta map[string]any + if err := json.Unmarshal(task.Metadata, &meta); err != nil { + t.Fatalf("failed to unmarshal metadata: %v", err) + } + if meta[MetadataKeyAgentRunState] != "completed" { + t.Errorf("expected agent_run_state completed, got: %v", meta[MetadataKeyAgentRunState]) + } + if meta[MetadataKeyStatusReason] != "success detail" { + t.Errorf("expected status_reason 'success detail', got: %v", meta[MetadataKeyStatusReason]) + } + + // Verify store calls in order: UpdateMetadata, then CompleteTask using global operations log + if len(store.operations) != 2 { + t.Fatalf("expected exactly 2 store operations, got: %d", len(store.operations)) + } + if store.operations[0] != "metadata:task-3" { + t.Errorf("expected first operation to be metadata:task-3, got: %s", store.operations[0]) + } + if store.operations[1] != "complete:task-3" { + t.Errorf("expected second operation to be complete:task-3, got: %s", store.operations[1]) + } + }) + + // 3. Running task can fail and returns failed. + t.Run("fail running task", func(t *testing.T) { + store := newFakeTaskStore() + store.tasks["task-4"] = storage.Task{ + ID: "task-4", + Status: string(StatusRunning), + Metadata: json.RawMessage(`{"attempt":1,"agent_run_state":"running"}`), + } + l := &Lifecycle{store: store} + + task, err := l.FailTask(ctx, "task-4", "something went wrong") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if task.Status != string(StatusFailed) { + t.Errorf("expected status failed, got: %s", task.Status) + } + + var meta map[string]any + if err := json.Unmarshal(task.Metadata, &meta); err != nil { + t.Fatalf("failed to unmarshal metadata: %v", err) + } + if meta[MetadataKeyAgentRunState] != "failed" { + t.Errorf("expected agent_run_state failed, got: %v", meta[MetadataKeyAgentRunState]) + } + if meta[MetadataKeyStatusReason] != "something went wrong" { + t.Errorf("expected status_reason 'something went wrong', got: %v", meta[MetadataKeyStatusReason]) + } + + // Verify store calls in order: UpdateMetadata, then FailTask using global operations log + if len(store.operations) != 2 { + t.Fatalf("expected exactly 2 store operations, got: %d", len(store.operations)) + } + if store.operations[0] != "metadata:task-4" { + t.Errorf("expected first operation to be metadata:task-4, got: %s", store.operations[0]) + } + if store.operations[1] != "fail:task-4" { + t.Errorf("expected second operation to be fail:task-4, got: %s", store.operations[1]) + } + }) + + // 4. Completed task cannot start. + t.Run("cannot start completed task", func(t *testing.T) { + store := newFakeTaskStore() + store.tasks["task-5"] = storage.Task{ + ID: "task-5", + Status: string(StatusCompleted), + Metadata: json.RawMessage(`{}`), + } + l := &Lifecycle{store: store} + + _, err := l.StartTask(ctx, "task-5") + if err != ErrInvalidTaskTransition { + t.Errorf("expected ErrInvalidTaskTransition, got: %v", err) + } + }) +} + +type fakeTaskEnqueuer struct { + err error + calls []string +} + +func (f *fakeTaskEnqueuer) EnqueueTask(ctx context.Context, taskID string) error { + f.calls = append(f.calls, taskID) + return f.err +} + +func TestServiceEnqueueTaskUsesLifecycleQueueTransition(t *testing.T) { + ctx := context.Background() + store := newFakeTaskStore() + store.tasks["task-1"] = storage.Task{ + ID: "task-1", + Status: string(StatusPending), + } + + enqueuer := &fakeTaskEnqueuer{} + service := NewService(nil, enqueuer, nil) + service.lifecycle = &Lifecycle{store: store} + + task, err := service.EnqueueTask(ctx, "task-1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if task.Status != string(StatusQueued) { + t.Errorf("expected status queued, got: %s", task.Status) + } + if len(enqueuer.calls) != 1 || enqueuer.calls[0] != "task-1" { + t.Errorf("expected enqueuer to be called with task-1, got: %v", enqueuer.calls) + } + + if len(store.statusCalls) != 1 || store.statusCalls[0].status != string(StatusQueued) { + t.Errorf("expected status transition to queued, got: %v", store.statusCalls) + } +} + +func TestServiceEnqueueFailureMarksTaskFailedThroughLifecycle(t *testing.T) { + ctx := context.Background() + store := newFakeTaskStore() + store.tasks["task-1"] = storage.Task{ + ID: "task-1", + Status: string(StatusPending), + Metadata: json.RawMessage(`{}`), + } + + enqueuer := &fakeTaskEnqueuer{err: errors.New("enqueue error")} + service := NewService(nil, enqueuer, nil) + service.lifecycle = &Lifecycle{store: store} + + _, err := service.EnqueueTask(ctx, "task-1") + if err == nil { + t.Fatal("expected error, got nil") + } + if err.Error() != "enqueue error" { + t.Errorf("expected 'enqueue error', got: %v", err) + } + + task, err := store.GetTask(ctx, "task-1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if task.Status != string(StatusFailed) { + t.Errorf("expected task to be marked failed, got status: %s", task.Status) + } + if task.Error == nil || *task.Error != "enqueue error" { + t.Errorf("expected error message 'enqueue error', got: %v", task.Error) + } + + var meta map[string]any + if err := json.Unmarshal(task.Metadata, &meta); err != nil { + t.Fatalf("failed to unmarshal metadata: %v", err) + } + if meta[MetadataKeyAgentRunState] != "failed" { + t.Errorf("expected agent_run_state failed, got: %v", meta[MetadataKeyAgentRunState]) + } +}