fix: add task lifecycle diagnostics (#1571)

apps/backend/internal/gateway/websocket/task_notifications.go

@@ -120,6 +120,9 @@ func (b *TaskEventBroadcaster) subscribe(eventBus bus.EventBus, subject, action
 				}
 			}
 		}
+		if data, ok := event.Data.(map[string]interface{}); ok {
+			b.logLifecycleBroadcast(action, data, sessionID)
+		}
 
 		switch action {
 		case ws.ActionSessionAgentctlStarting, ws.ActionSessionAgentctlReady, ws.ActionSessionAgentctlError:
@@ -157,3 +160,24 @@ func (b *TaskEventBroadcaster) subscribe(eventBus bus.EventBus, subject, action
 	}
 	b.subscriptions = append(b.subscriptions, sub)
 }
+
+func (b *TaskEventBroadcaster) logLifecycleBroadcast(action string, data map[string]interface{}, sessionID string) {
+	switch action {
+	case ws.ActionTaskCreated, ws.ActionTaskUpdated, ws.ActionTaskStateChanged,
+		ws.ActionTaskDeleted, ws.ActionSessionStateChanged:
+	default:
+		return
+	}
+	b.logger.Debug("ws lifecycle broadcast",
+		zap.String("action", action),
+		zap.Any("task_id", data["task_id"]),
+		zap.String("session_id", sessionID),
+		zap.Any("state", data["state"]),
+		zap.Any("old_state", data["old_state"]),
+		zap.Any("new_state", data["new_state"]),
+		zap.Any("primary_session_id", data["primary_session_id"]),
+		zap.Any("primary_session_state", data["primary_session_state"]),
+		zap.Any("updated_at", data["updated_at"]),
+		zap.Int("connected_clients", b.hub.GetClientCount()),
+	)
+}

apps/backend/internal/task/service/service_events.go

@@ -97,7 +97,28 @@ func (s *Service) publishTaskEventWithExtra(ctx context.Context, eventType strin
 			zap.String("event_type", eventType),
 			zap.String("task_id", task.ID),
 			zap.Error(err))
+		return
 	}
+	s.logTaskLifecycleEventPublished(eventType, task, data)
+}
+
+func (s *Service) logTaskLifecycleEventPublished(eventType string, task *models.Task, data map[string]interface{}) {
+	switch eventType {
+	case events.TaskCreated, events.TaskUpdated, events.TaskStateChanged, events.TaskDeleted:
+	default:
+		return
+	}
+	s.logger.Debug("task lifecycle event published",
+		zap.String("event_type", eventType),
+		zap.String("task_id", task.ID),
+		zap.Any("state", data["state"]),
+		zap.Any("workflow_step_id", data["workflow_step_id"]),
+		zap.Any("primary_session_id", data["primary_session_id"]),
+		zap.Any("primary_session_state", data["primary_session_state"]),
+		zap.Any("session_count", data["session_count"]),
+		zap.Any("old_state", data["old_state"]),
+		zap.Any("new_state", data["new_state"]),
+	)
 }
 
 // addTaskSessionEventFields merges session count, primary session info, and
@@ -116,6 +137,8 @@ func (s *Service) addTaskSessionEventFields(ctx context.Context, taskID string,
 	}
 	sessionInfo, ok := primarySessionInfoMap[taskID]
 	if !ok || sessionInfo == nil {
+		data["primary_session_id"] = nil
+		data["primary_session_state"] = nil
 		return
 	}
 	data["primary_session_id"] = sessionInfo.ID
@@ -124,6 +147,8 @@ func (s *Service) addTaskSessionEventFields(ctx context.Context, taskID string,
 	}
 	if sessionInfo.State != "" {
 		data["primary_session_state"] = string(sessionInfo.State)
+	} else {
+		data["primary_session_state"] = nil
 	}
 	if sessionInfo.ExecutorID != "" {
 		data["primary_executor_id"] = sessionInfo.ExecutorID

apps/backend/internal/task/service/service_events_test.go

@@ -14,6 +14,11 @@ type failingTaskRepoRepository struct {
 	err error
 }
 
+type primarySessionInfoRepository struct {
+	repository.SessionRepository
+	info map[string]*models.TaskSession
+}
+
 type taskEventTestRepository interface {
 	CreateWorkspace(context.Context, *models.Workspace) error
 	CreateWorkflow(context.Context, *models.Workflow) error
@@ -24,6 +29,14 @@ func (r failingTaskRepoRepository) ListTaskRepositories(ctx context.Context, tas
 	return nil, r.err
 }
 
+func (r primarySessionInfoRepository) GetPrimarySessionInfoByTaskIDs(ctx context.Context, taskIDs []string) (map[string]*models.TaskSession, error) {
+	return r.info, nil
+}
+
+func (r primarySessionInfoRepository) GetSessionCountsByTaskIDs(ctx context.Context, taskIDs []string) (map[string]int, error) {
+	return map[string]int{}, nil
+}
+
 // TestPublishTaskUpdated_FallbackRepositoryID exercises the DB fallback in
 // taskRepositoriesForEvent: orchestrator-originated events load the task via the
 // raw repo.GetTask, which does not populate Repositories. The publisher must
@@ -103,6 +116,45 @@ func TestPublishTaskUpdated_EmitsEmptyRepositories(t *testing.T) {
 	}
 }
 
+func TestPublishTaskUpdated_EmitsNullPrimarySessionFieldsWhenNoPrimaryExists(t *testing.T) {
+	svc, eventBus, repo := createTestService(t)
+	ctx := context.Background()
+
+	createTaskWithoutRepositories(t, ctx, repo)
+	eventBus.ClearEvents()
+
+	svc.PublishTaskUpdated(ctx, &models.Task{
+		ID: "task-1", WorkspaceID: "ws-1", WorkflowID: "wf-1", WorkflowStepID: "step-1",
+	})
+
+	data := singlePublishedEventData(t, eventBus)
+	if value, ok := data["primary_session_id"]; !ok || value != nil {
+		t.Fatalf("primary_session_id = %#v, want explicit nil", value)
+	}
+	if value, ok := data["primary_session_state"]; !ok || value != nil {
+		t.Fatalf("primary_session_state = %#v, want explicit nil", value)
+	}
+}
+
+func TestAddTaskSessionEventFields_EmitsNullPrimarySessionStateWhenEmpty(t *testing.T) {
+	svc, _, _ := createTestService(t)
+	svc.sessions = primarySessionInfoRepository{
+		info: map[string]*models.TaskSession{
+			"task-1": {ID: "session-1", TaskID: "task-1"},
+		},
+	}
+	data := map[string]interface{}{}
+
+	svc.addTaskSessionEventFields(context.Background(), "task-1", data)
+
+	if value := data["primary_session_id"]; value != "session-1" {
+		t.Fatalf("primary_session_id = %#v, want session-1", value)
+	}
+	if value, ok := data["primary_session_state"]; !ok || value != nil {
+		t.Fatalf("primary_session_state = %#v, want explicit nil", value)
+	}
+}
+
 func TestPublishTaskUpdated_OmitsRepositoriesOnLookupError(t *testing.T) {
 	svc, eventBus, repo := createTestService(t)
 	ctx := context.Background()