toki/nomadcode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(workflow): task lifecycle를 별도 레이어로 분리하고 interface 추상화한다

Browse source 
- Lifecycle에 taskStore interface를 도입해 storage 의존성 추상화
- validTaskStatus, terminalTaskStatus, canEnqueueStatus helper 함수로 상태 검사 로직 분리
- QueueTask 메서드를 Lifecycle로 이동해 상태 전이 통합 관리
- Service에 lifecycle 필드 추가하고 EnqueueTask에서 lifecycle 활용
- lifecycle 전이 검증 및 enqueue 흐름 테스트 추가
This commit is contained in:
toki 2026-06-01 10:23:08 +09:00
parent 18853a5d31
commit d7aafc47d3
3 changed files with 468 additions and 24 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

56
services/core/internal/workflow/lifecycle.go
View file

@ -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 {

33
services/core/internal/workflow/service.go
View file

@ -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))
}

403
services/core/internal/workflow/service_test.go
View file

@ -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])
}
}