From 05b550e122e68928f784682e23774d5f551504e5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20Flor=C3=AAncio?= Date: Fri, 10 Jul 2026 23:29:40 +0100 Subject: [PATCH] fix: show pending task input before messages load (#1663) --- .../backend/internal/backendapp/boot_state.go | 50 +++++++- .../internal/backendapp/boot_state_routes.go | 31 ++--- .../internal/backendapp/helpers_test.go | 83 +++++++++++- .../orchestrator/workflow_e2e_test.go | 5 +- apps/backend/internal/task/dto/dto.go | 105 ++++++++-------- .../task/handlers/process_handlers_test.go | 3 + .../task/handlers/task_http_handlers.go | 55 +++++++- .../task/handlers/workflow_handlers.go | 2 +- apps/backend/internal/task/models/models.go | 9 ++ .../internal/task/repository/interface.go | 1 + .../task/repository/sqlite/message.go | 104 +++++++++++++++ .../task/repository/sqlite/message_test.go | 88 +++++++++++++ .../internal/task/service/service_events.go | 23 ++++ .../task/service/service_events_test.go | 102 +++++++++++++++ .../internal/task/service/service_sessions.go | 6 + apps/web/components/kanban-card-content.tsx | 5 +- apps/web/components/kanban-card.tsx | 8 +- .../components/kanban/graph2-step-node.tsx | 5 +- .../kanban/swimlane-graph-content.tsx | 10 +- .../session-task-switcher-sheet-hooks.ts | 118 +++++++++++------- .../task-session-sidebar-aggregate.test.ts | 37 ++++++ .../task/task-session-sidebar-aggregate.ts | 32 ++++- .../components/task/task-session-sidebar.tsx | 5 +- .../task/sidebar-pending-question.spec.ts | 88 +++++++++++++ .../use-task-pending-clarification.test.tsx | 30 +++++ .../hooks/use-task-pending-clarification.ts | 25 +++- apps/web/lib/kanban/map-task.test.ts | 22 ++++ apps/web/lib/kanban/map-task.ts | 11 +- apps/web/lib/ssr/mapper.test.ts | 61 +++++++++ apps/web/lib/ssr/mapper.ts | 2 + apps/web/lib/state/slices/kanban/types.ts | 3 +- apps/web/lib/types/backend.ts | 4 + apps/web/lib/types/http.ts | 3 + .../ws/handlers/agent-session-kanban-sync.ts | 15 ++- apps/web/lib/ws/handlers/kanban.ts | 5 + apps/web/lib/ws/handlers/tasks.ts | 6 + 36 files changed, 1024 insertions(+), 138 deletions(-) create mode 100644 apps/web/e2e/tests/task/sidebar-pending-question.spec.ts create mode 100644 apps/web/lib/ssr/mapper.test.ts diff --git a/apps/backend/internal/backendapp/boot_state.go b/apps/backend/internal/backendapp/boot_state.go index 66e4acf60b..76c2d04965 100644 --- a/apps/backend/internal/backendapp/boot_state.go +++ b/apps/backend/internal/backendapp/boot_state.go @@ -516,12 +516,16 @@ func (b bootStateBuilder) workflowSnapshotState(ctx context.Context, workflow *t b.logBootError("list home workflow tasks", err) return nil, false } - taskStates := make([]map[string]any, 0, len(tasks)) + visibleTasks := make([]*taskmodels.Task, 0, len(tasks)) for _, task := range tasks { if task == nil || task.IsEphemeral || task.WorkflowStepID == "" { continue } - taskStates = append(taskStates, mapKanbanTaskState(taskdto.FromTask(task))) + visibleTasks = append(visibleTasks, task) + } + taskStates := make([]map[string]any, 0, len(visibleTasks)) + for _, task := range b.taskDTOsWithSessionInfo(ctx, visibleTasks) { + taskStates = append(taskStates, mapKanbanTaskState(task)) } return map[string]any{ "workflowId": workflow.ID, @@ -623,6 +627,11 @@ func (b bootStateBuilder) taskDTOsWithSessionInfo(ctx context.Context, tasks []* b.logBootError("get task detail primary session info", err) return taskDTOs(tasks) } + pendingActionsBySession, err := b.bootPendingActionsForWaitingPrimarySessions(ctx, primaryInfoByTask) + if err != nil { + b.logBootError("get task detail pending actions", err) + pendingActionsBySession = map[string]taskmodels.TaskPendingAction{} + } result := make([]taskdto.TaskDTO, 0, len(tasks)) for _, task := range tasks { if task == nil { @@ -654,6 +663,7 @@ func (b bootStateBuilder) taskDTOsWithSessionInfo(ctx context.Context, tasks []* info.agentName, info.workingDirectory, info.sessionState, + bootPendingActionPtr(info.sessionID, pendingActionsBySession), )) } return result @@ -670,6 +680,7 @@ func taskDTOs(tasks []*taskmodels.Task) []taskdto.TaskDTO { } type bootSessionInfoFields struct { + sessionID *string reviewStatus taskmodels.ReviewStatus sessionState *string executorID *string @@ -684,6 +695,10 @@ func bootSessionInfo(session *taskmodels.TaskSession) bootSessionInfoFields { if session == nil { return info } + if session.ID != "" { + value := session.ID + info.sessionID = &value + } info.reviewStatus = session.ReviewStatus if session.State != "" { value := string(session.State) @@ -714,6 +729,37 @@ func bootSessionInfo(session *taskmodels.TaskSession) bootSessionInfoFields { return info } +func (b bootStateBuilder) bootPendingActionsForWaitingPrimarySessions( + ctx context.Context, + primaryInfoByTask map[string]*taskmodels.TaskSession, +) (map[string]taskmodels.TaskPendingAction, error) { + sessionIDs := make([]string, 0, len(primaryInfoByTask)) + for _, info := range primaryInfoByTask { + if info != nil && info.State == taskmodels.TaskSessionStateWaitingForInput { + sessionIDs = append(sessionIDs, info.ID) + } + } + if len(sessionIDs) == 0 { + return map[string]taskmodels.TaskPendingAction{}, nil + } + return b.p.taskSvc.GetPendingActionsForSessions(ctx, sessionIDs) +} + +func bootPendingActionPtr( + sessionID *string, + pendingActionsBySession map[string]taskmodels.TaskPendingAction, +) *string { + if sessionID == nil { + return nil + } + action, ok := pendingActionsBySession[*sessionID] + if !ok { + return nil + } + value := string(action) + return &value +} + func (b bootStateBuilder) taskDetailInitialState( ctx context.Context, task *taskmodels.Task, diff --git a/apps/backend/internal/backendapp/boot_state_routes.go b/apps/backend/internal/backendapp/boot_state_routes.go index dc6ebce9ba..0ac80cc7d3 100644 --- a/apps/backend/internal/backendapp/boot_state_routes.go +++ b/apps/backend/internal/backendapp/boot_state_routes.go @@ -540,21 +540,22 @@ func mapKanbanTaskState(task taskdto.TaskDTO) map[string]any { }) } return map[string]any{ - "id": task.ID, - "workflowStepId": task.WorkflowStepID, - "title": task.Title, - "description": task.Description, - "position": task.Position, - "state": task.State, - "repositoryId": primaryRepositoryID, - "repositories": repositories, - "primarySessionId": task.PrimarySessionID, - "primarySessionState": task.PrimarySessionState, - "sessionCount": task.SessionCount, - "reviewStatus": nullString(string(task.ReviewStatus)), - "parentTaskId": nullString(task.ParentID), - "updatedAt": task.UpdatedAt, - "createdAt": task.CreatedAt, + "id": task.ID, + "workflowStepId": task.WorkflowStepID, + "title": task.Title, + "description": task.Description, + "position": task.Position, + "state": task.State, + "repositoryId": primaryRepositoryID, + "repositories": repositories, + "primarySessionId": task.PrimarySessionID, + "primarySessionState": task.PrimarySessionState, + "primarySessionPendingAction": task.PrimarySessionPendingAction, + "sessionCount": task.SessionCount, + "reviewStatus": nullString(string(task.ReviewStatus)), + "parentTaskId": nullString(task.ParentID), + "updatedAt": task.UpdatedAt, + "createdAt": task.CreatedAt, } } diff --git a/apps/backend/internal/backendapp/helpers_test.go b/apps/backend/internal/backendapp/helpers_test.go index e4a2c3e2bd..b7cf0b3dc3 100644 --- a/apps/backend/internal/backendapp/helpers_test.go +++ b/apps/backend/internal/backendapp/helpers_test.go @@ -290,7 +290,8 @@ func (s *shutdownDeadlineExecutor) ShouldApplyPreferredShell() bool { return tru func (s *shutdownDeadlineExecutor) IsAlwaysResumable() bool { return false } func TestBootInitialStateHomeIncludesKanbanFirstPaintState(t *testing.T) { - taskSvc, workflowSvc := newBootStateTestServices(t) + harness := newBootStateTestHarness(t) + taskSvc, workflowSvc := harness.taskSvc, harness.workflowSvc ctx := context.Background() workspaces, err := taskSvc.ListWorkspaces(ctx) @@ -307,6 +308,57 @@ func TestBootInitialStateHomeIncludesKanbanFirstPaintState(t *testing.T) { if len(workflows) == 0 { t.Fatal("expected seeded default workflow") } + steps, err := workflowSvc.ListStepsByWorkflow(ctx, workflows[0].ID) + if err != nil { + t.Fatalf("ListStepsByWorkflow: %v", err) + } + if len(steps) == 0 { + t.Fatal("expected seeded default workflow step") + } + task, err := taskSvc.CreateTask(ctx, &taskservice.CreateTaskRequest{ + WorkspaceID: workspaces[0].ID, + WorkflowID: workflows[0].ID, + WorkflowStepID: steps[0].ID, + Title: "Boot home task", + }) + if err != nil { + t.Fatalf("CreateTask: %v", err) + } + now := time.Now().UTC() + if err := harness.taskRepo.CreateTaskSession(ctx, &models.TaskSession{ + ID: "boot-session-waiting", + TaskID: task.ID, + State: models.TaskSessionStateWaitingForInput, + IsPrimary: true, + StartedAt: now, + UpdatedAt: now, + }); err != nil { + t.Fatalf("CreateTaskSession: %v", err) + } + if err := harness.taskRepo.CreateTurn(ctx, &models.Turn{ + ID: "boot-turn-waiting", + TaskID: task.ID, + TaskSessionID: "boot-session-waiting", + StartedAt: now, + CreatedAt: now, + UpdatedAt: now, + }); err != nil { + t.Fatalf("CreateTurn: %v", err) + } + if err := harness.taskRepo.CreateMessage(ctx, &models.Message{ + ID: "boot-clarification", + TaskID: task.ID, + TaskSessionID: "boot-session-waiting", + TurnID: "boot-turn-waiting", + AuthorType: models.MessageAuthorAgent, + Type: models.MessageTypeClarificationRequest, + Content: "Need input", + Metadata: map[string]interface{}{"status": "pending"}, + CreatedAt: now, + UpdatedAt: now, + }); err != nil { + t.Fatalf("CreateMessage: %v", err) + } state := bootInitialState( ctx, @@ -346,8 +398,10 @@ func TestBootInitialStateHomeIncludesKanbanFirstPaintState(t *testing.T) { Title string `json:"title"` } `json:"steps"` Tasks []struct { - ID string `json:"id"` - WorkflowStepID string `json:"workflowStepId"` + ID string `json:"id"` + WorkflowStepID string `json:"workflowStepId"` + PrimarySessionState *string `json:"primarySessionState"` + PrimarySessionPendingAction *string `json:"primarySessionPendingAction"` } `json:"tasks"` } `json:"snapshots"` IsLoading bool `json:"isLoading"` @@ -359,8 +413,10 @@ func TestBootInitialStateHomeIncludesKanbanFirstPaintState(t *testing.T) { Title string `json:"title"` } `json:"steps"` Tasks []struct { - ID string `json:"id"` - WorkflowStepID string `json:"workflowStepId"` + ID string `json:"id"` + WorkflowStepID string `json:"workflowStepId"` + PrimarySessionState *string `json:"primarySessionState"` + PrimarySessionPendingAction *string `json:"primarySessionPendingAction"` } `json:"tasks"` IsLoading bool `json:"isLoading"` } `json:"kanban"` @@ -378,9 +434,26 @@ func TestBootInitialStateHomeIncludesKanbanFirstPaintState(t *testing.T) { if _, ok := decoded.KanbanMulti.Snapshots[workflows[0].ID]; !ok { t.Fatalf("expected snapshot for workflow %q in boot payload", workflows[0].ID) } + snapshotTask := decoded.KanbanMulti.Snapshots[workflows[0].ID].Tasks[0] + if snapshotTask.PrimarySessionState == nil || *snapshotTask.PrimarySessionState != "WAITING_FOR_INPUT" { + t.Fatalf("snapshot primarySessionState = %#v, want WAITING_FOR_INPUT", snapshotTask.PrimarySessionState) + } + if snapshotTask.PrimarySessionPendingAction == nil || *snapshotTask.PrimarySessionPendingAction != "clarification" { + t.Fatalf( + "snapshot primarySessionPendingAction = %#v, want clarification", + snapshotTask.PrimarySessionPendingAction, + ) + } if decoded.Kanban.WorkflowID != workflows[0].ID { t.Fatalf("active kanban workflow = %q, want %q", decoded.Kanban.WorkflowID, workflows[0].ID) } + activeTask := decoded.Kanban.Tasks[0] + if activeTask.PrimarySessionPendingAction == nil || *activeTask.PrimarySessionPendingAction != "clarification" { + t.Fatalf( + "active primarySessionPendingAction = %#v, want clarification", + activeTask.PrimarySessionPendingAction, + ) + } if decoded.KanbanMulti.IsLoading || decoded.Kanban.IsLoading { t.Fatal("boot payload should mark kanban data loaded") } diff --git a/apps/backend/internal/orchestrator/workflow_e2e_test.go b/apps/backend/internal/orchestrator/workflow_e2e_test.go index d71044fcaf..7a5b966e26 100644 --- a/apps/backend/internal/orchestrator/workflow_e2e_test.go +++ b/apps/backend/internal/orchestrator/workflow_e2e_test.go @@ -423,7 +423,10 @@ func TestProcessOnTurnCompleteViaEngine_NonTerminalStepWritesTaskStateReview(t * setSessionExecID(t, repo, "s1", "exec-1") setSessionState(t, ctx, repo, "s1", models.TaskSessionStateRunning) - agentMgr := &mockAgentManager{repoForExecutionLookup: repo} + agentMgr := &mockAgentManager{ + repoForExecutionLookup: repo, + isAgentRunning: true, + } svc := createEngineService(t, repo, sg, agentMgr) taskRepo := svc.taskRepo.(*mockTaskRepo) seedMockTaskState(taskRepo, "t1", v1.TaskStateInProgress) diff --git a/apps/backend/internal/task/dto/dto.go b/apps/backend/internal/task/dto/dto.go index 52adafefda..2eea5b8808 100644 --- a/apps/backend/internal/task/dto/dto.go +++ b/apps/backend/internal/task/dto/dto.go @@ -119,31 +119,32 @@ type EnvironmentDTO struct { } type TaskDTO struct { - ID string `json:"id"` - WorkspaceID string `json:"workspace_id"` - WorkflowID string `json:"workflow_id"` - WorkflowStepID string `json:"workflow_step_id"` - Title string `json:"title"` - Description string `json:"description"` - State v1.TaskState `json:"state"` - Priority string `json:"priority"` - Repositories []TaskRepositoryDTO `json:"repositories,omitempty"` - Position int `json:"position"` - PrimarySessionID *string `json:"primary_session_id,omitempty"` - SessionCount *int `json:"session_count,omitempty"` - ReviewStatus models.ReviewStatus `json:"review_status,omitempty"` - PrimaryExecutorID *string `json:"primary_executor_id,omitempty"` - PrimaryExecutorType *string `json:"primary_executor_type,omitempty"` - PrimaryExecutorName *string `json:"primary_executor_name,omitempty"` - PrimaryAgentName *string `json:"primary_agent_name,omitempty"` - PrimaryWorkingDirectory *string `json:"primary_working_directory,omitempty"` - PrimarySessionState *string `json:"primary_session_state,omitempty"` - IsRemoteExecutor bool `json:"is_remote_executor,omitempty"` - ParentID string `json:"parent_id,omitempty"` - ArchivedAt *time.Time `json:"archived_at,omitempty"` - CreatedAt time.Time `json:"created_at"` - UpdatedAt time.Time `json:"updated_at"` - Metadata map[string]interface{} `json:"metadata,omitempty"` + ID string `json:"id"` + WorkspaceID string `json:"workspace_id"` + WorkflowID string `json:"workflow_id"` + WorkflowStepID string `json:"workflow_step_id"` + Title string `json:"title"` + Description string `json:"description"` + State v1.TaskState `json:"state"` + Priority string `json:"priority"` + Repositories []TaskRepositoryDTO `json:"repositories,omitempty"` + Position int `json:"position"` + PrimarySessionID *string `json:"primary_session_id,omitempty"` + SessionCount *int `json:"session_count,omitempty"` + ReviewStatus models.ReviewStatus `json:"review_status,omitempty"` + PrimaryExecutorID *string `json:"primary_executor_id,omitempty"` + PrimaryExecutorType *string `json:"primary_executor_type,omitempty"` + PrimaryExecutorName *string `json:"primary_executor_name,omitempty"` + PrimaryAgentName *string `json:"primary_agent_name,omitempty"` + PrimaryWorkingDirectory *string `json:"primary_working_directory,omitempty"` + PrimarySessionState *string `json:"primary_session_state,omitempty"` + PrimarySessionPendingAction *string `json:"primary_session_pending_action"` + IsRemoteExecutor bool `json:"is_remote_executor,omitempty"` + ParentID string `json:"parent_id,omitempty"` + ArchivedAt *time.Time `json:"archived_at,omitempty"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` + Metadata map[string]interface{} `json:"metadata,omitempty"` // Office extensions AssigneeAgentProfileID string `json:"assignee_agent_profile_id,omitempty"` @@ -539,7 +540,7 @@ func FromTask(task *models.Task) TaskDTO { // FromTaskWithPrimarySession converts a task model to a TaskDTO, including the primary session ID. func FromTaskWithPrimarySession(task *models.Task, primarySessionID *string) TaskDTO { - return FromTaskWithSessionInfo(task, primarySessionID, nil, models.ReviewStatusNone, nil, nil, nil, nil, nil, nil) + return FromTaskWithSessionInfo(task, primarySessionID, nil, models.ReviewStatusNone, nil, nil, nil, nil, nil, nil, nil) } // FromTaskWithSessionInfo converts a task model to a TaskDTO, including session information. @@ -554,6 +555,7 @@ func FromTaskWithSessionInfo( primaryAgentName *string, primaryWorkingDirectory *string, primarySessionState *string, + primarySessionPendingAction *string, ) TaskDTO { // Convert repositories var repositories []TaskRepositoryDTO @@ -572,31 +574,32 @@ func FromTaskWithSessionInfo( } return TaskDTO{ - ID: task.ID, - WorkspaceID: task.WorkspaceID, - WorkflowID: task.WorkflowID, - WorkflowStepID: task.WorkflowStepID, - Title: task.Title, - Description: task.Description, - State: task.State, - Priority: task.Priority, - Repositories: repositories, - Position: task.Position, - PrimarySessionID: primarySessionID, - SessionCount: sessionCount, - ReviewStatus: reviewStatus, - PrimaryExecutorID: primaryExecutorID, - PrimaryExecutorType: primaryExecutorType, - PrimaryExecutorName: primaryExecutorName, - PrimaryAgentName: primaryAgentName, - PrimaryWorkingDirectory: primaryWorkingDirectory, - PrimarySessionState: primarySessionState, - IsRemoteExecutor: primaryExecutorType != nil && models.IsRemoteExecutorType(models.ExecutorType(*primaryExecutorType)), - ParentID: task.ParentID, - ArchivedAt: task.ArchivedAt, - CreatedAt: task.CreatedAt, - UpdatedAt: task.UpdatedAt, - Metadata: task.Metadata, + ID: task.ID, + WorkspaceID: task.WorkspaceID, + WorkflowID: task.WorkflowID, + WorkflowStepID: task.WorkflowStepID, + Title: task.Title, + Description: task.Description, + State: task.State, + Priority: task.Priority, + Repositories: repositories, + Position: task.Position, + PrimarySessionID: primarySessionID, + SessionCount: sessionCount, + ReviewStatus: reviewStatus, + PrimaryExecutorID: primaryExecutorID, + PrimaryExecutorType: primaryExecutorType, + PrimaryExecutorName: primaryExecutorName, + PrimaryAgentName: primaryAgentName, + PrimaryWorkingDirectory: primaryWorkingDirectory, + PrimarySessionState: primarySessionState, + PrimarySessionPendingAction: primarySessionPendingAction, + IsRemoteExecutor: primaryExecutorType != nil && models.IsRemoteExecutorType(models.ExecutorType(*primaryExecutorType)), + ParentID: task.ParentID, + ArchivedAt: task.ArchivedAt, + CreatedAt: task.CreatedAt, + UpdatedAt: task.UpdatedAt, + Metadata: task.Metadata, // Office extensions. AssigneeAgentProfileID is a read-time // projection from workflow_step_participants (ADR 0005 Wave F); // the repo's task SELECTs hydrate it via a correlated subquery. diff --git a/apps/backend/internal/task/handlers/process_handlers_test.go b/apps/backend/internal/task/handlers/process_handlers_test.go index 4b5b5476d6..3c856bd3b3 100644 --- a/apps/backend/internal/task/handlers/process_handlers_test.go +++ b/apps/backend/internal/task/handlers/process_handlers_test.go @@ -203,6 +203,9 @@ func (m *mockRepository) FindMessageByPendingIDAndQuestion(ctx context.Context, func (m *mockRepository) FindPendingClarificationMessagesBySessionID(ctx context.Context, sessionID string) ([]*models.Message, error) { return nil, nil } +func (m *mockRepository) GetPendingActionsBySessionIDs(ctx context.Context, sessionIDs []string) (map[string]models.TaskPendingAction, error) { + return make(map[string]models.TaskPendingAction), nil +} func (m *mockRepository) UpdateMessage(ctx context.Context, message *models.Message) error { return nil } diff --git a/apps/backend/internal/task/handlers/task_http_handlers.go b/apps/backend/internal/task/handlers/task_http_handlers.go index 0edd3cbf88..d44fc4e2e4 100644 --- a/apps/backend/internal/task/handlers/task_http_handlers.go +++ b/apps/backend/internal/task/handlers/task_http_handlers.go @@ -11,6 +11,7 @@ import ( "github.com/gin-gonic/gin" "github.com/kandev/kandev/internal/common/constants" + "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/orchestrator" "github.com/kandev/kandev/internal/task/dto" "github.com/kandev/kandev/internal/task/models" @@ -90,7 +91,12 @@ func (h *TaskHandlers) httpListTasksByWorkspace(c *gin.Context) { // method (the persisted ExecutorSnapshot JSON uses different keys), so the // batch loader alone can't supply them without a regression. Two queries // total, down from three pre-batch. -func buildTaskDTOsWithSessionInfo(ctx context.Context, svc *service.Service, tasks []*models.Task) ([]dto.TaskDTO, error) { +func buildTaskDTOsWithSessionInfo( + ctx context.Context, + svc *service.Service, + log *logger.Logger, + tasks []*models.Task, +) ([]dto.TaskDTO, error) { if len(tasks) == 0 { return []dto.TaskDTO{}, nil } @@ -106,6 +112,11 @@ func buildTaskDTOsWithSessionInfo(ctx context.Context, svc *service.Service, tas if err != nil { return nil, err } + pendingActionsBySession, err := pendingActionsForWaitingPrimarySessions(ctx, svc, primarySessionInfoMap) + if err != nil { + log.Warn("failed to load pending actions for task list, using empty map", zap.Error(err)) + pendingActionsBySession = map[string]models.TaskPendingAction{} + } result := make([]dto.TaskDTO, 0, len(tasks)) for _, task := range tasks { sessions := sessionsByTask[task.ID] @@ -133,12 +144,14 @@ func buildTaskDTOsWithSessionInfo(ctx context.Context, svc *service.Service, tas si.agentName, si.workingDirectory, si.sessionState, + pendingActionPtr(si.sessionID, pendingActionsBySession), )) } return result, nil } type sessionInfoFields struct { + sessionID *string reviewStatus models.ReviewStatus sessionState *string executorID *string @@ -153,6 +166,10 @@ func extractSessionInfo(info *models.TaskSession) sessionInfoFields { if info == nil { return si } + if info.ID != "" { + val := info.ID + si.sessionID = &val + } si.reviewStatus = info.ReviewStatus if info.State != "" { val := string(info.State) @@ -183,8 +200,40 @@ func extractSessionInfo(info *models.TaskSession) sessionInfoFields { return si } +func pendingActionsForWaitingPrimarySessions( + ctx context.Context, + svc *service.Service, + primarySessionInfoMap map[string]*models.TaskSession, +) (map[string]models.TaskPendingAction, error) { + sessionIDs := make([]string, 0, len(primarySessionInfoMap)) + for _, info := range primarySessionInfoMap { + if info != nil && info.State == models.TaskSessionStateWaitingForInput { + sessionIDs = append(sessionIDs, info.ID) + } + } + if len(sessionIDs) == 0 { + return map[string]models.TaskPendingAction{}, nil + } + return svc.GetPendingActionsForSessions(ctx, sessionIDs) +} + +func pendingActionPtr( + sessionID *string, + pendingActionsBySession map[string]models.TaskPendingAction, +) *string { + if sessionID == nil { + return nil + } + action, ok := pendingActionsBySession[*sessionID] + if !ok { + return nil + } + value := string(action) + return &value +} + func (h *TaskHandlers) toTaskDTOsWithSessionInfo(ctx context.Context, tasks []*models.Task) ([]dto.TaskDTO, error) { - return buildTaskDTOsWithSessionInfo(ctx, h.service, tasks) + return buildTaskDTOsWithSessionInfo(ctx, h.service, h.logger, tasks) } func (h *TaskHandlers) httpGetTask(c *gin.Context) { @@ -193,7 +242,7 @@ func (h *TaskHandlers) httpGetTask(c *gin.Context) { handleNotFound(c, h.logger, err, "task not found") return } - dtos, err := buildTaskDTOsWithSessionInfo(c.Request.Context(), h.service, []*models.Task{task}) + dtos, err := buildTaskDTOsWithSessionInfo(c.Request.Context(), h.service, h.logger, []*models.Task{task}) if err != nil { h.logger.Error("failed to build task DTO with session info", zap.Error(err)) c.JSON(http.StatusOK, dto.FromTask(task)) diff --git a/apps/backend/internal/task/handlers/workflow_handlers.go b/apps/backend/internal/task/handlers/workflow_handlers.go index 6c23d48757..865977de8c 100644 --- a/apps/backend/internal/task/handlers/workflow_handlers.go +++ b/apps/backend/internal/task/handlers/workflow_handlers.go @@ -497,5 +497,5 @@ func (h *WorkflowHandlers) convertTasksWithPrimarySessions( ctx context.Context, tasks []*models.Task, ) ([]dto.TaskDTO, error) { - return buildTaskDTOsWithSessionInfo(ctx, h.service, tasks) + return buildTaskDTOsWithSessionInfo(ctx, h.service, h.logger, tasks) } diff --git a/apps/backend/internal/task/models/models.go b/apps/backend/internal/task/models/models.go index ffe7763b98..2c4e5b4d71 100644 --- a/apps/backend/internal/task/models/models.go +++ b/apps/backend/internal/task/models/models.go @@ -528,6 +528,15 @@ const ( PermissionStatusExpired PermissionStatus = "expired" ) +// TaskPendingAction is the compact task-list projection for a primary session +// blocked on user input. +type TaskPendingAction string + +const ( + TaskPendingActionClarification TaskPendingAction = "clarification" + TaskPendingActionPermission TaskPendingAction = "permission" +) + // Message represents a message in a task session type Message struct { ID string `json:"id"` diff --git a/apps/backend/internal/task/repository/interface.go b/apps/backend/internal/task/repository/interface.go index d300232498..76c31a5d11 100644 --- a/apps/backend/internal/task/repository/interface.go +++ b/apps/backend/internal/task/repository/interface.go @@ -122,6 +122,7 @@ type MessageRepository interface { FindMessagesByPendingID(ctx context.Context, pendingID string) ([]*models.Message, error) FindMessageByPendingIDAndQuestion(ctx context.Context, sessionID, pendingID, questionID string) (*models.Message, error) FindPendingClarificationMessagesBySessionID(ctx context.Context, sessionID string) ([]*models.Message, error) + GetPendingActionsBySessionIDs(ctx context.Context, sessionIDs []string) (map[string]models.TaskPendingAction, error) UpdateMessage(ctx context.Context, message *models.Message) error ListMessages(ctx context.Context, sessionID string) ([]*models.Message, error) ListMessagesByTurnID(ctx context.Context, turnID string) ([]*models.Message, error) diff --git a/apps/backend/internal/task/repository/sqlite/message.go b/apps/backend/internal/task/repository/sqlite/message.go index afc04cfa4c..7cc1b89e4a 100644 --- a/apps/backend/internal/task/repository/sqlite/message.go +++ b/apps/backend/internal/task/repository/sqlite/message.go @@ -377,6 +377,110 @@ func (r *Repository) FindPendingClarificationMessagesBySessionID(ctx context.Con return result, err } +// GetPendingActionsBySessionIDs returns the compact pending-action projection +// for sessions whose task-list state needs to render before messages are loaded. +func (r *Repository) GetPendingActionsBySessionIDs(ctx context.Context, sessionIDs []string) (map[string]models.TaskPendingAction, error) { + result := make(map[string]models.TaskPendingAction) + if len(sessionIDs) == 0 { + return result, nil + } + placeholders := make([]string, len(sessionIDs)) + args := make([]interface{}, 0, len(sessionIDs)*2) + for i, id := range sessionIDs { + placeholders[i] = "?" + args = append(args, id) + } + for _, id := range sessionIDs { + args = append(args, id) + } + query := pendingActionsBySessionQuery(r.ro.DriverName(), placeholders) + rows, err := r.ro.QueryxContext(ctx, r.ro.Rebind(query), args...) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + for rows.Next() { + var sessionID, action string + if err := rows.Scan(&sessionID, &action); err != nil { + return nil, err + } + if action == string(models.TaskPendingActionClarification) { + result[sessionID] = models.TaskPendingActionClarification + continue + } + if _, hasClarification := result[sessionID]; hasClarification { + continue + } + if action == string(models.TaskPendingActionPermission) { + result[sessionID] = models.TaskPendingActionPermission + } + } + return result, rows.Err() +} + +func pendingActionsBySessionQuery(driverName string, placeholders []string) string { + placeholderList := strings.Join(placeholders, ",") + statusExpr := dialect.JSONExtract(driverName, "m.metadata", "status") + latestOrderExpr := pendingActionMessageOrder(driverName, "") + permissionOrderExpr := pendingActionMessageOrder(driverName, "m") + return fmt.Sprintf(` + WITH latest_message AS ( + SELECT task_session_id, turn_id + FROM ( + SELECT task_session_id, + turn_id, + ROW_NUMBER() OVER ( + PARTITION BY task_session_id + ORDER BY created_at DESC, %s DESC + ) AS rn + FROM task_session_messages + WHERE task_session_id IN (%s) + ) ranked + WHERE rn = 1 + ), + pending_clarifications AS ( + SELECT DISTINCT m.task_session_id, 'clarification' AS action + FROM task_session_messages m + JOIN latest_message latest + ON latest.task_session_id = m.task_session_id + AND latest.turn_id = m.turn_id + WHERE m.task_session_id IN (%s) + AND m.type = 'clarification_request' + AND COALESCE(%s, '') IN ('', 'pending') + ), + latest_permissions AS ( + SELECT m.task_session_id, + COALESCE(%s, '') AS status, + ROW_NUMBER() OVER ( + PARTITION BY m.task_session_id + ORDER BY m.created_at DESC, %s DESC + ) AS rn + FROM task_session_messages m + JOIN latest_message latest + ON latest.task_session_id = m.task_session_id + AND latest.turn_id = m.turn_id + WHERE m.type = 'permission_request' + ) + SELECT task_session_id, action + FROM pending_clarifications + UNION ALL + SELECT task_session_id, 'permission' AS action + FROM latest_permissions + WHERE rn = 1 AND status IN ('', 'pending') + `, latestOrderExpr, placeholderList, placeholderList, statusExpr, statusExpr, permissionOrderExpr) +} + +func pendingActionMessageOrder(driverName string, qualifier string) string { + column := "id" + if driverName == dialect.SQLite3 { + column = "rowid" + } + if qualifier == "" { + return column + } + return qualifier + "." + column +} + // FindMessageByPendingIDAndQuestion finds the message for a specific (pending_id, // question_id) pair within a session. Used to flip per-question status (answered / // rejected) on multi-question clarification bundles. The persistence layer stores diff --git a/apps/backend/internal/task/repository/sqlite/message_test.go b/apps/backend/internal/task/repository/sqlite/message_test.go index 7c09220173..50658c0734 100644 --- a/apps/backend/internal/task/repository/sqlite/message_test.go +++ b/apps/backend/internal/task/repository/sqlite/message_test.go @@ -159,3 +159,91 @@ func TestCountToolCallMessagesBySession_Multi(t *testing.T) { t.Errorf("s3 should be omitted (zero tool_call rows), got %d", got["s3"]) } } + +func createPendingActionMessage( + t *testing.T, + repo *Repository, + id string, + taskID string, + sessionID string, + turnID string, + msgType models.MessageType, + status string, + createdAt time.Time, +) { + t.Helper() + metadata := map[string]interface{}{} + if status != "" { + metadata["status"] = status + } + if err := repo.CreateMessage(context.Background(), &models.Message{ + ID: id, + TaskSessionID: sessionID, + TaskID: taskID, + TurnID: turnID, + AuthorType: models.MessageAuthorAgent, + Content: id, + Type: msgType, + Metadata: metadata, + CreatedAt: createdAt, + }); err != nil { + t.Fatalf("CreateMessage(%s): %v", id, err) + } +} + +func TestGetPendingActionsBySessionIDs(t *testing.T) { + repo := newRepoForSessionTests(t) + ctx := context.Background() + now := time.Now().UTC() + + seedForMsgTest(t, repo, "task-clar", "sess-clar", "turn-clar") + createPendingActionMessage(t, repo, "perm-clar", "task-clar", "sess-clar", "turn-clar", models.MessageTypePermissionRequest, "", now) + createPendingActionMessage(t, repo, "clar-clar", "task-clar", "sess-clar", "turn-clar", models.MessageTypeClarificationRequest, "pending", now.Add(time.Second)) + + seedForMsgTest(t, repo, "task-resolved", "sess-resolved", "turn-resolved") + createPendingActionMessage(t, repo, "perm-old", "task-resolved", "sess-resolved", "turn-resolved", models.MessageTypePermissionRequest, "pending", now) + createPendingActionMessage(t, repo, "perm-new", "task-resolved", "sess-resolved", "turn-resolved", models.MessageTypePermissionRequest, "approved", now.Add(time.Second)) + + seedForMsgTest(t, repo, "task-perm", "sess-perm", "turn-perm") + createPendingActionMessage(t, repo, "perm-pending", "task-perm", "sess-perm", "turn-perm", models.MessageTypePermissionRequest, "pending", now) + + seedForMsgTest(t, repo, "task-perm-tie", "sess-perm-tie", "turn-perm-tie") + createPendingActionMessage(t, repo, "z-approved", "task-perm-tie", "sess-perm-tie", "turn-perm-tie", models.MessageTypePermissionRequest, "approved", now) + createPendingActionMessage(t, repo, "a-pending", "task-perm-tie", "sess-perm-tie", "turn-perm-tie", models.MessageTypePermissionRequest, "pending", now) + + seedForMsgTest(t, repo, "task-stale", "sess-stale", "turn-stale") + createPendingActionMessage(t, repo, "perm-stale", "task-stale", "sess-stale", "turn-stale", models.MessageTypePermissionRequest, "pending", now) + createPendingActionMessage(t, repo, "clar-stale", "task-stale", "sess-stale", "turn-stale", models.MessageTypeClarificationRequest, "pending", now) + seedForMsgTest(t, repo, "task-stale", "sess-stale", "turn-current") + createPendingActionMessage(t, repo, "message-current", "task-stale", "sess-stale", "turn-current", models.MessageTypeMessage, "", now.Add(time.Second)) + + got, err := repo.GetPendingActionsBySessionIDs(ctx, []string{ + "sess-clar", + "sess-resolved", + "sess-perm", + "sess-perm-tie", + "sess-stale", + "sess-missing", + }) + if err != nil { + t.Fatalf("GetPendingActionsBySessionIDs: %v", err) + } + if got["sess-clar"] != models.TaskPendingActionClarification { + t.Fatalf("sess-clar action = %q, want clarification", got["sess-clar"]) + } + if _, ok := got["sess-resolved"]; ok { + t.Fatalf("sess-resolved should not have a pending action: %#v", got["sess-resolved"]) + } + if got["sess-perm"] != models.TaskPendingActionPermission { + t.Fatalf("sess-perm action = %q, want permission", got["sess-perm"]) + } + if got["sess-perm-tie"] != models.TaskPendingActionPermission { + t.Fatalf("sess-perm-tie action = %q, want permission from last inserted row", got["sess-perm-tie"]) + } + if _, ok := got["sess-stale"]; ok { + t.Fatalf("sess-stale should not inherit previous turn actions: %#v", got["sess-stale"]) + } + if _, ok := got["sess-missing"]; ok { + t.Fatalf("sess-missing should not have a pending action: %#v", got["sess-missing"]) + } +} diff --git a/apps/backend/internal/task/service/service_events.go b/apps/backend/internal/task/service/service_events.go index 9532fd8d4c..385c5dcc4c 100644 --- a/apps/backend/internal/task/service/service_events.go +++ b/apps/backend/internal/task/service/service_events.go @@ -145,6 +145,7 @@ func (s *Service) addTaskSessionEventFields(ctx context.Context, taskID string, if !ok || sessionInfo == nil { data["primary_session_id"] = nil data["primary_session_state"] = nil + data["primary_session_pending_action"] = nil return } data["primary_session_id"] = sessionInfo.ID @@ -156,6 +157,7 @@ func (s *Service) addTaskSessionEventFields(ctx context.Context, taskID string, } else { data["primary_session_state"] = nil } + s.addPrimarySessionPendingActionEventField(ctx, taskID, sessionInfo, data) if sessionInfo.ExecutorID != "" { data["primary_executor_id"] = sessionInfo.ExecutorID } @@ -174,6 +176,27 @@ func (s *Service) addTaskSessionEventFields(ctx context.Context, taskID string, } } +func (s *Service) addPrimarySessionPendingActionEventField(ctx context.Context, taskID string, sessionInfo *models.TaskSession, data map[string]interface{}) { + if sessionInfo.State != models.TaskSessionStateWaitingForInput { + data["primary_session_pending_action"] = nil + return + } + actions, err := s.GetPendingActionsForSessions(ctx, []string{sessionInfo.ID}) + if err != nil { + s.logger.Warn("failed to load pending action for task event", + zap.String("task_id", taskID), + zap.String("session_id", sessionInfo.ID), + zap.Error(err)) + return + } + action, ok := actions[sessionInfo.ID] + if !ok { + data["primary_session_pending_action"] = nil + return + } + data["primary_session_pending_action"] = string(action) +} + // taskRepositoriesForEvent returns the task's full repository list, ordered by // position. Prefers Task.Repositories when already loaded; falls back to a // lookup so publishers that pass a task without eagerly loaded repositories diff --git a/apps/backend/internal/task/service/service_events_test.go b/apps/backend/internal/task/service/service_events_test.go index 6a6a43c18f..e036d8fd2f 100644 --- a/apps/backend/internal/task/service/service_events_test.go +++ b/apps/backend/internal/task/service/service_events_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "testing" + "time" "github.com/kandev/kandev/internal/task/models" "github.com/kandev/kandev/internal/task/repository" @@ -14,6 +15,11 @@ type failingTaskRepoRepository struct { err error } +type failingMessageRepository struct { + repository.MessageRepository + err error +} + type primarySessionInfoRepository struct { repository.SessionRepository info map[string]*models.TaskSession @@ -29,6 +35,10 @@ func (r failingTaskRepoRepository) ListTaskRepositories(ctx context.Context, tas return nil, r.err } +func (r failingMessageRepository) GetPendingActionsBySessionIDs(ctx context.Context, sessionIDs []string) (map[string]models.TaskPendingAction, error) { + return nil, r.err +} + func (r primarySessionInfoRepository) GetPrimarySessionInfoByTaskIDs(ctx context.Context, taskIDs []string) (map[string]*models.TaskSession, error) { return r.info, nil } @@ -134,6 +144,71 @@ func TestPublishTaskUpdated_EmitsNullPrimarySessionFieldsWhenNoPrimaryExists(t * if value, ok := data["primary_session_state"]; !ok || value != nil { t.Fatalf("primary_session_state = %#v, want explicit nil", value) } + if value, ok := data["primary_session_pending_action"]; !ok || value != nil { + t.Fatalf("primary_session_pending_action = %#v, want explicit nil", value) + } +} + +func TestPublishTaskUpdated_EmitsPrimarySessionPendingAction(t *testing.T) { + svc, eventBus, repo := createTestService(t) + ctx := context.Background() + now := time.Now().UTC() + + 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) + } + task := &models.Task{ + ID: "task-1", + WorkspaceID: "ws-1", + WorkflowID: "wf-1", + WorkflowStepID: "step-1", + Title: "T", + Priority: "medium", + } + if err := repo.CreateTask(ctx, task); err != nil { + t.Fatalf("CreateTask: %v", err) + } + if err := repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: "session-1", + TaskID: task.ID, + State: models.TaskSessionStateWaitingForInput, + IsPrimary: true, + StartedAt: now, + UpdatedAt: now, + }); err != nil { + t.Fatalf("CreateTaskSession: %v", err) + } + if err := repo.CreateTurn(ctx, &models.Turn{ + ID: "turn-1", + TaskSessionID: "session-1", + TaskID: task.ID, + }); err != nil { + t.Fatalf("CreateTurn: %v", err) + } + if err := repo.CreateMessage(ctx, &models.Message{ + ID: "message-1", + TaskSessionID: "session-1", + TaskID: task.ID, + TurnID: "turn-1", + AuthorType: models.MessageAuthorAgent, + Content: "question", + Type: models.MessageTypeClarificationRequest, + Metadata: map[string]interface{}{"status": "pending"}, + CreatedAt: now, + }); err != nil { + t.Fatalf("CreateMessage: %v", err) + } + eventBus.ClearEvents() + + svc.PublishTaskUpdated(ctx, task) + + data := singlePublishedEventData(t, eventBus) + if value := data["primary_session_pending_action"]; value != "clarification" { + t.Fatalf("primary_session_pending_action = %#v, want clarification", value) + } } func TestAddTaskSessionEventFields_EmitsNullPrimarySessionStateWhenEmpty(t *testing.T) { @@ -155,6 +230,33 @@ func TestAddTaskSessionEventFields_EmitsNullPrimarySessionStateWhenEmpty(t *test } } +func TestAddTaskSessionEventFields_OmitsPendingActionOnLookupErrorForWaitingSession(t *testing.T) { + svc, _, _ := createTestService(t) + svc.sessions = primarySessionInfoRepository{ + info: map[string]*models.TaskSession{ + "task-1": { + ID: "session-1", + TaskID: "task-1", + State: models.TaskSessionStateWaitingForInput, + }, + }, + } + svc.messages = failingMessageRepository{err: errors.New("pending lookup failed")} + 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 := data["primary_session_state"]; value != string(models.TaskSessionStateWaitingForInput) { + t.Fatalf("primary_session_state = %#v, want WAITING_FOR_INPUT", value) + } + if value, ok := data["primary_session_pending_action"]; ok { + t.Fatalf("primary_session_pending_action should be omitted on lookup error, got %#v", value) + } +} + func TestPublishTaskUpdated_OmitsRepositoriesOnLookupError(t *testing.T) { svc, eventBus, repo := createTestService(t) ctx := context.Background() diff --git a/apps/backend/internal/task/service/service_sessions.go b/apps/backend/internal/task/service/service_sessions.go index 20924e52f2..7dc5bc2c40 100644 --- a/apps/backend/internal/task/service/service_sessions.go +++ b/apps/backend/internal/task/service/service_sessions.go @@ -107,6 +107,12 @@ func (s *Service) GetPrimarySessionInfoForTasks(ctx context.Context, taskIDs []s return s.sessions.GetPrimarySessionInfoByTaskIDs(ctx, taskIDs) } +// GetPendingActionsForSessions returns pending user-action projections for +// sessions whose messages may not be loaded by list views. +func (s *Service) GetPendingActionsForSessions(ctx context.Context, sessionIDs []string) (map[string]models.TaskPendingAction, error) { + return s.messages.GetPendingActionsBySessionIDs(ctx, sessionIDs) +} + // BatchGetSessionsForTasks returns all sessions for the given task IDs grouped // by task ID. Wraps the repository batch loader so callers in the handler // layer can derive primary session, session count, and per-task session info diff --git a/apps/web/components/kanban-card-content.tsx b/apps/web/components/kanban-card-content.tsx index 2abec0720b..4395c0c914 100644 --- a/apps/web/components/kanban-card-content.tsx +++ b/apps/web/components/kanban-card-content.tsx @@ -224,7 +224,10 @@ function KanbanCardActions({ const storeApi = useAppStoreApi(); const debugEnabled = isDebug(); const effectiveMenuOpen = menuOpen || Boolean(isDeleting) || Boolean(isArchiving); - const hasPendingClarificationRequest = useTaskPendingClarification(task.primarySessionId); + const hasPendingClarificationRequest = useTaskPendingClarification(task.primarySessionId, { + primarySessionState: task.primarySessionState, + primarySessionPendingAction: task.primarySessionPendingAction, + }); const showQuestionIcon = shouldUseQuestionTaskIcon(task.state, hasPendingClarificationRequest); const showRunningSpinner = shouldShowTaskRunningSpinner(task.state, task.primarySessionState); const storeWouldShowRunningSpinner = diff --git a/apps/web/components/kanban-card.tsx b/apps/web/components/kanban-card.tsx index aaff8b2d67..692d383968 100644 --- a/apps/web/components/kanban-card.tsx +++ b/apps/web/components/kanban-card.tsx @@ -22,7 +22,12 @@ import { useTaskWorkflowMove } from "@/hooks/use-task-workflow-move"; import { useTaskMultiSelectStore } from "@/hooks/use-task-multi-select"; import { repositorySlug } from "@/lib/repository-slug"; import { formatUserHomePath } from "@/lib/utils"; -import { repositoryId as toRepositoryId, type Repository, type TaskState } from "@/lib/types/http"; +import { + repositoryId as toRepositoryId, + type Repository, + type TaskPendingAction, + type TaskState, +} from "@/lib/types/http"; export interface Task { id: string; @@ -42,6 +47,7 @@ export interface Task { * finished — the workflow may leave the task in IN_PROGRESS for review. */ primarySessionState?: string | null; + primarySessionPendingAction?: TaskPendingAction | null; reviewStatus?: "pending" | "approved" | "changes_requested" | "rejected" | null; primaryExecutorId?: string | null; primaryExecutorType?: string | null; diff --git a/apps/web/components/kanban/graph2-step-node.tsx b/apps/web/components/kanban/graph2-step-node.tsx index 86d2488cf9..47515f5954 100644 --- a/apps/web/components/kanban/graph2-step-node.tsx +++ b/apps/web/components/kanban/graph2-step-node.tsx @@ -101,7 +101,10 @@ export function Graph2StepNode({ }: Graph2StepNodeProps) { const router = useRouter(); const [isHovered, setIsHovered] = useState(false); - const hasPendingClarificationRequest = useTaskPendingClarification(task.primarySessionId); + const hasPendingClarificationRequest = useTaskPendingClarification(task.primarySessionId, { + primarySessionState: task.primarySessionState, + primarySessionPendingAction: task.primarySessionPendingAction, + }); if (phase === "past") return ; if (phase === "future") return ; diff --git a/apps/web/components/kanban/swimlane-graph-content.tsx b/apps/web/components/kanban/swimlane-graph-content.tsx index 3c74cf49f7..bfeb381d6b 100644 --- a/apps/web/components/kanban/swimlane-graph-content.tsx +++ b/apps/web/components/kanban/swimlane-graph-content.tsx @@ -74,7 +74,10 @@ function DraggableTaskChip({ id: task.id, }); const isPreviewed = useAppStore((state) => state.kanbanPreviewedTaskId === task.id); - const hasPendingClarificationRequest = useTaskPendingClarification(task.primarySessionId); + const hasPendingClarificationRequest = useTaskPendingClarification(task.primarySessionId, { + primarySessionState: task.primarySessionState, + primarySessionPendingAction: task.primarySessionPendingAction, + }); const statusIcon = getTaskStateIcon(task.state, "h-3 w-3", hasPendingClarificationRequest); return ( @@ -101,7 +104,10 @@ function DraggableTaskChip({ } function TaskChipPreview({ task }: { task: Task }) { - const hasPendingClarificationRequest = useTaskPendingClarification(task.primarySessionId); + const hasPendingClarificationRequest = useTaskPendingClarification(task.primarySessionId, { + primarySessionState: task.primarySessionState, + primarySessionPendingAction: task.primarySessionPendingAction, + }); const statusIcon = getTaskStateIcon(task.state, "h-3 w-3", hasPendingClarificationRequest); return (
({ - id: task.id, - workflowStepId: task.workflow_step_id, - parentTaskId: task.parent_id ?? undefined, - title: task.title, - description: task.description ?? undefined, - position: task.position ?? 0, - state: task.state, - repositoryId: task.repositories?.[0]?.repository_id ?? undefined, - // Carry the full TaskRepository array so the mobile repo picker - // (useTaskRepoCount + MobileReposSection) keeps working after a workspace - // switch. Without this, the picker silently disappears for multi-repo - // tasks because length defaults to 0. - repositories: mapTaskRepositories(task.repositories), - 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, - updatedAt: task.updated_at, - })), + tasks: snapshot.tasks.map(toKanbanTask), }; } @@ -95,7 +74,7 @@ type SheetItemCtx = { sessionsByTaskId: Parameters[1]; gitStatusByEnvId: Parameters[2]; envIdBySessionId: Parameters[3]; - messagesBySession: Parameters[0]; + messagesBySession: Record; dismissedAgentErrors: Record; acknowledgedAgentErrors: Record; }; @@ -110,6 +89,9 @@ function toSheetItem( ctx.gitStatusByEnvId, ctx.envIdBySessionId, ); + const resolvedSessionState = + sessionInfo.sessionState ?? (task.primarySessionState as TaskSessionState | undefined); + const pending = pendingFlagsForTask(task, resolvedSessionState, ctx.messagesBySession); return { id: task.id, title: task.title, @@ -117,8 +99,7 @@ function toSheetItem( // way the desktop sidebar does (applyView/TaskSwitcher read parentTaskId). parentTaskId: task.parentTaskId ?? undefined, state: task.state as TaskState | undefined, - sessionState: - sessionInfo.sessionState ?? (task.primarySessionState as TaskSessionState | undefined), + sessionState: resolvedSessionState, description: task.description, workflowId: task._workflowId, workflowName: ctx.workflowNameById.get(task._workflowId), @@ -133,18 +114,34 @@ function toSheetItem( remoteExecutorType: task.primaryExecutorType ?? undefined, remoteExecutorName: task.primaryExecutorName ?? undefined, primarySessionId: task.primarySessionId ?? null, - hasPendingClarification: hasPendingClarificationForSession( - ctx.messagesBySession, - task.primarySessionId, - ), - hasPendingPermission: hasPendingPermissionForSession( - ctx.messagesBySession, - task.primarySessionId, - ), + hasPendingClarification: pending.clarification, + hasPendingPermission: pending.permission, agentErrorMessage: agentErrorMessageForTask(task, ctx.sessionsById, ctx.sessionsByTaskId, ctx), }; } +function pendingFlagsForTask( + task: Pick, + primarySessionState: TaskSessionState | undefined, + messagesBySession: Record, +): { clarification: boolean; permission: boolean } { + if (!task.primarySessionId) return { clarification: false, permission: false }; + const messages = messagesBySession[task.primarySessionId]; + if (messages !== undefined) { + return { + clarification: hasPendingClarification(messages), + permission: hasPendingPermissionRequest(messages), + }; + } + if (primarySessionState !== "WAITING_FOR_INPUT") { + return { clarification: false, permission: false }; + } + return { + clarification: task.primarySessionPendingAction === "clarification", + permission: task.primarySessionPendingAction === "permission", + }; +} + export function useSheetData(workspaceId: string | null) { const activeTaskId = useAppStore((state) => state.tasks.activeTaskId); const activeSessionId = useAppStore((state) => state.tasks.activeSessionId); @@ -319,14 +316,51 @@ function mergeSessionFields( taskSessionId: string | null, ) { return { - primarySessionId: - taskSessionId ?? task.primary_session_id ?? existing?.primarySessionId ?? undefined, - primarySessionState: task.primary_session_state ?? existing?.primarySessionState ?? undefined, - sessionCount: task.session_count ?? existing?.sessionCount ?? (taskSessionId ? 1 : undefined), - reviewStatus: task.review_status ?? existing?.reviewStatus ?? undefined, + primarySessionId: resolvePrimarySessionId(task, existing, taskSessionId), + primarySessionState: resolvePrimarySessionState(task, existing), + primarySessionPendingAction: resolvePrimarySessionPendingAction(task, existing), + sessionCount: resolveSessionCount(task, existing, taskSessionId), + reviewStatus: resolveReviewStatus(task, existing), }; } +function resolvePrimarySessionId( + task: Task, + existing: KanbanState["tasks"][number] | undefined, + taskSessionId: string | null, +) { + return taskSessionId ?? task.primary_session_id ?? existing?.primarySessionId ?? undefined; +} + +function resolvePrimarySessionState( + task: Task, + existing: KanbanState["tasks"][number] | undefined, +) { + return task.primary_session_state ?? existing?.primarySessionState ?? undefined; +} + +function resolvePrimarySessionPendingAction( + task: Task, + existing: KanbanState["tasks"][number] | undefined, +) { + if ("primary_session_pending_action" in task) { + return task.primary_session_pending_action ?? undefined; + } + return existing?.primarySessionPendingAction ?? undefined; +} + +function resolveSessionCount( + task: Task, + existing: KanbanState["tasks"][number] | undefined, + taskSessionId: string | null, +) { + return task.session_count ?? existing?.sessionCount ?? (taskSessionId ? 1 : undefined); +} + +function resolveReviewStatus(task: Task, existing: KanbanState["tasks"][number] | undefined) { + return task.review_status ?? existing?.reviewStatus ?? undefined; +} + /** * Build the kanban-store representation of a task for an upsert. Session- * derived fields (primarySessionId, sessionCount, etc.) fall through new diff --git a/apps/web/components/task/task-session-sidebar-aggregate.test.ts b/apps/web/components/task/task-session-sidebar-aggregate.test.ts index f16285b058..0736a63428 100644 --- a/apps/web/components/task/task-session-sidebar-aggregate.test.ts +++ b/apps/web/components/task/task-session-sidebar-aggregate.test.ts @@ -184,4 +184,41 @@ describe("buildPendingFlags / readPendingFlags", () => { expect(readPendingFlags({}, null)).toEqual({ clarification: false, permission: false }); expect(readPendingFlags({}, "missing")).toEqual({ clarification: false, permission: false }); }); + + it("falls back to the snapshot pending action when messages are not loaded", () => { + expect( + readPendingFlags({}, "sess-1", { + primarySessionState: "WAITING_FOR_INPUT", + primarySessionPendingAction: "clarification", + }), + ).toEqual({ clarification: true, permission: false }); + }); + + it("falls back to pending permission from the snapshot", () => { + expect( + readPendingFlags({}, "sess-1", { + primarySessionState: "WAITING_FOR_INPUT", + primarySessionPendingAction: "permission", + }), + ).toEqual({ clarification: false, permission: true }); + }); + + it("prefers loaded messages over a stale snapshot pending action", () => { + const flags = buildPendingFlags({ "sess-1": [] }, ["sess-1"]); + expect( + readPendingFlags(flags, "sess-1", { + primarySessionState: "WAITING_FOR_INPUT", + primarySessionPendingAction: "clarification", + }), + ).toEqual({ clarification: false, permission: false }); + }); + + it("does not use the snapshot pending action after the session moves on", () => { + expect( + readPendingFlags({}, "sess-1", { + primarySessionState: "RUNNING", + primarySessionPendingAction: "clarification", + }), + ).toEqual({ clarification: false, permission: false }); + }); }); diff --git a/apps/web/components/task/task-session-sidebar-aggregate.ts b/apps/web/components/task/task-session-sidebar-aggregate.ts index 4c426cecb2..6751af38ad 100644 --- a/apps/web/components/task/task-session-sidebar-aggregate.ts +++ b/apps/web/components/task/task-session-sidebar-aggregate.ts @@ -1,5 +1,5 @@ import type { KanbanState } from "@/lib/state/slices"; -import type { Message } from "@/lib/types/http"; +import type { Message, TaskPendingAction } from "@/lib/types/http"; import { hasPendingClarification, hasPendingPermissionRequest, @@ -18,20 +18,46 @@ export function buildPendingFlags( const flags: Record = {}; for (const id of sessionIds) { const msgs = bySession[id]; + if (msgs === undefined) continue; flags[pendingClarKey(id)] = hasPendingClarification(msgs); flags[pendingPermKey(id)] = hasPendingPermissionRequest(msgs); } return flags; } +export type PendingActionFallback = { + primarySessionState?: string | null; + primarySessionPendingAction?: TaskPendingAction | null; +}; + +function fallbackPendingFlags(fallback?: PendingActionFallback): { + clarification: boolean; + permission: boolean; +} { + if (fallback?.primarySessionState !== "WAITING_FOR_INPUT") { + return { clarification: false, permission: false }; + } + return { + clarification: fallback.primarySessionPendingAction === "clarification", + permission: fallback.primarySessionPendingAction === "permission", + }; +} + export function readPendingFlags( pendingFlags: Record, sessionId?: string | null, + fallback?: PendingActionFallback, ): { clarification: boolean; permission: boolean } { if (!sessionId) return { clarification: false, permission: false }; + const clarKey = pendingClarKey(sessionId); + const permKey = pendingPermKey(sessionId); + const hasMessageFlags = + Object.prototype.hasOwnProperty.call(pendingFlags, clarKey) || + Object.prototype.hasOwnProperty.call(pendingFlags, permKey); + if (!hasMessageFlags) return fallbackPendingFlags(fallback); return { - clarification: pendingFlags[pendingClarKey(sessionId)] ?? false, - permission: pendingFlags[pendingPermKey(sessionId)] ?? false, + clarification: pendingFlags[clarKey] ?? false, + permission: pendingFlags[permKey] ?? false, }; } diff --git a/apps/web/components/task/task-session-sidebar.tsx b/apps/web/components/task/task-session-sidebar.tsx index d9fe928fd0..7ec4b149b7 100644 --- a/apps/web/components/task/task-session-sidebar.tsx +++ b/apps/web/components/task/task-session-sidebar.tsx @@ -113,7 +113,10 @@ function toSidebarItem( const repoSlug = task.repositoryId ? ctx.repositorySlugById.get(task.repositoryId) : undefined; // Sidebar shows just one slot; pick the primary PR (first by created_at). const pr = ctx.taskPRsByTaskId[task.id]?.[0]; - const pending = readPendingFlags(ctx.pendingFlags, task.primarySessionId); + const pending = readPendingFlags(ctx.pendingFlags, task.primarySessionId, { + primarySessionState: resolvedSessionState, + primarySessionPendingAction: task.primarySessionPendingAction, + }); const diffStats = resolveDiffStats( sessionInfo.diffStats, diff --git a/apps/web/e2e/tests/task/sidebar-pending-question.spec.ts b/apps/web/e2e/tests/task/sidebar-pending-question.spec.ts new file mode 100644 index 0000000000..b69af15cea --- /dev/null +++ b/apps/web/e2e/tests/task/sidebar-pending-question.spec.ts @@ -0,0 +1,88 @@ +/** + * Regression test for kdlbs/kandev#1657: the sidebar "waiting for input" + * question icon must show for a task blocked on an agent clarification even + * when the task has never been opened, purely from the task snapshot. + */ +import { expect, test } from "../../fixtures/test-base"; +import type { ApiClient } from "../../helpers/api-client"; +import { SessionPage } from "../../pages/session-page"; + +async function waitForSessionWaitingForInput( + apiClient: ApiClient, + taskId: string, + message: string, +): Promise { + await expect + .poll( + async () => { + const { sessions } = await apiClient.listTaskSessions(taskId); + return sessions[0]?.state ?? ""; + }, + { message, timeout: 60_000 }, + ) + .toBe("WAITING_FOR_INPUT"); +} + +test.describe("Sidebar pending-question indicator without opening the task", () => { + test("blocked task shows the question icon on a fresh page load; idle task does not", async ({ + apiClient, + seedData, + testPage, + }) => { + const blockedTask = await apiClient.createTaskWithAgent( + seedData.workspaceId, + "Blocked On Question", + seedData.agentProfileId, + { + description: "/e2e:clarification", + repository_ids: [seedData.repositoryId], + workflow_id: seedData.workflowId, + workflow_step_id: seedData.startStepId, + }, + ); + + const idleTask = await apiClient.createTaskWithAgent( + seedData.workspaceId, + "Finished Quietly", + seedData.agentProfileId, + { + description: "/e2e:simple-message", + repository_ids: [seedData.repositoryId], + workflow_id: seedData.workflowId, + workflow_step_id: seedData.startStepId, + }, + ); + + await waitForSessionWaitingForInput( + apiClient, + blockedTask.id, + "blocked task session should park on the clarification", + ); + await waitForSessionWaitingForInput( + apiClient, + idleTask.id, + "idle task session should finish its turn", + ); + + const navTask = await apiClient.createTask(seedData.workspaceId, "Nav Task", { + workflow_id: seedData.workflowId, + workflow_step_id: seedData.startStepId, + }); + await testPage.goto(`/t/${navTask.id}`); + + const session = new SessionPage(testPage); + await session.waitForLoad(); + await expect(session.sidebar).toBeVisible({ timeout: 10_000 }); + + const blockedRow = session.sidebarTaskItem("Blocked On Question"); + await expect(blockedRow).toBeVisible({ timeout: 10_000 }); + await expect(blockedRow.getByTestId("task-state-waiting-for-input")).toBeVisible({ + timeout: 10_000, + }); + + const idleRow = session.sidebarTaskItem("Finished Quietly"); + await expect(idleRow).toBeVisible({ timeout: 10_000 }); + await expect(idleRow.getByTestId("task-state-waiting-for-input")).toHaveCount(0); + await expect(idleRow.getByTestId("task-state-pending-permission")).toHaveCount(0); + }); +}); diff --git a/apps/web/hooks/use-task-pending-clarification.test.tsx b/apps/web/hooks/use-task-pending-clarification.test.tsx index 4e1f256ece..cbe0643bfa 100644 --- a/apps/web/hooks/use-task-pending-clarification.test.tsx +++ b/apps/web/hooks/use-task-pending-clarification.test.tsx @@ -47,6 +47,36 @@ describe("useTaskPendingClarification", () => { expect(result.current).toBe(false); }); + it("falls back to the snapshot pending clarification when messages are not loaded", () => { + const { result } = renderHook( + () => + useTaskPendingClarification("session-1", { + primarySessionState: "WAITING_FOR_INPUT", + primarySessionPendingAction: "clarification", + }), + { + wrapper: wrapper(), + }, + ); + + expect(result.current).toBe(true); + }); + + it("prefers loaded messages over a stale snapshot pending clarification", () => { + const { result } = renderHook( + () => + useTaskPendingClarification("session-1", { + primarySessionState: "WAITING_FOR_INPUT", + primarySessionPendingAction: "clarification", + }), + { + wrapper: wrapper({ "session-1": [] }), + }, + ); + + expect(result.current).toBe(false); + }); + it("returns true when the session has a pending clarification", () => { const { result } = renderHook(() => useTaskPendingClarification("session-1"), { wrapper: wrapper({ diff --git a/apps/web/hooks/use-task-pending-clarification.ts b/apps/web/hooks/use-task-pending-clarification.ts index e9f148312f..69e3a386d8 100644 --- a/apps/web/hooks/use-task-pending-clarification.ts +++ b/apps/web/hooks/use-task-pending-clarification.ts @@ -1,8 +1,23 @@ import { useAppStore } from "@/components/state-provider"; -import { hasPendingClarificationForSession } from "@/lib/utils/pending-clarification"; +import type { TaskPendingAction } from "@/lib/types/http"; +import { hasPendingClarification } from "@/lib/utils/pending-clarification"; -export function useTaskPendingClarification(primarySessionId: string | null | undefined): boolean { - return useAppStore((state) => - hasPendingClarificationForSession(state.messages.bySession, primarySessionId), - ); +type PendingActionFallback = { + primarySessionState?: string | null; + primarySessionPendingAction?: TaskPendingAction | null; +}; + +export function useTaskPendingClarification( + primarySessionId: string | null | undefined, + fallback?: PendingActionFallback, +): boolean { + return useAppStore((state) => { + if (!primarySessionId) return false; + const messages = state.messages.bySession[primarySessionId]; + if (messages !== undefined) return hasPendingClarification(messages); + return ( + fallback?.primarySessionState === "WAITING_FOR_INPUT" && + fallback.primarySessionPendingAction === "clarification" + ); + }); } diff --git a/apps/web/lib/kanban/map-task.test.ts b/apps/web/lib/kanban/map-task.test.ts index c9f33775a8..7d6541ca18 100644 --- a/apps/web/lib/kanban/map-task.test.ts +++ b/apps/web/lib/kanban/map-task.test.ts @@ -96,6 +96,28 @@ describe("toKanbanTask — HTTP DTO / WS payload parity", () => { expect(toKanbanTask(wsPayload()).repositoryId).toBe("repo-a"); }); + it("maps primary session pending action from HTTP and WS shapes", () => { + const pendingAction = { + primary_session_pending_action: "clarification", + } as Record as Partial; + const http = toKanbanTask(httpDTO(pendingAction)); + const ws = toKanbanTask(wsPayload(pendingAction)); + + expect(http).toEqual(ws); + expect((http as Record).primarySessionPendingAction).toBe("clarification"); + }); + + it("drops unrecognized primary session pending action values", () => { + const invalidPendingAction = { + primary_session_pending_action: "unknown", + } as Record as Partial; + + expect(toKanbanTask(httpDTO(invalidPendingAction)).primarySessionPendingAction).toBeUndefined(); + expect( + toKanbanTask(wsPayload(invalidPendingAction)).primarySessionPendingAction, + ).toBeUndefined(); + }); + it("missing repository on either shape: repositoryId is undefined", () => { const http = httpDTO({ repositories: undefined }); const ws = wsPayload({ repository_id: undefined, repositories: undefined }); diff --git a/apps/web/lib/kanban/map-task.ts b/apps/web/lib/kanban/map-task.ts index 7d1ecaf577..27ec5a22e3 100644 --- a/apps/web/lib/kanban/map-task.ts +++ b/apps/web/lib/kanban/map-task.ts @@ -4,7 +4,7 @@ import { issueFieldsFromMetadata, } from "@/lib/metadata-utils"; import type { KanbanState } from "@/lib/state/slices/kanban/types"; -import type { TaskState, TaskSessionState } from "@/lib/types/http"; +import type { TaskPendingAction, TaskState, TaskSessionState } from "@/lib/types/http"; type KanbanTask = KanbanState["tasks"][number]; @@ -35,6 +35,7 @@ export type TaskLike = { repository_id?: string; primary_session_id?: string | null; primary_session_state?: TaskSessionState | string | null; + primary_session_pending_action?: TaskPendingAction | null; session_count?: number | null; review_status?: "pending" | "approved" | "changes_requested" | "rejected" | null; primary_executor_id?: string | null; @@ -55,6 +56,13 @@ function pickId(source: TaskLike): string { return (source.id ?? source.task_id ?? "") as string; } +export function pickPendingAction(action: unknown): TaskPendingAction | undefined { + if (action === "clarification" || action === "permission") { + return action; + } + return undefined; +} + type KanbanTaskRepository = NonNullable[number]; function pickRepositories(source: TaskLike): KanbanTaskRepository[] | undefined { @@ -86,6 +94,7 @@ export function toKanbanTask(source: TaskLike): KanbanTask { repositories: pickRepositories(source), primarySessionId: source.primary_session_id ?? undefined, primarySessionState: source.primary_session_state ?? undefined, + primarySessionPendingAction: pickPendingAction(source.primary_session_pending_action), sessionCount: source.session_count ?? undefined, reviewStatus: source.review_status ?? undefined, primaryExecutorId: source.primary_executor_id ?? undefined, diff --git a/apps/web/lib/ssr/mapper.test.ts b/apps/web/lib/ssr/mapper.test.ts new file mode 100644 index 0000000000..2e0f032043 --- /dev/null +++ b/apps/web/lib/ssr/mapper.test.ts @@ -0,0 +1,61 @@ +import { describe, expect, it } from "vitest"; + +import { snapshotToState } from "./mapper"; +import { taskId, workflowId, workspaceId } from "@/lib/types/ids"; +import type { WorkflowSnapshot } from "@/lib/types/http"; + +const now = "2026-07-10T12:00:00Z"; +const workflowID = workflowId("workflow-1"); +const workspaceID = workspaceId("workspace-1"); + +function snapshotWithPendingAction(action: unknown): WorkflowSnapshot { + return { + workflow: { + id: workflowID, + workspace_id: workspaceID, + name: "Workflow", + created_at: now, + updated_at: now, + }, + steps: [ + { + id: "step-1", + workflow_id: workflowID, + name: "Todo", + position: 0, + color: "bg-neutral-400", + allow_manual_move: true, + }, + ], + tasks: [ + { + id: taskId("task-1"), + workspace_id: workspaceID, + workflow_id: workflowID, + workflow_step_id: "step-1", + position: 0, + title: "Task", + description: "", + state: "TODO", + priority: 0, + primary_session_pending_action: action, + created_at: now, + updated_at: now, + } as WorkflowSnapshot["tasks"][number], + ], + }; +} + +describe("snapshotToState", () => { + it("keeps known primary session pending action values", () => { + const state = snapshotToState(snapshotWithPendingAction("permission")); + + expect(state.kanban?.tasks[0]?.primarySessionPendingAction).toBe("permission"); + }); + + it("drops unrecognized primary session pending action values", () => { + const state = snapshotToState(snapshotWithPendingAction("unknown")); + + expect(state.kanban?.tasks[0]?.primarySessionPendingAction).toBeUndefined(); + }); +}); diff --git a/apps/web/lib/ssr/mapper.ts b/apps/web/lib/ssr/mapper.ts index 081f036208..7070019bf7 100644 --- a/apps/web/lib/ssr/mapper.ts +++ b/apps/web/lib/ssr/mapper.ts @@ -1,6 +1,7 @@ import type { AppState, KanbanState } from "@/lib/state/store"; import { primaryTaskRepository } from "@/lib/types/http"; import type { WorkflowSnapshot, Message, Task } from "@/lib/types/http"; +import { pickPendingAction } from "@/lib/kanban/map-task"; import { isPRReviewFromMetadata, isIssueWatchFromMetadata, @@ -45,6 +46,7 @@ export function snapshotToState(snapshot: WorkflowSnapshot): Partial { })), primarySessionId: task.primary_session_id ?? undefined, primarySessionState: task.primary_session_state ?? undefined, + primarySessionPendingAction: pickPendingAction(task.primary_session_pending_action), sessionCount: task.session_count ?? undefined, reviewStatus: task.review_status ?? undefined, parentTaskId: task.parent_id ?? undefined, diff --git a/apps/web/lib/state/slices/kanban/types.ts b/apps/web/lib/state/slices/kanban/types.ts index 54189e99a2..35611be251 100644 --- a/apps/web/lib/state/slices/kanban/types.ts +++ b/apps/web/lib/state/slices/kanban/types.ts @@ -1,4 +1,4 @@ -import type { TaskState as TaskStatus } from "@/lib/types/http"; +import type { TaskPendingAction, TaskState as TaskStatus } from "@/lib/types/http"; export type KanbanStepEvents = { on_enter?: Array<{ type: string; config?: Record }>; @@ -57,6 +57,7 @@ export type KanbanState = { }>; primarySessionId?: string | null; primarySessionState?: string | null; + primarySessionPendingAction?: TaskPendingAction | null; sessionCount?: number | null; reviewStatus?: "pending" | "approved" | "changes_requested" | "rejected" | null; primaryExecutorId?: string | null; diff --git a/apps/web/lib/types/backend.ts b/apps/web/lib/types/backend.ts index 2df097e7df..c6ba00cb79 100644 --- a/apps/web/lib/types/backend.ts +++ b/apps/web/lib/types/backend.ts @@ -14,6 +14,8 @@ import type { SidebarViewDraftApi, SidebarTaskPrefsApi, TaskCreateLastUsedApi, + TaskPendingAction, + TaskSessionState, StepEvents, TaskState, ToolStatus, @@ -80,6 +82,8 @@ export type TaskEventPayload = { position?: number; }>; primary_session_id?: string | null; + primary_session_state?: TaskSessionState | null; + primary_session_pending_action?: TaskPendingAction | null; session_count?: number | null; review_status?: "pending" | "approved" | "changes_requested" | "rejected" | null; archived_at?: string | null; diff --git a/apps/web/lib/types/http.ts b/apps/web/lib/types/http.ts index ce379a2141..2e8b8147e8 100644 --- a/apps/web/lib/types/http.ts +++ b/apps/web/lib/types/http.ts @@ -147,6 +147,8 @@ export type TaskSessionState = | "FAILED" | "CANCELLED"; +export type TaskPendingAction = "clarification" | "permission"; + export type Workflow = { id: WorkflowId; workspace_id: WorkspaceId; @@ -278,6 +280,7 @@ export type Task = { repositories?: TaskRepository[]; primary_session_id?: SessionId | null; primary_session_state?: TaskSessionState | null; + primary_session_pending_action?: TaskPendingAction | null; session_count?: number | null; review_status?: "pending" | "approved" | "changes_requested" | "rejected" | null; primary_executor_id?: string | null; diff --git a/apps/web/lib/ws/handlers/agent-session-kanban-sync.ts b/apps/web/lib/ws/handlers/agent-session-kanban-sync.ts index d4522c8802..3a2d0e08b5 100644 --- a/apps/web/lib/ws/handlers/agent-session-kanban-sync.ts +++ b/apps/web/lib/ws/handlers/agent-session-kanban-sync.ts @@ -28,9 +28,20 @@ function patchTaskPrimarySessionState( let changed = false; const nextTasks = tasks.map((task) => { if (task.id !== taskId || task.primarySessionId !== sessionId) return task; - if (task.primarySessionState === newState) return task; + const nextPendingAction = + newState === "WAITING_FOR_INPUT" ? task.primarySessionPendingAction : undefined; + if ( + task.primarySessionState === newState && + task.primarySessionPendingAction === nextPendingAction + ) { + return task; + } changed = true; - return { ...task, primarySessionState: newState }; + return { + ...task, + primarySessionState: newState, + primarySessionPendingAction: nextPendingAction, + }; }); return changed ? nextTasks : tasks; } diff --git a/apps/web/lib/ws/handlers/kanban.ts b/apps/web/lib/ws/handlers/kanban.ts index 75d49d8b5d..4f72b193d6 100644 --- a/apps/web/lib/ws/handlers/kanban.ts +++ b/apps/web/lib/ws/handlers/kanban.ts @@ -58,6 +58,7 @@ export function registerKanbanHandlers(store: StoreApi): WsHandlers { ...repoFields, primarySessionId: existing?.primarySessionId, primarySessionState: existing?.primarySessionState, + primarySessionPendingAction: existing?.primarySessionPendingAction, }; }); @@ -86,6 +87,10 @@ export function registerKanbanHandlers(store: StoreApi): WsHandlers { t.primarySessionState === undefined ? fallback?.primarySessionState : t.primarySessionState, + primarySessionPendingAction: + t.primarySessionPendingAction === undefined + ? fallback?.primarySessionPendingAction + : t.primarySessionPendingAction, }; }); return { diff --git a/apps/web/lib/ws/handlers/tasks.ts b/apps/web/lib/ws/handlers/tasks.ts index e7b8e244bf..be674cf5b5 100644 --- a/apps/web/lib/ws/handlers/tasks.ts +++ b/apps/web/lib/ws/handlers/tasks.ts @@ -42,6 +42,12 @@ function mergeTaskUpdate( ) { merged.primarySessionState = existing.primarySessionState; } + if ( + !hasPayloadField(payload, "primary_session_pending_action") && + nextTask.primarySessionPendingAction === undefined + ) { + merged.primarySessionPendingAction = existing.primarySessionPendingAction; + } return merged; }