fix: show pending task input before messages load (#1663)

This commit is contained in:
Carlos Florêncio
2026-07-10 23:29:40 +01:00
committed by GitHub
parent 49e2e2e27f
commit 05b550e122
36 changed files with 1024 additions and 138 deletions
+48 -2
View File
@@ -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,
@@ -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,
}
}
@@ -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")
}
@@ -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)
+54 -51
View File
@@ -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.
@@ -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
}
@@ -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))
@@ -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)
}
@@ -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"`
@@ -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)
@@ -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
@@ -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 != "<missing>" {
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, "<missing>", 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, "<missing>", 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"])
}
}
@@ -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
@@ -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()
@@ -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
+4 -1
View File
@@ -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 =
+7 -1
View File
@@ -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;
@@ -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 <PastNode step={step} />;
if (phase === "future") return <FutureNode step={step} />;
@@ -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 (
<div
@@ -11,9 +11,10 @@ import { useTaskActions, useArchiveAndSwitchTask } from "@/hooks/use-task-action
import { useTaskRemoval } from "@/hooks/use-task-removal";
import { getSessionInfoForTask } from "@/lib/utils/session-info";
import {
hasPendingClarificationForSession,
hasPendingPermissionForSession,
hasPendingClarification,
hasPendingPermissionRequest,
} from "@/lib/utils/pending-clarification";
import { toKanbanTask } from "@/lib/kanban/map-task";
import {
repositoryId as toRepositoryId,
type TaskState,
@@ -22,6 +23,7 @@ import {
type Repository,
type Task,
type WorkflowSnapshot,
type Message,
} from "@/lib/types/http";
import type { KanbanState } from "@/lib/state/slices";
import { findTaskInSnapshots } from "@/lib/kanban/find-task";
@@ -52,30 +54,7 @@ function mapSnapshotToKanban(snapshot: WorkflowSnapshot, newWorkflowId: string)
show_in_command_panel: step.show_in_command_panel,
agent_profile_id: step.agent_profile_id,
})),
tasks: snapshot.tasks.map((task) => ({
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<typeof getSessionInfoForTask>[1];
gitStatusByEnvId: Parameters<typeof getSessionInfoForTask>[2];
envIdBySessionId: Parameters<typeof getSessionInfoForTask>[3];
messagesBySession: Parameters<typeof hasPendingClarificationForSession>[0];
messagesBySession: Record<string, Message[] | undefined>;
dismissedAgentErrors: Record<string, string>;
acknowledgedAgentErrors: Record<string, string>;
};
@@ -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<KanbanState["tasks"][number], "primarySessionId" | "primarySessionPendingAction">,
primarySessionState: TaskSessionState | undefined,
messagesBySession: Record<string, Message[] | undefined>,
): { 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
@@ -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 });
});
});
@@ -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<string, boolean> = {};
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<string, boolean>,
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,
};
}
@@ -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,
@@ -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<void> {
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);
});
});
@@ -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({
@@ -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"
);
});
}
+22
View File
@@ -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<string, unknown> as Partial<TaskLike>;
const http = toKanbanTask(httpDTO(pendingAction));
const ws = toKanbanTask(wsPayload(pendingAction));
expect(http).toEqual(ws);
expect((http as Record<string, unknown>).primarySessionPendingAction).toBe("clarification");
});
it("drops unrecognized primary session pending action values", () => {
const invalidPendingAction = {
primary_session_pending_action: "unknown",
} as Record<string, unknown> as Partial<TaskLike>;
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 });
+10 -1
View File
@@ -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<KanbanTask["repositories"]>[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,
+61
View File
@@ -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();
});
});
+2
View File
@@ -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<AppState> {
})),
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,
+2 -1
View File
@@ -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<string, unknown> }>;
@@ -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;
+4
View File
@@ -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;
+3
View File
@@ -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;
@@ -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;
}
+5
View File
@@ -58,6 +58,7 @@ export function registerKanbanHandlers(store: StoreApi<AppState>): WsHandlers {
...repoFields,
primarySessionId: existing?.primarySessionId,
primarySessionState: existing?.primarySessionState,
primarySessionPendingAction: existing?.primarySessionPendingAction,
};
});
@@ -86,6 +87,10 @@ export function registerKanbanHandlers(store: StoreApi<AppState>): WsHandlers {
t.primarySessionState === undefined
? fallback?.primarySessionState
: t.primarySessionState,
primarySessionPendingAction:
t.primarySessionPendingAction === undefined
? fallback?.primarySessionPendingAction
: t.primarySessionPendingAction,
};
});
return {
+6
View File
@@ -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;
}