diff --git a/apps/backend/internal/gateway/websocket/task_notifications.go b/apps/backend/internal/gateway/websocket/task_notifications.go index 7f25a123ef..9d975b3179 100644 --- a/apps/backend/internal/gateway/websocket/task_notifications.go +++ b/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()), + ) +} diff --git a/apps/backend/internal/task/service/service_events.go b/apps/backend/internal/task/service/service_events.go index aa6dd23d1b..e47cd00e8a 100644 --- a/apps/backend/internal/task/service/service_events.go +++ b/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 diff --git a/apps/backend/internal/task/service/service_events_test.go b/apps/backend/internal/task/service/service_events_test.go index 1bd3694f30..6a6a43c18f 100644 --- a/apps/backend/internal/task/service/service_events_test.go +++ b/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() diff --git a/apps/web/components/kanban-card-content.tsx b/apps/web/components/kanban-card-content.tsx index 7fc4457aac..301f246d57 100644 --- a/apps/web/components/kanban-card-content.tsx +++ b/apps/web/components/kanban-card-content.tsx @@ -1,6 +1,6 @@ "use client"; -import { useState } from "react"; +import { useEffect, useState } from "react"; import { CSS, type Transform } from "@dnd-kit/utilities"; import type { DraggableAttributes, DraggableSyntheticListeners } from "@dnd-kit/core"; import { @@ -20,9 +20,10 @@ import { KanbanCardDropdownMenuItems, type KanbanCardMenuEntry, } from "@/components/kanban-card-menu-items"; -import { useAppStore } from "@/components/state-provider"; +import { useAppStore, useAppStoreApi } from "@/components/state-provider"; import { RemoteCloudTooltip } from "@/components/task/remote-cloud-tooltip"; import { useTaskPendingClarification } from "@/hooks/use-task-pending-clarification"; +import { createDebugLogger, isDebug } from "@/lib/debug/log"; import { getTaskStateIcon, shouldShowTaskRunningSpinner, @@ -32,6 +33,8 @@ import { cn } from "@/lib/utils"; import { needsAction } from "@/lib/utils/needs-action"; import type { RepositoryChip, Task } from "@/components/kanban-card"; +const kanbanStatusDebug = createDebugLogger("kanban:task-status"); + type KanbanCardActionProps = { task: Task; showMaximizeButton?: boolean; @@ -217,10 +220,21 @@ function KanbanCardActions({ isArchiving, }: KanbanCardActionProps) { const [menuOpen, setMenuOpen] = useState(false); + const [storePrimarySessionState, setStorePrimarySessionState] = useState(null); + const storeApi = useAppStoreApi(); + const debugEnabled = isDebug(); const effectiveMenuOpen = menuOpen || Boolean(isDeleting) || Boolean(isArchiving); const hasPendingClarificationRequest = useTaskPendingClarification(task.primarySessionId); const showQuestionIcon = shouldUseQuestionTaskIcon(task.state, hasPendingClarificationRequest); const showRunningSpinner = shouldShowTaskRunningSpinner(task.state, task.primarySessionState); + const storeWouldShowRunningSpinner = + storePrimarySessionState === null + ? null + : shouldShowTaskRunningSpinner(task.state, storePrimarySessionState); + const hasSpinnerMismatch = + showRunningSpinner && + storeWouldShowRunningSpinner === false && + task.primarySessionState !== storePrimarySessionState; const statusIcon = showRunningSpinner ? ( ) : ( @@ -229,6 +243,45 @@ function KanbanCardActions({ const hasKnownSession = Boolean(task.primarySessionId) || Boolean(task.sessionCount && task.sessionCount > 0); + useEffect(() => { + if (!debugEnabled || !task.primarySessionId) { + setStorePrimarySessionState(null); + return; + } + + const primarySessionId = task.primarySessionId; + const readPrimarySessionState = () => + storeApi.getState().taskSessions.items[primarySessionId]?.state ?? null; + const syncPrimarySessionState = () => { + const nextState = readPrimarySessionState(); + setStorePrimarySessionState((current) => (current === nextState ? current : nextState)); + }; + + syncPrimarySessionState(); + return storeApi.subscribe(syncPrimarySessionState); + }, [debugEnabled, storeApi, task.primarySessionId]); + + useEffect(() => { + if (!hasSpinnerMismatch || !debugEnabled) return; + kanbanStatusDebug("spinner mismatch", { + task_id: task.id, + taskState: task.state ?? "-", + primarySessionId: task.primarySessionId ?? "-", + taskPrimarySessionState: task.primarySessionState ?? "-", + storePrimarySessionState: storePrimarySessionState ?? "-", + showSpinner: showRunningSpinner, + }); + }, [ + debugEnabled, + hasSpinnerMismatch, + showRunningSpinner, + storePrimarySessionState, + task.id, + task.primarySessionId, + task.primarySessionState, + task.state, + ]); + return (
{(showRunningSpinner || showQuestionIcon) && statusIcon} diff --git a/apps/web/lib/debug/log.ts b/apps/web/lib/debug/log.ts index 206049a075..85541f8a67 100644 --- a/apps/web/lib/debug/log.ts +++ b/apps/web/lib/debug/log.ts @@ -155,6 +155,20 @@ * handlers. A `turn.completed` line with no following * `[session:state] newState=WAITING_FOR_INPUT` is the * smoking gun: the turn ended but running-state stuck. + * [task-lifecycle:ws] task/session lifecycle WS cache merge trace: + * task.created / task.updated / task.state_changed + * logs payload-vs-existing primary session state, + * and session.state_changed logs whether kanban and + * multi-workflow snapshots were patched. For stale + * kanban spinners, look for task.updated preserving + * beforeTaskPrimaryState=RUNNING after the session + * store already moved to WAITING_FOR_INPUT. + * [kanban:task-status] Kanban card mismatch detector. Emits only when + * the card would render a spinner from + * task.primarySessionState while the loaded + * taskSessions row for the primary session would + * not spin. This is the direct smoking gun for + * sidebar-correct / kanban-still-spinning reports. * * Review / Diff pipeline (bug: Review or Diff panel shows empty when * commits exist — distinguishes "no cumulative diff" vs "missing PR data" diff --git a/apps/web/lib/ws/handlers/agent-session-diagnostics.ts b/apps/web/lib/ws/handlers/agent-session-diagnostics.ts new file mode 100644 index 0000000000..0448ba1d53 --- /dev/null +++ b/apps/web/lib/ws/handlers/agent-session-diagnostics.ts @@ -0,0 +1,95 @@ +import { createDebugLogger, isDebug } from "@/lib/debug/log"; +import type { KanbanState, WorkflowSnapshotData } from "@/lib/state/slices/kanban/types"; +import type { AppState } from "@/lib/state/store"; +import type { TaskSessionState } from "@/lib/types/http"; + +type KanbanTask = KanbanState["tasks"][number]; + +const lifecycleDebug = createDebugLogger("task-lifecycle:ws"); + +function findSnapshotTask( + snapshots: Record, + workflowId: string, + taskId: string, + sessionId: string, +): KanbanTask | undefined { + return snapshots[workflowId]?.tasks.find( + (task) => task.id === taskId && task.primarySessionId === sessionId, + ); +} + +function didPatchTaskPrimaryState( + beforeTask: KanbanTask | undefined, + afterTask: KanbanTask | undefined, + newState: TaskSessionState, +): boolean { + return ( + beforeTask?.primarySessionState !== newState && afterTask?.primarySessionState === newState + ); +} + +export function buildKanbanPrimarySessionSyncLog(params: { + beforeState: AppState; + afterState: AppState; + taskId: string; + sessionId: string; + newState: TaskSessionState; +}): { + beforeTask: KanbanTask | undefined; + patchedKanban: boolean; + patchedSnapshotIds: string[]; +} { + const beforeTask = + params.beforeState.kanban.tasks.find((task) => task.id === params.taskId) ?? + Object.values(params.beforeState.kanbanMulti.snapshots) + .flatMap((snapshot) => snapshot.tasks) + .find((task) => task.id === params.taskId); + const beforeKanbanTask = params.beforeState.kanban.tasks.find( + (task) => task.id === params.taskId && task.primarySessionId === params.sessionId, + ); + const afterKanbanTask = params.afterState.kanban.tasks.find( + (task) => task.id === params.taskId && task.primarySessionId === params.sessionId, + ); + const patchedSnapshotIds = Object.keys(params.afterState.kanbanMulti.snapshots).filter( + (workflowId) => { + const beforeSnapshotTask = findSnapshotTask( + params.beforeState.kanbanMulti.snapshots, + workflowId, + params.taskId, + params.sessionId, + ); + const afterSnapshotTask = findSnapshotTask( + params.afterState.kanbanMulti.snapshots, + workflowId, + params.taskId, + params.sessionId, + ); + return didPatchTaskPrimaryState(beforeSnapshotTask, afterSnapshotTask, params.newState); + }, + ); + return { + beforeTask, + patchedKanban: didPatchTaskPrimaryState(beforeKanbanTask, afterKanbanTask, params.newState), + patchedSnapshotIds, + }; +} + +export function logKanbanPrimarySessionSync(params: { + taskId: string; + sessionId: string; + newState: TaskSessionState; + beforeTask: KanbanTask | undefined; + patchedKanban: boolean; + patchedSnapshotIds: string[]; +}): void { + if (!isDebug()) return; + lifecycleDebug("session.state_changed kanban sync", { + task_id: params.taskId, + sessionId: params.sessionId, + newState: params.newState, + beforeTaskPrimarySessionId: params.beforeTask?.primarySessionId ?? "-", + beforeTaskPrimaryState: params.beforeTask?.primarySessionState ?? "-", + patchedKanban: params.patchedKanban, + patchedSnapshots: params.patchedSnapshotIds.join(",") || "-", + }); +} diff --git a/apps/web/lib/ws/handlers/agent-session-kanban-sync.ts b/apps/web/lib/ws/handlers/agent-session-kanban-sync.ts new file mode 100644 index 0000000000..d4522c8802 --- /dev/null +++ b/apps/web/lib/ws/handlers/agent-session-kanban-sync.ts @@ -0,0 +1,130 @@ +import type { StoreApi } from "zustand"; +import { isDebug } from "@/lib/debug/log"; +import type { KanbanState, WorkflowSnapshotData } from "@/lib/state/slices/kanban/types"; +import type { AppState } from "@/lib/state/store"; +import type { SessionId, TaskId, TaskSessionState } from "@/lib/types/http"; +import { + buildKanbanPrimarySessionSyncLog, + logKanbanPrimarySessionSync, +} from "@/lib/ws/handlers/agent-session-diagnostics"; + +function snapshotKanbanStateForLog(state: AppState): AppState { + return { + ...state, + kanban: { ...state.kanban }, + kanbanMulti: { + ...state.kanbanMulti, + snapshots: { ...state.kanbanMulti.snapshots }, + }, + }; +} + +function patchTaskPrimarySessionState( + tasks: KanbanState["tasks"], + taskId: string, + sessionId: string, + newState: TaskSessionState, +): KanbanState["tasks"] { + let changed = false; + const nextTasks = tasks.map((task) => { + if (task.id !== taskId || task.primarySessionId !== sessionId) return task; + if (task.primarySessionState === newState) return task; + changed = true; + return { ...task, primarySessionState: newState }; + }); + return changed ? nextTasks : tasks; +} + +function patchSnapshotPrimarySessionState( + snapshot: WorkflowSnapshotData, + taskId: string, + sessionId: string, + newState: TaskSessionState, +): WorkflowSnapshotData { + const tasks = patchTaskPrimarySessionState(snapshot.tasks, taskId, sessionId, newState); + return tasks === snapshot.tasks ? snapshot : { ...snapshot, tasks }; +} + +function patchWorkflowSnapshotPrimarySessionStates( + snapshots: Record, + taskId: string, + sessionId: string, + newState: TaskSessionState, +): { + nextSnapshots: Record; + snapshotsChanged: boolean; +} { + let snapshotsChanged = false; + const nextSnapshots = Object.fromEntries( + Object.entries(snapshots).map(([workflowId, snapshot]) => { + const nextSnapshot = patchSnapshotPrimarySessionState(snapshot, taskId, sessionId, newState); + if (nextSnapshot !== snapshot) snapshotsChanged = true; + return [workflowId, nextSnapshot]; + }), + ); + return { nextSnapshots, snapshotsChanged }; +} + +export function syncKanbanPrimarySessionState( + store: StoreApi, + taskId: TaskId, + sessionId: SessionId, + newState: TaskSessionState | undefined, +): void { + if (!newState) return; + + const state = store.getState(); + if (!state.kanban?.tasks || !state.kanbanMulti?.snapshots) return; + + const debugMode = isDebug(); + const beforeState = debugMode ? snapshotKanbanStateForLog(state) : state; + + store.setState((currentState) => { + if (!currentState.kanban?.tasks || !currentState.kanbanMulti?.snapshots) return currentState; + const nextKanbanTasks = patchTaskPrimarySessionState( + currentState.kanban.tasks, + taskId, + sessionId, + newState, + ); + const { nextSnapshots, snapshotsChanged } = patchWorkflowSnapshotPrimarySessionStates( + currentState.kanbanMulti.snapshots, + taskId, + sessionId, + newState, + ); + + if (nextKanbanTasks === currentState.kanban.tasks && !snapshotsChanged) return currentState; + + return { + ...currentState, + kanban: + nextKanbanTasks === currentState.kanban.tasks + ? currentState.kanban + : { ...currentState.kanban, tasks: nextKanbanTasks }, + kanbanMulti: snapshotsChanged + ? { + ...currentState.kanbanMulti, + snapshots: nextSnapshots, + } + : currentState.kanbanMulti, + }; + }); + + if (!debugMode) return; + const syncLog = buildKanbanPrimarySessionSyncLog({ + beforeState, + afterState: store.getState(), + taskId, + sessionId, + newState, + }); + logKanbanPrimarySessionSync({ + taskId, + sessionId, + newState, + beforeTask: syncLog.beforeTask, + patchedKanban: syncLog.patchedKanban, + patchedSnapshotIds: syncLog.patchedSnapshotIds, + }); +} diff --git a/apps/web/lib/ws/handlers/agent-session-kanban.test.ts b/apps/web/lib/ws/handlers/agent-session-kanban.test.ts index 82cec26416..ef03021eaa 100644 --- a/apps/web/lib/ws/handlers/agent-session-kanban.test.ts +++ b/apps/web/lib/ws/handlers/agent-session-kanban.test.ts @@ -3,12 +3,15 @@ import type { StoreApi } from "zustand"; import { registerTaskSessionHandlers } from "./agent-session"; import type { AppState } from "@/lib/state/store"; import type { TaskSessionStateChangedPayload } from "@/lib/types/backend"; -import { sessionId, taskId } from "@/lib/types/http"; +import { sessionId, taskId, type TaskSessionState } from "@/lib/types/http"; const STATE_CHANGED_EVENT = "session.state_changed"; const TASK_TITLE = "Running task"; -function makeStore(overrides: Partial = {}) { +function makeStore( + overrides: Partial = {}, + beforeApplyState?: (state: AppState) => void, +) { const state = { kanban: { workflowId: null, steps: [], tasks: [] }, kanbanMulti: { snapshots: {}, isLoading: false }, @@ -28,6 +31,7 @@ function makeStore(overrides: Partial = {}) { ...overrides, } as unknown as AppState; const setState = vi.fn((updater: unknown) => { + beforeApplyState?.(state); const next = typeof updater === "function" ? updater(state) : updater; if (next && next !== state) Object.assign(state, next); }); @@ -49,52 +53,55 @@ function makeMessage(payload: TaskSessionStateChangedPayload) { }; } +function makeKanbanTask(primarySessionId: string) { + return { + id: "t-1", + workflowStepId: "step-1", + title: TASK_TITLE, + position: 0, + primarySessionId, + primarySessionState: "WAITING_FOR_INPUT", + }; +} + +function makeTaskSession(id: string) { + return { + id: sessionId(id), + task_id: taskId("t-1"), + state: "WAITING_FOR_INPUT" as TaskSessionState, + started_at: "", + updated_at: "", + }; +} + +function makeSnapshotTaskStore(primarySessionId: string) { + const task = makeKanbanTask(primarySessionId); + return { + isLoading: false, + snapshots: { + "wf-1": { + workflowId: "wf-1", + workflowName: "Development", + steps: [], + tasks: [task], + }, + }, + }; +} + describe("session.state_changed -> primary kanban card state", () => { it("updates kanban card primary session state in workflow snapshots", () => { + const task = makeKanbanTask("s-1"); const store = makeStore({ kanban: { workflowId: "wf-1", steps: [], - tasks: [ - { - id: "t-1", - workflowStepId: "step-1", - title: TASK_TITLE, - position: 0, - primarySessionId: "s-1", - primarySessionState: "WAITING_FOR_INPUT", - }, - ], - }, - kanbanMulti: { - isLoading: false, - snapshots: { - "wf-1": { - workflowId: "wf-1", - workflowName: "Development", - steps: [], - tasks: [ - { - id: "t-1", - workflowStepId: "step-1", - title: TASK_TITLE, - position: 0, - primarySessionId: "s-1", - primarySessionState: "WAITING_FOR_INPUT", - }, - ], - }, - }, + tasks: [task], }, + kanbanMulti: makeSnapshotTaskStore("s-1"), taskSessions: { items: { - "s-1": { - id: sessionId("s-1"), - task_id: taskId("t-1"), - state: "WAITING_FOR_INPUT", - started_at: "", - updated_at: "", - }, + "s-1": makeTaskSession("s-1"), }, }, }); @@ -107,6 +114,64 @@ describe("session.state_changed -> primary kanban card state", () => { "RUNNING", ); }); + + it("logs the previous primary state when the task only exists in a workflow snapshot", () => { + const debugSpy = vi.spyOn(console, "debug").mockImplementation(() => undefined); + const store = makeStore({ + kanban: { + workflowId: "wf-other", + steps: [], + tasks: [], + }, + kanbanMulti: makeSnapshotTaskStore("s-1"), + taskSessions: { + items: { + "s-1": makeTaskSession("s-1"), + }, + }, + }); + const handler = registerTaskSessionHandlers(store)[STATE_CHANGED_EVENT]!; + + handler(makeMessage({ task_id: "t-1", session_id: "s-1", new_state: "RUNNING" })); + + expect( + debugSpy.mock.calls.some((call) => + String(call[0]).includes("beforeTaskPrimaryState=WAITING_FOR_INPUT"), + ), + ).toBe(true); + debugSpy.mockRestore(); + }); + + it("preserves concurrent store changes while syncing kanban session state", () => { + const task = makeKanbanTask("s-1"); + const store = makeStore( + { + kanban: { + workflowId: "wf-1", + steps: [], + tasks: [task], + }, + kanbanMulti: makeSnapshotTaskStore("s-1"), + taskSessions: { + items: { + "s-1": makeTaskSession("s-1"), + }, + }, + }, + (state) => { + state.tasks = { + ...state.tasks, + activeTaskId: taskId("concurrent-task"), + }; + }, + ); + const handler = registerTaskSessionHandlers(store)[STATE_CHANGED_EVENT]!; + + handler(makeMessage({ task_id: "t-1", session_id: "s-1", new_state: "RUNNING" })); + + expect(store.getState().kanban.tasks[0]?.primarySessionState).toBe("RUNNING"); + expect(store.getState().tasks.activeTaskId).toBe(taskId("concurrent-task")); + }); }); describe("session.state_changed -> non-primary kanban card state", () => { diff --git a/apps/web/lib/ws/handlers/agent-session.ts b/apps/web/lib/ws/handlers/agent-session.ts index c3d549bc2b..e4b5ad8c8d 100644 --- a/apps/web/lib/ws/handlers/agent-session.ts +++ b/apps/web/lib/ws/handlers/agent-session.ts @@ -10,7 +10,7 @@ import { type TaskSessionState, } from "@/lib/types/http"; import type { QueuedMessage } from "@/lib/state/slices/session/types"; -import type { KanbanState, WorkflowSnapshotData } from "@/lib/state/slices/kanban/types"; +import { syncKanbanPrimarySessionState } from "@/lib/ws/handlers/agent-session-kanban-sync"; const debug = createDebugLogger("session:state"); @@ -224,79 +224,6 @@ function maybeFanOutOfficeRefetch( setOfficeTrigger("agents"); } -function patchTaskPrimarySessionState( - tasks: KanbanState["tasks"], - taskId: string, - sessionId: string, - newState: TaskSessionState, -): KanbanState["tasks"] { - let changed = false; - const nextTasks = tasks.map((task) => { - if (task.id !== taskId || task.primarySessionId !== sessionId) return task; - if (task.primarySessionState === newState) return task; - changed = true; - return { ...task, primarySessionState: newState }; - }); - return changed ? nextTasks : tasks; -} - -function patchSnapshotPrimarySessionState( - snapshot: WorkflowSnapshotData, - taskId: string, - sessionId: string, - newState: TaskSessionState, -): WorkflowSnapshotData { - const tasks = patchTaskPrimarySessionState(snapshot.tasks, taskId, sessionId, newState); - return tasks === snapshot.tasks ? snapshot : { ...snapshot, tasks }; -} - -function syncKanbanPrimarySessionState( - store: StoreApi, - taskId: TaskId, - sessionId: SessionId, - newState: TaskSessionState | undefined, -): void { - if (!newState) return; - - store.setState((state) => { - const nextKanbanTasks = patchTaskPrimarySessionState( - state.kanban.tasks, - taskId, - sessionId, - newState, - ); - let snapshotsChanged = false; - const nextSnapshots = Object.fromEntries( - Object.entries(state.kanbanMulti.snapshots).map(([workflowId, snapshot]) => { - const nextSnapshot = patchSnapshotPrimarySessionState( - snapshot, - taskId, - sessionId, - newState, - ); - if (nextSnapshot !== snapshot) snapshotsChanged = true; - return [workflowId, nextSnapshot]; - }), - ); - - if (nextKanbanTasks === state.kanban.tasks && !snapshotsChanged) return state; - - return { - ...state, - kanban: - nextKanbanTasks === state.kanban.tasks - ? state.kanban - : { ...state.kanban, tasks: nextKanbanTasks }, - kanbanMulti: snapshotsChanged - ? { - ...state.kanbanMulti, - snapshots: nextSnapshots, - } - : state.kanbanMulti, - }; - }); -} - /** Extract context window data from payload metadata and store it. */ // eslint-disable-next-line @typescript-eslint/no-explicit-any function extractContextWindow(store: StoreApi, sessionId: string, payload: any): void { diff --git a/apps/web/lib/ws/handlers/task-lifecycle-diagnostics.test.ts b/apps/web/lib/ws/handlers/task-lifecycle-diagnostics.test.ts new file mode 100644 index 0000000000..dd4f79f4c4 --- /dev/null +++ b/apps/web/lib/ws/handlers/task-lifecycle-diagnostics.test.ts @@ -0,0 +1,282 @@ +import { describe, expect, it, vi } from "vitest"; +import type { StoreApi } from "zustand"; +import type { AppState } from "@/lib/state/store"; +import { didPreservePrimaryState, registerTasksHandlers } from "./tasks"; + +type Listener = (state: AppState) => void; + +function makeStore(initial: Partial = {}) { + let state = { + kanban: { workflowId: "wf1", steps: [], tasks: [] }, + kanbanMulti: { snapshots: {}, isLoading: false }, + tasks: { + activeTaskId: null, + activeSessionId: null, + pinnedSessionId: null, + lastSessionByTaskId: {}, + }, + taskSessionsByTask: { itemsByTaskId: {}, loadedByTaskId: {}, loadingByTaskId: {} }, + environmentIdBySessionId: {}, + setActiveSessionAuto: vi.fn(), + removeTaskFromSidebarPrefs: vi.fn(), + ...initial, + } as unknown as AppState; + + const listeners = new Set(); + return { + getState: () => state, + setState: (updater: AppState | ((s: AppState) => AppState)) => { + const next = + typeof updater === "function" ? (updater as (s: AppState) => AppState)(state) : updater; + state = { ...state, ...next }; + for (const l of listeners) l(state); + }, + subscribe: (l: Listener) => { + listeners.add(l); + return () => listeners.delete(l); + }, + destroy: vi.fn(), + getInitialState: vi.fn(), + } as unknown as StoreApi & { getState: () => AppState }; +} + +function makeTask(id: string, primarySessionId: string | null, workflowId = "wf1") { + return { + task_id: id, + workflow_id: workflowId, + workflow_step_id: "step1", + title: "Test", + description: "", + state: "IN_PROGRESS", + primary_session_id: primarySessionId, + is_ephemeral: false, + } as Record; +} + +function makeUpdatedMessage(payload: Record) { + return { + id: "msg-1", + type: "notification" as const, + action: "task.updated" as const, + payload, + } as Parameters["task.updated"]>>[0]; +} + +function makeCreatedMessage(payload: Record) { + return { + id: "msg-1", + type: "notification" as const, + action: "task.created" as const, + payload, + } as Parameters["task.created"]>>[0]; +} + +function makeStateChangedMessage(payload: Record) { + return { + id: "msg-1", + type: "notification" as const, + action: "task.state_changed" as const, + payload, + } as Parameters["task.state_changed"]>>[0]; +} + +const existingTask = { + id: "t1", + workflowStepId: "step1", + title: "Old", + position: 0, + state: "IN_PROGRESS" as const, + primarySessionId: "sess-1", + primarySessionState: "RUNNING", +}; + +function taskWithPrimaryState(primarySessionState: string | undefined) { + return { + ...existingTask, + primarySessionState, + }; +} + +describe("didPreservePrimaryState", () => { + it("detects only omitted payload primary state that stayed preserved", () => { + expect( + didPreservePrimaryState( + { workflow_id: "wf1", primary_session_state: "RUNNING" }, + taskWithPrimaryState("COMPLETED"), + taskWithPrimaryState("COMPLETED"), + ), + ).toBe(false); + expect( + didPreservePrimaryState( + { workflow_id: "wf1" }, + taskWithPrimaryState(undefined), + taskWithPrimaryState("RUNNING"), + ), + ).toBe(false); + expect( + didPreservePrimaryState( + { workflow_id: "wf1" }, + taskWithPrimaryState("RUNNING"), + taskWithPrimaryState("RUNNING"), + ), + ).toBe(true); + expect( + didPreservePrimaryState( + { workflow_id: "wf1" }, + taskWithPrimaryState("RUNNING"), + taskWithPrimaryState("COMPLETED"), + ), + ).toBe(false); + }); +}); + +describe("task lifecycle diagnostics", () => { + it("upserts task state and primary session state into both kanban stores", () => { + const store = makeStore({ + kanban: { + workflowId: "wf1", + steps: [], + tasks: [existingTask], + } as unknown as AppState["kanban"], + kanbanMulti: { + isLoading: false, + snapshots: { + wf1: { workflowId: "wf1", workflowName: "WF1", steps: [], tasks: [existingTask] }, + }, + } as unknown as AppState["kanbanMulti"], + }); + + registerTasksHandlers(store)["task.state_changed"]!( + makeStateChangedMessage({ + ...makeTask("t1", "sess-1"), + state: "REVIEW", + primary_session_state: "WAITING_FOR_INPUT", + }), + ); + + const state = store.getState(); + const kanbanTask = state.kanban.tasks.find((task) => task.id === "t1"); + const snapshotTask = state.kanbanMulti.snapshots.wf1.tasks.find((task) => task.id === "t1"); + expect(kanbanTask?.state).toBe("REVIEW"); + expect(kanbanTask?.primarySessionState).toBe("WAITING_FOR_INPUT"); + expect(snapshotTask?.state).toBe("REVIEW"); + expect(snapshotTask?.primarySessionState).toBe("WAITING_FOR_INPUT"); + }); + + it("skips ephemeral task state changes", () => { + const store = makeStore(); + + registerTasksHandlers(store)["task.state_changed"]!( + makeStateChangedMessage({ + ...makeTask("t-ephemeral", "sess-1"), + is_ephemeral: true, + state: "REVIEW", + }), + ); + + expect(store.getState().kanban.tasks).toEqual([]); + }); + + it("skips ephemeral task creation", () => { + const store = makeStore(); + + registerTasksHandlers(store)["task.created"]!( + makeCreatedMessage({ + ...makeTask("t-ephemeral", "sess-1"), + is_ephemeral: true, + }), + ); + + expect(store.getState().kanban.tasks).toEqual([]); + }); + + it("logs whether a task update preserved an omitted primary session state", () => { + const debugSpy = vi.spyOn(console, "debug").mockImplementation(() => undefined); + const store = makeStore({ + kanban: { + workflowId: "wf1", + steps: [], + tasks: [existingTask], + } as unknown as AppState["kanban"], + }); + const handlers = registerTasksHandlers(store); + + handlers["task.updated"]!(makeUpdatedMessage(makeTask("t1", "sess-1"))); + handlers["task.updated"]!( + makeUpdatedMessage({ + ...makeTask("t1", "sess-1"), + primary_session_state: null, + }), + ); + + const logs = debugSpy.mock.calls.map((call) => String(call[0])); + expect(logs.some((line) => line.includes("preservedPrimaryState=true"))).toBe(true); + expect(logs.some((line) => line.includes("payloadPrimarySessionState=null"))).toBe(true); + expect(logs.some((line) => line.includes("preservedPrimaryState=false"))).toBe(true); + debugSpy.mockRestore(); + }); +}); + +describe("task primary-session clears", () => { + it("does not preserve primary session fields when the payload explicitly clears them", () => { + const store = makeStore({ + kanban: { + workflowId: "wf1", + steps: [], + tasks: [existingTask], + } as unknown as AppState["kanban"], + kanbanMulti: { + isLoading: false, + snapshots: { + wf1: { workflowId: "wf1", workflowName: "WF1", steps: [], tasks: [existingTask] }, + }, + } as unknown as AppState["kanbanMulti"], + }); + + registerTasksHandlers(store)["task.updated"]!( + makeUpdatedMessage({ + ...makeTask("t1", null), + primary_session_state: null, + }), + ); + + const state = store.getState(); + const kanbanTask = state.kanban.tasks.find((task) => task.id === "t1"); + const snapshotTask = state.kanbanMulti.snapshots.wf1.tasks.find((task) => task.id === "t1"); + expect(kanbanTask?.primarySessionId).toBeUndefined(); + expect(kanbanTask?.primarySessionState).toBeUndefined(); + expect(snapshotTask?.primarySessionId).toBeUndefined(); + expect(snapshotTask?.primarySessionState).toBeUndefined(); + }); + + it("does not preserve primary session fields on explicit state-change clears", () => { + const store = makeStore({ + kanban: { + workflowId: "wf1", + steps: [], + tasks: [existingTask], + } as unknown as AppState["kanban"], + kanbanMulti: { + isLoading: false, + snapshots: { + wf1: { workflowId: "wf1", workflowName: "WF1", steps: [], tasks: [existingTask] }, + }, + } as unknown as AppState["kanbanMulti"], + }); + + registerTasksHandlers(store)["task.state_changed"]!( + makeStateChangedMessage({ + ...makeTask("t1", null), + primary_session_state: null, + }), + ); + + const state = store.getState(); + const kanbanTask = state.kanban.tasks.find((task) => task.id === "t1"); + const snapshotTask = state.kanbanMulti.snapshots.wf1.tasks.find((task) => task.id === "t1"); + expect(kanbanTask?.primarySessionId).toBeUndefined(); + expect(kanbanTask?.primarySessionState).toBeUndefined(); + expect(snapshotTask?.primarySessionId).toBeUndefined(); + expect(snapshotTask?.primarySessionState).toBeUndefined(); + }); +}); diff --git a/apps/web/lib/ws/handlers/tasks.ts b/apps/web/lib/ws/handlers/tasks.ts index e5bd9dd46e..a59a4ac894 100644 --- a/apps/web/lib/ws/handlers/tasks.ts +++ b/apps/web/lib/ws/handlers/tasks.ts @@ -1,4 +1,5 @@ import type { StoreApi } from "zustand"; +import { createDebugLogger, isDebug } from "@/lib/debug/log"; import type { AppState } from "@/lib/state/store"; import type { WsHandlers } from "@/lib/ws/handlers/types"; import type { KanbanState } from "@/lib/state/slices/kanban/types"; @@ -16,24 +17,52 @@ import { } from "@/lib/ws/handlers/agent-session"; type KanbanTask = KanbanState["tasks"][number]; +const lifecycleDebug = createDebugLogger("task-lifecycle:ws"); -function mergeTaskUpdate(existing: KanbanTask | undefined, nextTask: KanbanTask): KanbanTask { +function hasPayloadField(payload: TaskEventPayload, field: keyof TaskEventPayload): boolean { + return Object.prototype.hasOwnProperty.call(payload, field); +} + +function mergeTaskUpdate( + existing: KanbanTask | undefined, + nextTask: KanbanTask, + payload: TaskEventPayload, +): KanbanTask { if (!existing) return nextTask; - return { + const merged = { ...nextTask, ...mergeTaskRepositoryFields(existing, nextTask), }; + if (!hasPayloadField(payload, "primary_session_id") && nextTask.primarySessionId === undefined) { + merged.primarySessionId = existing.primarySessionId; + } + if ( + !hasPayloadField(payload, "primary_session_state") && + nextTask.primarySessionState === undefined + ) { + merged.primarySessionState = existing.primarySessionState; + } + return merged; } -function upsertTask(tasks: KanbanTask[], nextTask: KanbanTask): KanbanTask[] { +function upsertTask( + tasks: KanbanTask[], + nextTask: KanbanTask, + payload: TaskEventPayload, +): KanbanTask[] { const existing = tasks.find((task) => task.id === nextTask.id); - const merged = mergeTaskUpdate(existing, nextTask); + const merged = mergeTaskUpdate(existing, nextTask, payload); return existing ? tasks.map((task) => (task.id === nextTask.id ? merged : task)) : [...tasks, merged]; } -function upsertMultiTask(state: AppState, workflowId: string, task: KanbanTask): AppState { +function upsertMultiTask( + state: AppState, + workflowId: string, + task: KanbanTask, + payload: TaskEventPayload, +): AppState { const snapshot = state.kanbanMulti.snapshots[workflowId]; if (!snapshot) return state; return { @@ -44,7 +73,7 @@ function upsertMultiTask(state: AppState, workflowId: string, task: KanbanTask): ...state.kanbanMulti.snapshots, [workflowId]: { ...snapshot, - tasks: upsertTask(snapshot.tasks, task), + tasks: upsertTask(snapshot.tasks, task, payload), }, }, }, @@ -73,11 +102,14 @@ function upsertTaskInBothKanbans( let next = state; if (state.kanban.workflowId === wfId) { - next = { ...next, kanban: { ...next.kanban, tasks: upsertTask(next.kanban.tasks, nextTask) } }; + next = { + ...next, + kanban: { ...next.kanban, tasks: upsertTask(next.kanban.tasks, nextTask, payload) }, + }; } if (state.kanbanMulti.snapshots[wfId]) { - next = upsertMultiTask(next, wfId, nextTask); + next = upsertMultiTask(next, wfId, nextTask, payload); } return next; @@ -94,6 +126,61 @@ function findTaskInState(state: AppState, taskId: string): KanbanTask | undefine return undefined; } +function taskEventIdForLog(payload: TaskEventPayload): string { + return payload.task_id ?? payload.id ?? ""; +} + +function valueForLog(value: string | null | undefined): string { + return value ?? "-"; +} + +function payloadPrimaryStateForLog(payload: TaskEventPayload): string { + if (payload.primary_session_state === undefined) return "-"; + return payload.primary_session_state ?? "null"; +} + +function taskStateForLog(task: KanbanTask | undefined): string { + return task?.state ?? "-"; +} + +function taskPrimaryStateForLog(task: KanbanTask | undefined): string { + return task?.primarySessionState ?? "-"; +} + +export function didPreservePrimaryState( + payload: TaskEventPayload, + beforeTask: KanbanTask | undefined, + afterTask: KanbanTask | undefined, +): boolean { + if (payload.primary_session_state !== undefined) return false; + const previousPrimaryState = beforeTask?.primarySessionState; + if (previousPrimaryState === undefined) return false; + return previousPrimaryState === afterTask?.primarySessionState; +} + +function logTaskMerge( + action: "task.created" | "task.updated" | "task.state_changed", + beforeState: AppState, + afterState: AppState, + payload: TaskEventPayload, +): void { + if (!isDebug()) return; + const taskId = taskEventIdForLog(payload); + const beforeTask = findTaskInState(beforeState, taskId); + const afterTask = findTaskInState(afterState, taskId); + lifecycleDebug(`${action} merge`, { + task_id: taskId, + payloadState: valueForLog(payload.state), + payloadPrimarySessionId: valueForLog(payload.primary_session_id), + payloadPrimarySessionState: payloadPrimaryStateForLog(payload), + beforeTaskState: taskStateForLog(beforeTask), + beforeTaskPrimaryState: taskPrimaryStateForLog(beforeTask), + afterTaskState: taskStateForLog(afterTask), + afterTaskPrimaryState: taskPrimaryStateForLog(afterTask), + preservedPrimaryState: didPreservePrimaryState(payload, beforeTask, afterTask), + }); +} + /** Remove a task from both single-kanban and multi-kanban snapshots. */ function removeTaskFromBothKanbans(state: AppState, taskId: string): AppState { let next = state; @@ -166,6 +253,10 @@ function redirectAwayFromRemovedTask(taskId: string): void { } type TaskUpdatedMessage = Parameters>[0]; +type TaskCreatedMessage = Parameters>[0]; +type TaskStateChangedMessage = Parameters>[0]; +type TaskUpsertMessage = TaskCreatedMessage | TaskStateChangedMessage; +type TaskUpsertAction = "task.created" | "task.state_changed"; function handleTaskUpdated(store: StoreApi, message: TaskUpdatedMessage): void { // Skip ephemeral tasks (e.g., quick chat) - they shouldn't appear on the Kanban board @@ -215,6 +306,7 @@ function handleTaskUpdated(store: StoreApi, message: TaskUpdatedMessag // This makes workflow profile switches transparent for unpinned users // without yanking users off a live session they deliberately selected. const afterState = store.getState(); + logTaskMerge("task.updated", beforeState, afterState, message.payload); const newPrimary = findTaskInState(afterState, taskId)?.primarySessionId ?? null; if ( newPrimary && @@ -228,14 +320,25 @@ function handleTaskUpdated(store: StoreApi, message: TaskUpdatedMessag } } +function handleTaskUpsert( + action: TaskUpsertAction, + store: StoreApi, + message: TaskUpsertMessage, +): void { + // Skip ephemeral tasks (e.g., quick chat) - they shouldn't appear on the Kanban board + if (message.payload.is_ephemeral) return; + + const beforeState = store.getState(); + store.setState((state) => + upsertTaskInBothKanbans(state, message.payload.workflow_id, message.payload), + ); + logTaskMerge(action, beforeState, store.getState(), message.payload); +} + export function registerTasksHandlers(store: StoreApi): WsHandlers { return { "task.created": (message) => { - // Skip ephemeral tasks (e.g., quick chat) - they shouldn't appear on the Kanban board - if (message.payload.is_ephemeral) return; - store.setState((state) => - upsertTaskInBothKanbans(state, message.payload.workflow_id, message.payload), - ); + handleTaskUpsert("task.created", store, message); }, "task.updated": (message) => handleTaskUpdated(store, message), "task.deleted": (message) => { @@ -299,12 +402,7 @@ export function registerTasksHandlers(store: StoreApi): WsHandlers { } }, "task.state_changed": (message) => { - // Skip ephemeral tasks (e.g., quick chat) - they shouldn't appear on the Kanban board - if (message.payload.is_ephemeral) return; - - store.setState((state) => - upsertTaskInBothKanbans(state, message.payload.workflow_id, message.payload), - ); + handleTaskUpsert("task.state_changed", store, message); }, }; }