From c078438621e6f6765d35bcca2f6aa859045a41c1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Almeida?= <57680069+zeval@users.noreply.github.com> Date: Thu, 23 Apr 2026 15:16:31 +0100 Subject: [PATCH] refactor: unify task.updated via single publisher and shared mapper (#676) --- apps/backend/cmd/kandev/orchestrator.go | 5 + .../internal/orchestrator/event_handlers.go | 19 --- .../orchestrator/event_handlers_workflow.go | 11 +- apps/backend/internal/orchestrator/service.go | 39 ++++- .../internal/orchestrator/task_operations.go | 21 +-- .../internal/orchestrator/workflow_store.go | 22 ++- .../orchestrator/workflow_store_test.go | 17 ++- .../repository/session_repository_test.go | 38 +++++ .../task/repository/sqlite/session.go | 6 +- .../internal/task/service/service_events.go | 96 +++++++++---- .../internal/task/service/service_test.go | 57 ++++++++ .../kanban/use-all-workflow-snapshots.ts | 35 +---- apps/web/lib/kanban/map-task.test.ts | 133 ++++++++++++++++++ apps/web/lib/kanban/map-task.ts | 82 +++++++++++ apps/web/lib/ws/handlers/tasks.ts | 82 ++--------- 15 files changed, 469 insertions(+), 194 deletions(-) create mode 100644 apps/web/lib/kanban/map-task.test.ts create mode 100644 apps/web/lib/kanban/map-task.ts diff --git a/apps/backend/cmd/kandev/orchestrator.go b/apps/backend/cmd/kandev/orchestrator.go index c5832e33d3..342996877f 100644 --- a/apps/backend/cmd/kandev/orchestrator.go +++ b/apps/backend/cmd/kandev/orchestrator.go @@ -70,6 +70,11 @@ func provideOrchestrator( orchestratorSvc.SetTurnService(newTurnServiceAdapter(taskSvc)) + // Route orchestrator task.updated events through the task service, which + // owns the canonical rich payload. Covers workflow transitions, workflow + // step moves, and the primary-session-set callback below. + orchestratorSvc.SetTaskEventPublisher(taskSvc) + // Publish task.updated when the first session is marked primary so the // frontend receives primary_session_id for newly created tasks. orchestratorSvc.SetOnPrimarySessionSet(func(ctx context.Context, taskID, _ string) { diff --git a/apps/backend/internal/orchestrator/event_handlers.go b/apps/backend/internal/orchestrator/event_handlers.go index ee5daac0b8..ff36c5874e 100644 --- a/apps/backend/internal/orchestrator/event_handlers.go +++ b/apps/backend/internal/orchestrator/event_handlers.go @@ -3,7 +3,6 @@ package orchestrator import ( "context" - "time" "go.uber.org/zap" @@ -21,24 +20,6 @@ const ( agentEventFailed = "failed" ) -// buildTaskEventPayload builds the standard map payload for TaskUpdated events. -// This is the single source of truth; all TaskUpdated publishers should call it. -func buildTaskEventPayload(task *models.Task) map[string]interface{} { - return map[string]interface{}{ - "task_id": task.ID, - "workflow_id": task.WorkflowID, - "workflow_step_id": task.WorkflowStepID, - "title": task.Title, - "description": task.Description, - "state": string(task.State), - "priority": task.Priority, - "position": task.Position, - "created_at": task.CreatedAt.Format(time.RFC3339Nano), - "updated_at": task.UpdatedAt.Format(time.RFC3339Nano), - "is_ephemeral": task.IsEphemeral, - } -} - // toolKindToMessageType maps the normalized tool kind to a frontend message type. func toolKindToMessageType(normalized *streams.NormalizedPayload) string { if normalized == nil { diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow.go b/apps/backend/internal/orchestrator/event_handlers_workflow.go index d0c7c4d3a1..5867c9e7c8 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow.go +++ b/apps/backend/internal/orchestrator/event_handlers_workflow.go @@ -241,14 +241,9 @@ func (s *Service) executeStepTransition(ctx context.Context, taskID, sessionID s return } - // Publish task updated event - if s.eventBus != nil { - _ = s.eventBus.Publish(ctx, events.TaskUpdated, bus.NewEvent( - events.TaskUpdated, - "orchestrator", - buildTaskEventPayload(task), - )) - } + // Publish task updated event via the task service so the payload carries + // the full context (session counts, primary session, repositories). + s.publishTaskUpdated(ctx, task) s.logger.Info("workflow transition completed", zap.String("task_id", taskID), diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index a28c6dba61..67cdd955ba 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -93,6 +93,15 @@ type TurnService interface { GetActiveTurn(ctx context.Context, sessionID string) (*models.Turn, error) } +// TaskEventPublisher is the orchestrator's collaborator for publishing +// task.updated events. Implemented by the task service (which owns the rich +// payload build — session counts, primary session info, repositories, +// metadata, parent_id). The orchestrator does not construct task.updated +// payloads itself. +type TaskEventPublisher interface { + PublishTaskUpdated(ctx context.Context, task *models.Task) +} + // WorkflowStepGetter retrieves workflow step information for prompt building. type WorkflowStepGetter interface { GetStep(ctx context.Context, stepID string) (*wfmodels.WorkflowStep, error) @@ -187,6 +196,10 @@ type Service struct { // Turn service for managing session turns turnService TurnService + // Task event publisher for emitting task.updated events. + // Task service owns the rich payload; orchestrator delegates. + taskEvents TaskEventPublisher + // Workflow step getter for prompt building workflowStepGetter WorkflowStepGetter @@ -387,6 +400,30 @@ func (s *Service) SetTurnService(turnService TurnService) { s.turnService = turnService } +// SetTaskEventPublisher wires the publisher used for task.updated events. +// +// The task service is the canonical publisher: it loads session counts, +// primary session info, repositories, metadata, and parent_id, and emits a +// single rich payload. Orchestrator code paths that mutate a task (workflow +// transitions, primary session assignment, workflow-step moves) delegate to +// this publisher instead of constructing their own partial payloads. +// +// If not set: orchestrator task.updated publishers no-op (nothing is emitted +// on those paths). Task service's own publishTaskEvent calls are unaffected. +func (s *Service) SetTaskEventPublisher(publisher TaskEventPublisher) { + s.taskEvents = publisher +} + +// publishTaskUpdated forwards to the configured TaskEventPublisher. +// No-op when the publisher isn't wired (tests, or before SetTaskEventPublisher +// has been called during startup). +func (s *Service) publishTaskUpdated(ctx context.Context, task *models.Task) { + if s.taskEvents == nil || task == nil { + return + } + s.taskEvents.PublishTaskUpdated(ctx, task) +} + // SetWorkflowStepGetter sets the workflow step getter for prompt building. // // When workflow_step_id is provided to StartTask, the orchestrator uses this getter @@ -415,7 +452,7 @@ func (s *Service) initWorkflowEngine() { if s.workflowStepGetter == nil { return } - store := newWorkflowStore(s.repo, s.workflowStepGetter, s.agentManager, s.eventBus, s.logger) + store := newWorkflowStore(s.repo, s.workflowStepGetter, s.agentManager, s.publishTaskUpdated, s.logger) callbacks := buildWorkflowCallbacks(s) s.workflowStore = store s.workflowEngine = engine.New(store, callbacks) diff --git a/apps/backend/internal/orchestrator/task_operations.go b/apps/backend/internal/orchestrator/task_operations.go index c99c2eb521..9de23be53e 100644 --- a/apps/backend/internal/orchestrator/task_operations.go +++ b/apps/backend/internal/orchestrator/task_operations.go @@ -13,8 +13,6 @@ import ( "github.com/kandev/kandev/internal/agent/lifecycle" "github.com/kandev/kandev/internal/agentctl/client" - "github.com/kandev/kandev/internal/events" - "github.com/kandev/kandev/internal/events/bus" "github.com/kandev/kandev/internal/orchestrator/dto" "github.com/kandev/kandev/internal/orchestrator/executor" "github.com/kandev/kandev/internal/orchestrator/queue" @@ -502,13 +500,9 @@ func (s *Service) moveTaskToWorkflowStep(ctx context.Context, taskID, workflowSt zap.String("task_id", taskID), zap.String("workflow_step_id", workflowStepID), zap.Error(err)) - } else if s.eventBus != nil { - _ = s.eventBus.Publish(ctx, events.TaskUpdated, bus.NewEvent( - events.TaskUpdated, - "orchestrator", - buildTaskEventPayload(dbTask), - )) + return } + s.publishTaskUpdated(ctx, dbTask) } // resolveEffectiveAgentProfile checks whether the task's workflow step overrides @@ -1303,6 +1297,8 @@ func (s *Service) SetPrimarySession(ctx context.Context, sessionID string) error } // Broadcast task.updated so frontend updates the primary star indicator. + // The task service's publisher loads primary-session info from the DB, + // which already reflects the SetSessionPrimary write above. session, err := s.repo.GetTaskSession(ctx, sessionID) if err != nil { s.logger.Warn("failed to fetch session after setting primary", zap.Error(err)) @@ -1313,14 +1309,7 @@ func (s *Service) SetPrimarySession(ctx context.Context, sessionID string) error s.logger.Warn("failed to fetch task after setting primary", zap.Error(err)) return nil } - if s.eventBus != nil { - payload := buildTaskEventPayload(task) - payload["primary_session_id"] = sessionID - payload["primary_session_state"] = string(session.State) - _ = s.eventBus.Publish(ctx, events.TaskUpdated, bus.NewEvent( - events.TaskUpdated, "orchestrator", payload, - )) - } + s.publishTaskUpdated(ctx, task) return nil } diff --git a/apps/backend/internal/orchestrator/workflow_store.go b/apps/backend/internal/orchestrator/workflow_store.go index 8b06ec0463..c1abdb9428 100644 --- a/apps/backend/internal/orchestrator/workflow_store.go +++ b/apps/backend/internal/orchestrator/workflow_store.go @@ -9,19 +9,23 @@ import ( "go.uber.org/zap" "github.com/kandev/kandev/internal/common/logger" - "github.com/kandev/kandev/internal/events" - "github.com/kandev/kandev/internal/events/bus" "github.com/kandev/kandev/internal/orchestrator/executor" + "github.com/kandev/kandev/internal/task/models" "github.com/kandev/kandev/internal/workflow/engine" ) +// taskUpdatedPublisher is the minimal hook the workflow store needs to emit +// task.updated events. The orchestrator Service binds this to its shared +// publishTaskUpdated helper so the publisher wiring stays in one place. +type taskUpdatedPublisher func(ctx context.Context, task *models.Task) + // workflowStore implements engine.TransitionStore by delegating to the // orchestrator's existing repositories and services. type workflowStore struct { repo sessionExecutorStore workflowStepGetter WorkflowStepGetter agentManager executor.AgentManagerClient - eventBus bus.EventBus + publishTaskUpdated taskUpdatedPublisher logger *logger.Logger appliedOps sync.Map } @@ -30,14 +34,14 @@ func newWorkflowStore( repo sessionExecutorStore, stepGetter WorkflowStepGetter, agentMgr executor.AgentManagerClient, - eventBus bus.EventBus, + publishTaskUpdated taskUpdatedPublisher, log *logger.Logger, ) *workflowStore { return &workflowStore{ repo: repo, workflowStepGetter: stepGetter, agentManager: agentMgr, - eventBus: eventBus, + publishTaskUpdated: publishTaskUpdated, logger: log, } } @@ -103,13 +107,7 @@ func (s *workflowStore) ApplyTransition(ctx context.Context, taskID, sessionID, return fmt.Errorf("update task workflow step: %w", err) } - if s.eventBus != nil { - _ = s.eventBus.Publish(ctx, events.TaskUpdated, bus.NewEvent( - events.TaskUpdated, - "orchestrator", - buildTaskEventPayload(task), - )) - } + s.publishTaskUpdated(ctx, task) if err := s.repo.UpdateSessionReviewStatus(ctx, sessionID, ""); err != nil { s.logger.Warn("failed to clear session review status", diff --git a/apps/backend/internal/orchestrator/workflow_store_test.go b/apps/backend/internal/orchestrator/workflow_store_test.go index f26aebd8a7..e25a67d113 100644 --- a/apps/backend/internal/orchestrator/workflow_store_test.go +++ b/apps/backend/internal/orchestrator/workflow_store_test.go @@ -4,16 +4,21 @@ import ( "context" "testing" + "github.com/kandev/kandev/internal/task/models" wfmodels "github.com/kandev/kandev/internal/workflow/models" ) +// noopPublisher satisfies the taskUpdatedPublisher contract without touching +// an event bus. Workflow-store unit tests don't exercise the event path. +func noopPublisher(_ context.Context, _ *models.Task) {} + func TestWorkflowStore_LoadState(t *testing.T) { ctx := context.Background() repo := setupTestRepo(t) seedSession(t, repo, "t1", "s1", "step1") agentMgr := &mockAgentManager{isPassthrough: true} - store := newWorkflowStore(repo, newMockStepGetter(), agentMgr, nil, testLogger()) + store := newWorkflowStore(repo, newMockStepGetter(), agentMgr, noopPublisher, testLogger()) state, err := store.LoadState(ctx, "t1", "s1") if err != nil { @@ -51,7 +56,7 @@ func TestWorkflowStore_LoadStep(t *testing.T) { Position: 0, } - store := newWorkflowStore(nil, stepGetter, nil, nil, testLogger()) + store := newWorkflowStore(nil, stepGetter, nil, noopPublisher, testLogger()) spec, err := store.LoadStep(ctx, "wf1", "step1") if err != nil { @@ -83,7 +88,7 @@ func TestWorkflowStore_LoadNextStep(t *testing.T) { ID: "step3", WorkflowID: "wf1", Name: "Step 3", Position: 2, } - store := newWorkflowStore(nil, stepGetter, nil, nil, testLogger()) + store := newWorkflowStore(nil, stepGetter, nil, noopPublisher, testLogger()) t.Run("returns next step by position", func(t *testing.T) { spec, err := store.LoadNextStep(ctx, "wf1", 0) @@ -111,7 +116,7 @@ func TestWorkflowStore_ApplyTransition(t *testing.T) { repo := setupTestRepo(t) seedSession(t, repo, "t1", "s1", "step1") - store := newWorkflowStore(repo, newMockStepGetter(), nil, nil, testLogger()) + store := newWorkflowStore(repo, newMockStepGetter(), nil, noopPublisher, testLogger()) err := store.ApplyTransition(ctx, "t1", "s1", "step1", "step2", "on_turn_complete") if err != nil { @@ -142,7 +147,7 @@ func TestWorkflowStore_PersistData(t *testing.T) { repo := setupTestRepo(t) seedSession(t, repo, "t1", "s1", "step1") - store := newWorkflowStore(repo, newMockStepGetter(), nil, nil, testLogger()) + store := newWorkflowStore(repo, newMockStepGetter(), nil, noopPublisher, testLogger()) // Persist initial data err := store.PersistData(ctx, "s1", map[string]any{"plan_mode": true}) @@ -187,7 +192,7 @@ func TestWorkflowStore_PersistData(t *testing.T) { func TestWorkflowStore_OperationIdempotency(t *testing.T) { ctx := context.Background() - store := newWorkflowStore(nil, newMockStepGetter(), nil, nil, testLogger()) + store := newWorkflowStore(nil, newMockStepGetter(), nil, noopPublisher, testLogger()) t.Run("empty operation ID returns false", func(t *testing.T) { applied, err := store.IsOperationApplied(ctx, "") diff --git a/apps/backend/internal/task/repository/session_repository_test.go b/apps/backend/internal/task/repository/session_repository_test.go index b386675fe6..741039f5a8 100644 --- a/apps/backend/internal/task/repository/session_repository_test.go +++ b/apps/backend/internal/task/repository/session_repository_test.go @@ -439,3 +439,41 @@ func TestSQLiteRepository_CompletePendingToolCallsForTurn(t *testing.T) { t.Errorf("expected 0 affected rows on second call, got %d", affected2) } } + +// TestGetPrimarySessionInfoByTaskIDs_PopulatesID locks in the SQL fix: the +// SELECT now includes ts.id so session.ID is populated on the returned +// TaskSession. Before the fix, publishTaskEvent saw sessionInfo.ID == "" and +// emitted an empty primary_session_id in every WS payload. +func TestGetPrimarySessionInfoByTaskIDs_PopulatesID(t *testing.T) { + repo, cleanup := createTestSQLiteRepo(t) + defer cleanup() + ctx := context.Background() + + if err := repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf-1", Name: "WF"}); err != nil { + t.Fatalf("CreateWorkflow: %v", err) + } + if err := repo.CreateTask(ctx, &models.Task{ID: "task-1", WorkflowID: "wf-1", WorkflowStepID: "step-1", Title: "T"}); err != nil { + t.Fatalf("CreateTask: %v", err) + } + session := &models.TaskSession{ + ID: "session-abc", TaskID: "task-1", State: models.TaskSessionStateRunning, + } + if err := repo.CreateTaskSession(ctx, session); err != nil { + t.Fatalf("CreateTaskSession: %v", err) + } + if err := repo.SetSessionPrimary(ctx, "session-abc"); err != nil { + t.Fatalf("SetSessionPrimary: %v", err) + } + + info, err := repo.GetPrimarySessionInfoByTaskIDs(ctx, []string{"task-1"}) + if err != nil { + t.Fatalf("GetPrimarySessionInfoByTaskIDs: %v", err) + } + got, ok := info["task-1"] + if !ok || got == nil { + t.Fatalf("expected primary session info for task-1, got %#v", info) + } + if got.ID != "session-abc" { + t.Errorf("expected session ID %q, got %q (ts.id missing from SELECT?)", "session-abc", got.ID) + } +} diff --git a/apps/backend/internal/task/repository/sqlite/session.go b/apps/backend/internal/task/repository/sqlite/session.go index 0d408d7361..d039be215a 100644 --- a/apps/backend/internal/task/repository/sqlite/session.go +++ b/apps/backend/internal/task/repository/sqlite/session.go @@ -992,7 +992,7 @@ func (r *Repository) GetPrimarySessionInfoByTaskIDs(ctx context.Context, taskIDs } query := fmt.Sprintf(` - SELECT ts.task_id, ts.review_status, ts.executor_id, ts.state, + SELECT ts.id, ts.task_id, ts.review_status, ts.executor_id, ts.state, ts.agent_profile_snapshot, ts.repository_snapshot, e.type, e.name FROM task_sessions ts @@ -1008,6 +1008,7 @@ func (r *Repository) GetPrimarySessionInfoByTaskIDs(ctx context.Context, taskIDs result := make(map[string]*models.TaskSession) for rows.Next() { + var sessionID string var taskID string var reviewStatus sql.NullString var executorID sql.NullString @@ -1016,10 +1017,11 @@ func (r *Repository) GetPrimarySessionInfoByTaskIDs(ctx context.Context, taskIDs var repositorySnapshotJSON sql.NullString var executorType sql.NullString var executorName sql.NullString - if err := rows.Scan(&taskID, &reviewStatus, &executorID, &sessionState, &agentProfileSnapshotJSON, &repositorySnapshotJSON, &executorType, &executorName); err != nil { + if err := rows.Scan(&sessionID, &taskID, &reviewStatus, &executorID, &sessionState, &agentProfileSnapshotJSON, &repositorySnapshotJSON, &executorType, &executorName); err != nil { return nil, err } session := &models.TaskSession{ + ID: sessionID, TaskID: taskID, } if sessionState.Valid { diff --git a/apps/backend/internal/task/service/service_events.go b/apps/backend/internal/task/service/service_events.go index a9c92c5166..7eb95eb6c5 100644 --- a/apps/backend/internal/task/service/service_events.go +++ b/apps/backend/internal/task/service/service_events.go @@ -38,55 +38,33 @@ func (s *Service) publishTaskEvent(ctx context.Context, eventType string, task * "position": task.Position, "created_at": task.CreatedAt.Format(time.RFC3339), "updated_at": task.UpdatedAt.Format(time.RFC3339), + "is_ephemeral": task.IsEphemeral, } - // Fetch session count and primary session info for the task - sessionCountMap, err := s.GetSessionCountsForTasks(ctx, []string{task.ID}) - if err == nil { - if count, ok := sessionCountMap[task.ID]; ok { - data["session_count"] = count - } - } - - primarySessionInfoMap, err := s.GetPrimarySessionInfoForTasks(ctx, []string{task.ID}) - if err == nil { - if sessionInfo, ok := primarySessionInfoMap[task.ID]; ok && sessionInfo != nil { - data["primary_session_id"] = sessionInfo.ID - if sessionInfo.ReviewStatus != nil { - data["review_status"] = *sessionInfo.ReviewStatus - } - if sessionInfo.State != "" { - data["primary_session_state"] = string(sessionInfo.State) - } - } - } + s.addTaskSessionEventFields(ctx, task.ID, data) if task.ParentID != "" { data["parent_id"] = task.ParentID } - if task.ArchivedAt != nil { data["archived_at"] = task.ArchivedAt.Format(time.RFC3339) } - - // Always include is_ephemeral field so frontend filters work correctly - data["is_ephemeral"] = task.IsEphemeral - - if len(task.Repositories) > 0 { - data["repository_id"] = task.Repositories[0].RepositoryID + // Orchestrator-originated events fetch the task via the raw repo.GetTask, + // which does not populate Repositories. Load the primary on demand so the + // payload always carries repository_id — matching the HTTP DTO and + // preventing the frontend from losing repositoryId on WS updates. + if repoID := primaryRepositoryID(ctx, s, task); repoID != "" { + data["repository_id"] = repoID } - if task.Metadata != nil { data["metadata"] = task.Metadata } - if oldState != nil { data["old_state"] = string(*oldState) data["new_state"] = string(task.State) } event := bus.NewEvent(eventType, "task-service", data) - if err := s.eventBus.Publish(ctx, eventType, event); err != nil { s.logger.Error("failed to publish task event", zap.String("event_type", eventType), @@ -95,6 +73,64 @@ func (s *Service) publishTaskEvent(ctx context.Context, eventType string, task * } } +// addTaskSessionEventFields merges session count, primary session info, and +// primary executor details into the task event payload. Extracted to keep +// publishTaskEvent under the project's function-length limit. +func (s *Service) addTaskSessionEventFields(ctx context.Context, taskID string, data map[string]interface{}) { + if sessionCountMap, err := s.GetSessionCountsForTasks(ctx, []string{taskID}); err == nil { + if count, ok := sessionCountMap[taskID]; ok { + data["session_count"] = count + } + } + + primarySessionInfoMap, err := s.GetPrimarySessionInfoForTasks(ctx, []string{taskID}) + if err != nil { + return + } + sessionInfo, ok := primarySessionInfoMap[taskID] + if !ok || sessionInfo == nil { + return + } + data["primary_session_id"] = sessionInfo.ID + if sessionInfo.ReviewStatus != nil { + data["review_status"] = *sessionInfo.ReviewStatus + } + if sessionInfo.State != "" { + data["primary_session_state"] = string(sessionInfo.State) + } + if sessionInfo.ExecutorID != "" { + data["primary_executor_id"] = sessionInfo.ExecutorID + } + var execType string + if sessionInfo.ExecutorSnapshot != nil { + if t, ok := sessionInfo.ExecutorSnapshot["executor_type"].(string); ok && t != "" { + execType = t + data["primary_executor_type"] = t + } + if n, ok := sessionInfo.ExecutorSnapshot["executor_name"].(string); ok && n != "" { + data["primary_executor_name"] = n + } + } + if execType != "" { + data["is_remote_executor"] = models.IsRemoteExecutorType(models.ExecutorType(execType)) + } +} + +// primaryRepositoryID returns the primary repository_id for the task. Prefers +// the already-loaded Task.Repositories slice; falls back to a lookup so +// publishers that pass a task without eagerly loaded repositories (e.g. the +// orchestrator's raw repo.GetTask) still emit repository_id. +func primaryRepositoryID(ctx context.Context, s *Service, task *models.Task) string { + if len(task.Repositories) > 0 { + return task.Repositories[0].RepositoryID + } + repo, err := s.taskRepos.GetPrimaryTaskRepository(ctx, task.ID) + if err != nil || repo == nil { + return "" + } + return repo.RepositoryID +} + // publishTaskMovedEvent publishes a task.moved event so the orchestrator can process // on_exit/on_enter actions for the new workflow step. func (s *Service) publishTaskMovedEvent(ctx context.Context, task *models.Task, fromStepID, toStepID, sessionID string) { diff --git a/apps/backend/internal/task/service/service_test.go b/apps/backend/internal/task/service/service_test.go index b23fcd3b73..a427174ae8 100644 --- a/apps/backend/internal/task/service/service_test.go +++ b/apps/backend/internal/task/service/service_test.go @@ -663,3 +663,60 @@ func TestService_DeleteMessage(t *testing.T) { t.Error("expected comment to be deleted") } } + +// TestPublishTaskUpdated_FallbackRepositoryID exercises the DB fallback in +// primaryRepositoryID: orchestrator-originated events load the task via the +// raw repo.GetTask, which does not populate Repositories. The publisher must +// still emit repository_id so the frontend doesn't lose the repo link on +// workflow transitions or state changes. +func TestPublishTaskUpdated_FallbackRepositoryID(t *testing.T) { + svc, eventBus, repo := createTestService(t) + ctx := context.Background() + + // Seed workspace + workflow + repo + task with an association. + if err := repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-1", Name: "Workspace"}); err != nil { + t.Fatalf("CreateWorkspace: %v", err) + } + if err := repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf-1", WorkspaceID: "ws-1", Name: "WF"}); err != nil { + t.Fatalf("CreateWorkflow: %v", err) + } + if err := repo.CreateRepository(ctx, &models.Repository{ID: "repo-x", WorkspaceID: "ws-1", Name: "Repo"}); err != nil { + t.Fatalf("CreateRepository: %v", err) + } + if err := repo.CreateTask(ctx, &models.Task{ + ID: "task-1", WorkspaceID: "ws-1", WorkflowID: "wf-1", WorkflowStepID: "step-1", Title: "T", + }); err != nil { + t.Fatalf("CreateTask: %v", err) + } + if err := repo.CreateTaskRepository(ctx, &models.TaskRepository{ + TaskID: "task-1", RepositoryID: "repo-x", BaseBranch: "main", + }); err != nil { + t.Fatalf("CreateTaskRepository: %v", err) + } + eventBus.ClearEvents() + + // Mimic the orchestrator path: pass a task with Repositories nil. + task := &models.Task{ + ID: "task-1", WorkspaceID: "ws-1", WorkflowID: "wf-1", WorkflowStepID: "step-1", + } + if len(task.Repositories) != 0 { + t.Fatal("pre-condition: task.Repositories must be nil for this test") + } + svc.PublishTaskUpdated(ctx, task) + + events := eventBus.GetPublishedEvents() + if len(events) != 1 { + t.Fatalf("expected 1 published event, got %d", len(events)) + } + data, ok := events[0].Data.(map[string]interface{}) + if !ok { + t.Fatalf("event Data wrong type: %T", events[0].Data) + } + got, ok := data["repository_id"].(string) + if !ok { + t.Fatalf("repository_id missing from payload or wrong type: %#v", data["repository_id"]) + } + if got != "repo-x" { + t.Fatalf("expected repository_id=repo-x via DB fallback, got %q", got) + } +} diff --git a/apps/web/hooks/domains/kanban/use-all-workflow-snapshots.ts b/apps/web/hooks/domains/kanban/use-all-workflow-snapshots.ts index 73e8038eca..dda875334a 100644 --- a/apps/web/hooks/domains/kanban/use-all-workflow-snapshots.ts +++ b/apps/web/hooks/domains/kanban/use-all-workflow-snapshots.ts @@ -1,13 +1,9 @@ import { useEffect, useRef, type MutableRefObject } from "react"; import { fetchWorkflowSnapshot } from "@/lib/api"; import { useAppStore, useAppStoreApi } from "@/components/state-provider"; +import { toKanbanTask } from "@/lib/kanban/map-task"; import type { KanbanState } from "@/lib/state/slices/kanban/types"; import type { Task } from "@/lib/types/http"; -import { - isPRReviewFromMetadata, - isIssueWatchFromMetadata, - issueFieldsFromMetadata, -} from "@/lib/metadata-utils"; import type { StoreApi } from "zustand"; import type { AppState } from "@/lib/state/store"; @@ -70,34 +66,9 @@ async function fetchAndWriteSnapshot( } } -// eslint-disable-next-line complexity -- pure field mapping, no real branching logic function mapSnapshotTask(task: Task, stepIds: Set): KanbanTask | null { - const workflowStepId = task.workflow_step_id; - if (!workflowStepId || !stepIds.has(workflowStepId)) return null; - - return { - id: task.id, - workflowStepId, - title: task.title, - description: task.description ?? undefined, - position: task.position ?? 0, - state: task.state, - repositoryId: task.repositories?.[0]?.repository_id ?? undefined, - primarySessionId: task.primary_session_id ?? undefined, - primarySessionState: task.primary_session_state ?? undefined, - sessionCount: task.session_count ?? undefined, - reviewStatus: task.review_status ?? undefined, - primaryExecutorId: task.primary_executor_id ?? undefined, - primaryExecutorType: task.primary_executor_type ?? undefined, - primaryExecutorName: task.primary_executor_name ?? undefined, - isRemoteExecutor: task.is_remote_executor ?? false, - parentTaskId: task.parent_id ?? undefined, - updatedAt: task.updated_at, - createdAt: task.created_at, - isPRReview: isPRReviewFromMetadata(task.metadata), - isIssueWatch: isIssueWatchFromMetadata(task.metadata), - ...issueFieldsFromMetadata(task.metadata), - } as KanbanTask; + if (!task.workflow_step_id || !stepIds.has(task.workflow_step_id)) return null; + return toKanbanTask(task); } export function useAllWorkflowSnapshots(workspaceId: string | null) { diff --git a/apps/web/lib/kanban/map-task.test.ts b/apps/web/lib/kanban/map-task.test.ts new file mode 100644 index 0000000000..88dd98f997 --- /dev/null +++ b/apps/web/lib/kanban/map-task.test.ts @@ -0,0 +1,133 @@ +import { describe, it, expect } from "vitest"; +import { toKanbanTask, type TaskLike } from "./map-task"; + +/** + * Parity matrix: the HTTP DTO and the WS payload describe the same task from + * two shapes. Both must produce the same KanbanTask. If someone changes the + * backend publisher (or adds a metadata-derived field to one side only), a + * test here fails loudly — instead of silently diverging until a sidebar bug + * surfaces. + */ + +const BASE_SCALARS = { + workflow_step_id: "step-1", + title: "Ship the thing", + description: "Do it well", + position: 3, + state: "TODO" as const, + priority: 0, + is_ephemeral: false, + created_at: "2026-04-22T10:00:00Z", + updated_at: "2026-04-22T10:05:00Z", +}; + +function httpDTO(overrides: Partial = {}): TaskLike { + return { + id: "task-1", + workspace_id: "ws-1", + workflow_id: "wf-1", + ...BASE_SCALARS, + repositories: [{ repository_id: "repo-a" }], + primary_session_id: "session-p", + primary_session_state: "RUNNING", + session_count: 2, + review_status: "pending", + primary_executor_id: "exec-1", + primary_executor_type: "local_docker", + primary_executor_name: "Docker", + is_remote_executor: false, + parent_id: "parent-1", + metadata: null, + ...overrides, + } as TaskLike; +} + +function wsPayload(overrides: Partial = {}): TaskLike { + return { + task_id: "task-1", + workflow_id: "wf-1", + ...BASE_SCALARS, + repository_id: "repo-a", + primary_session_id: "session-p", + primary_session_state: "RUNNING", + session_count: 2, + review_status: "pending", + primary_executor_id: "exec-1", + primary_executor_type: "local_docker", + primary_executor_name: "Docker", + is_remote_executor: false, + parent_id: "parent-1", + metadata: null, + ...overrides, + } as TaskLike; +} + +describe("toKanbanTask — HTTP DTO / WS payload parity", () => { + it("plain task: both shapes produce identical KanbanTask", () => { + expect(toKanbanTask(httpDTO())).toEqual(toKanbanTask(wsPayload())); + }); + + it("PR review task: review_watch_id flags isPRReview from both shapes", () => { + const metadata = { review_watch_id: "watch-123" }; + const out = toKanbanTask(httpDTO({ metadata })); + expect(out.isPRReview).toBe(true); + expect(out.isIssueWatch).toBe(false); + expect(toKanbanTask(wsPayload({ metadata }))).toEqual(out); + }); + + it("issue watch task: issue_watch_id + issue_url/issue_number mirrored on both shapes", () => { + const metadata = { + issue_watch_id: "watch-9", + issue_url: "https://github.com/owner/repo/issues/42", + issue_number: 42, + }; + const out = toKanbanTask(httpDTO({ metadata })); + expect(out.isIssueWatch).toBe(true); + expect(out.issueUrl).toBe("https://github.com/owner/repo/issues/42"); + expect(out.issueNumber).toBe(42); + expect(toKanbanTask(wsPayload({ metadata }))).toEqual(out); + }); + + it("repositoryId comes from nested HTTP repositories[0] or flat WS repository_id", () => { + expect(toKanbanTask(httpDTO()).repositoryId).toBe("repo-a"); + expect(toKanbanTask(wsPayload()).repositoryId).toBe("repo-a"); + }); + + it("missing repository on either shape: repositoryId is undefined", () => { + const http = httpDTO({ repositories: undefined }); + const ws = wsPayload({ repository_id: undefined }); + expect(toKanbanTask(http).repositoryId).toBeUndefined(); + expect(toKanbanTask(ws).repositoryId).toBeUndefined(); + }); + + it("null/empty/omitted metadata yields false flags and no issue fields", () => { + // WS omits the field entirely when Task.Metadata is nil; HTTP may send + // null/{}. All three must derive the same derived flags. + const cases: TaskLike[] = [ + httpDTO({ metadata: null }), + wsPayload({ metadata: undefined }), + wsPayload({ metadata: {} }), + ]; + const mapped = cases.map(toKanbanTask); + const first = mapped[0]; + expect(first.isPRReview).toBe(false); + expect(first.isIssueWatch).toBe(false); + expect(first.issueUrl).toBeUndefined(); + expect(first.issueNumber).toBeUndefined(); + for (const out of mapped.slice(1)) { + expect(out).toEqual(first); + } + }); + + it("picks id from `id` (HTTP) or `task_id` (WS)", () => { + expect(toKanbanTask(httpDTO({ id: "x", task_id: undefined })).id).toBe("x"); + expect(toKanbanTask(wsPayload({ id: undefined, task_id: "y" })).id).toBe("y"); + }); + + it("defaults isRemoteExecutor to false when missing", () => { + const http = httpDTO({ is_remote_executor: undefined }); + const ws = wsPayload({ is_remote_executor: undefined }); + expect(toKanbanTask(http).isRemoteExecutor).toBe(false); + expect(toKanbanTask(ws).isRemoteExecutor).toBe(false); + }); +}); diff --git a/apps/web/lib/kanban/map-task.ts b/apps/web/lib/kanban/map-task.ts new file mode 100644 index 0000000000..8cff76e0a1 --- /dev/null +++ b/apps/web/lib/kanban/map-task.ts @@ -0,0 +1,82 @@ +import { + isPRReviewFromMetadata, + isIssueWatchFromMetadata, + issueFieldsFromMetadata, +} from "@/lib/metadata-utils"; +import type { KanbanState } from "@/lib/state/slices/kanban/types"; +import type { TaskState, TaskSessionState } from "@/lib/types/http"; + +type KanbanTask = KanbanState["tasks"][number]; + +/** + * Shape accepted by {@link toKanbanTask}. Satisfied by both the HTTP Task DTO + * (which nests repositories) and the backend's `task.updated` WebSocket + * payload (which flattens `repository_id` and uses `task_id`). + * + * The single publisher / single mapper design relies on this shape being the + * contract; anyone changing what the backend emits needs to keep it in sync, + * and the parity test in `map-task.test.ts` will catch drift. + */ +export type TaskLike = { + id?: string; + task_id?: string; + workflow_step_id?: string; + title?: string; + description?: string | null; + position?: number; + state?: TaskState; + repositories?: Array<{ repository_id: string }>; + repository_id?: string; + primary_session_id?: string | null; + primary_session_state?: TaskSessionState | string | null; + session_count?: number | null; + review_status?: "pending" | "approved" | "changes_requested" | "rejected" | null; + primary_executor_id?: string | null; + primary_executor_type?: string | null; + primary_executor_name?: string | null; + is_remote_executor?: boolean; + parent_id?: string | null; + updated_at?: string; + created_at?: string; + metadata?: Record | null; +}; + +function pickRepositoryId(source: TaskLike): string | undefined { + return source.repository_id ?? source.repositories?.[0]?.repository_id ?? undefined; +} + +function pickId(source: TaskLike): string { + return (source.id ?? source.task_id ?? "") as string; +} + +/** + * Build a canonical {@link KanbanTask} from either an HTTP DTO or a WebSocket + * payload. Both paths share this helper so a single publisher change can never + * leave them out of sync again (cf. sidebar filter regressions where the HTTP + * snapshot derived `isPRReview` but the WS handler didn't). + */ +export function toKanbanTask(source: TaskLike): KanbanTask { + return { + id: pickId(source), + workflowStepId: source.workflow_step_id ?? "", + title: source.title ?? "", + description: source.description ?? undefined, + position: source.position ?? 0, + state: source.state, + repositoryId: pickRepositoryId(source), + primarySessionId: source.primary_session_id ?? undefined, + primarySessionState: source.primary_session_state ?? undefined, + sessionCount: source.session_count ?? undefined, + reviewStatus: source.review_status ?? undefined, + primaryExecutorId: source.primary_executor_id ?? undefined, + primaryExecutorType: source.primary_executor_type ?? undefined, + primaryExecutorName: source.primary_executor_name ?? undefined, + isRemoteExecutor: source.is_remote_executor ?? false, + parentTaskId: source.parent_id ?? undefined, + updatedAt: source.updated_at, + createdAt: source.created_at, + isPRReview: isPRReviewFromMetadata(source.metadata), + isIssueWatch: isIssueWatchFromMetadata(source.metadata), + ...issueFieldsFromMetadata(source.metadata), + } as KanbanTask; +} diff --git a/apps/web/lib/ws/handlers/tasks.ts b/apps/web/lib/ws/handlers/tasks.ts index a6815256b4..3a7cdbb197 100644 --- a/apps/web/lib/ws/handlers/tasks.ts +++ b/apps/web/lib/ws/handlers/tasks.ts @@ -4,69 +4,10 @@ import type { WsHandlers } from "@/lib/ws/handlers/types"; import type { KanbanState } from "@/lib/state/slices/kanban/types"; import { cleanupTaskStorage } from "@/lib/local-storage"; import { useContextFilesStore } from "@/lib/state/context-files-store"; +import { toKanbanTask, type TaskLike } from "@/lib/kanban/map-task"; type KanbanTask = KanbanState["tasks"][number]; -/** Falls back to existing value when incoming is null, undefined, or empty string. */ -function withFallback(value: T | null | undefined, fallback: T | undefined): T | undefined { - return value || fallback; -} - -/** Falls back only on null/undefined -- preserves 0 and other falsy non-string values. */ -function withNullishFallback( - value: T | null | undefined, - fallback: T | undefined, -): T | undefined { - return value ?? fallback; -} - -function buildNullableFields( - // eslint-disable-next-line @typescript-eslint/no-explicit-any - payload: any, - existing?: KanbanTask, -): Pick< - KanbanTask, - | "repositoryId" - | "primarySessionId" - | "primarySessionState" - | "sessionCount" - | "reviewStatus" - | "primaryExecutorId" - | "primaryExecutorType" - | "primaryExecutorName" - | "parentTaskId" - | "updatedAt" - | "createdAt" -> { - return { - repositoryId: withFallback(payload.repository_id, existing?.repositoryId), - primarySessionId: withFallback(payload.primary_session_id, existing?.primarySessionId), - primarySessionState: withFallback(payload.primary_session_state, existing?.primarySessionState), - sessionCount: withNullishFallback(payload.session_count, existing?.sessionCount), - reviewStatus: withFallback(payload.review_status, existing?.reviewStatus), - primaryExecutorId: withFallback(payload.primary_executor_id, existing?.primaryExecutorId), - primaryExecutorType: withFallback(payload.primary_executor_type, existing?.primaryExecutorType), - primaryExecutorName: withFallback(payload.primary_executor_name, existing?.primaryExecutorName), - parentTaskId: withFallback(payload.parent_id, existing?.parentTaskId), - updatedAt: withFallback(payload.updated_at, existing?.updatedAt), - createdAt: withFallback(payload.created_at, existing?.createdAt), - }; -} - -// eslint-disable-next-line @typescript-eslint/no-explicit-any -function buildTaskFromPayload(payload: any, existing?: KanbanTask): KanbanTask { - return { - id: payload.task_id, - workflowStepId: payload.workflow_step_id, - title: payload.title, - description: payload.description, - position: payload.position ?? 0, - state: payload.state, - isRemoteExecutor: payload.is_remote_executor ?? existing?.isRemoteExecutor ?? false, - ...buildNullableFields(payload, existing), - }; -} - function upsertTask(tasks: KanbanTask[], nextTask: KanbanTask): KanbanTask[] { const exists = tasks.some((task) => task.id === nextTask.id); return exists @@ -92,26 +33,31 @@ function upsertMultiTask(state: AppState, workflowId: string, task: KanbanTask): }; } +type TaskEventPayload = TaskLike & { + workflow_id: string; + is_ephemeral?: boolean; + archived_at?: string | null; +}; + /** Upsert a task in both single-kanban and multi-kanban snapshots. */ -// eslint-disable-next-line @typescript-eslint/no-explicit-any -function upsertTaskInBothKanbans(state: AppState, wfId: string, payload: any): AppState { +function upsertTaskInBothKanbans( + state: AppState, + wfId: string, + payload: TaskEventPayload, +): AppState { // Skip ephemeral tasks - they should never be added to kanban if (payload.is_ephemeral) { return state; } + const nextTask = toKanbanTask(payload); let next = state; if (state.kanban.workflowId === wfId) { - const existing = state.kanban.tasks.find((t) => t.id === payload.task_id); - const nextTask = buildTaskFromPayload(payload, existing); next = { ...next, kanban: { ...next.kanban, tasks: upsertTask(next.kanban.tasks, nextTask) } }; } - const snapshot = state.kanbanMulti.snapshots[wfId]; - if (snapshot) { - const existing = snapshot.tasks.find((t) => t.id === payload.task_id); - const nextTask = buildTaskFromPayload(payload, existing); + if (state.kanbanMulti.snapshots[wfId]) { next = upsertMultiTask(next, wfId, nextTask); }