refactor: unify task.updated via single publisher and shared mapper (#676)

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

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 {

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),

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)

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
 }
 

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",

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, "")

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

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 {

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

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

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<string>): 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) {

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> = {}): 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> = {}): 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);
+  });
+});

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<string, unknown> | 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;
+}

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<T>(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<T>(
-  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);
   }