From 084ecc8b54c257eec0da78ce6e7d94496a08d2fa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Cl=C3=A9ment=20Fiere?= Date: Fri, 7 Aug 2026 16:34:39 +0200 Subject: [PATCH] feat: show queued prompt count badge in the task sidebar (#2354) --- .../backend/internal/backendapp/boot_state.go | 18 ++ apps/backend/internal/backendapp/gateway.go | 4 + .../internal/backendapp/orchestrator.go | 5 + .../backendapp/status_summary_boot_test.go | 71 ++++++ .../orchestrator/event_handlers_agent.go | 7 + .../orchestrator/handlers/queue_handlers.go | 63 ++++-- .../handlers/queue_handlers_publish_test.go | 111 +++++++++ .../handlers/queue_handlers_send_now_test.go | 4 +- .../handlers/queue_handlers_test.go | 8 +- .../messagequeue/count_pending_test.go | 141 ++++++++++++ .../orchestrator/messagequeue/repository.go | 7 + .../messagequeue/repository_memory.go | 25 +++ .../messagequeue/repository_sqlite.go | 40 ++++ .../orchestrator/messagequeue/service.go | 24 ++ apps/backend/internal/orchestrator/service.go | 18 ++ .../orchestrator/session_task_id_test.go | 74 ++++++ .../task/handlers/task_http_handlers.go | 24 ++ .../task/handlers/task_http_queued_test.go | 153 +++++++++++++ apps/backend/internal/task/service/service.go | 1 + .../task/service/service_queued_prompt.go | 32 +++ .../service/service_status_summary_rebuild.go | 24 +- .../service_status_summary_rebuild_test.go | 16 ++ .../internal/task/statussummary/model.go | 10 + .../internal/task/statussummary/model_test.go | 34 +++ .../internal/task/statussummary/projector.go | 89 ++++++-- .../statussummary/projector_queued_test.go | 212 ++++++++++++++++++ .../internal/task/statussummary/rebuild.go | 6 +- .../task/statussummary/rebuild_queued_test.go | 27 +++ .../session-task-switcher-sheet-hooks.test.ts | 18 ++ .../session-task-switcher-sheet-hooks.ts | 3 + .../components/task/task-item-stats-row.tsx | 53 +++-- apps/web/components/task/task-item.test.tsx | 35 +++ apps/web/components/task/task-item.tsx | 13 +- .../task/task-session-sidebar-item.test.ts | 18 ++ .../task/task-session-sidebar-item.ts | 1 + .../web/components/task/task-switcher-row.tsx | 1 + .../components/task/task-switcher-types.ts | 2 + apps/web/e2e/helpers/api-client.ts | 5 + .../task/mobile-sidebar-queued-count.spec.ts | 57 +++++ .../tests/task/sidebar-queued-count.spec.ts | 105 +++++++++ apps/web/lib/types/task-status-summary.ts | 2 + apps/web/src/locales/en/sidebar.json | 5 + apps/web/src/locales/pseudo/sidebar.json | 5 + .../plans/sidebar-queued-prompt-count/plan.md | 182 +++++++++++++++ .../task-01-pending-prompt-count-queries.md | 67 ++++++ .../task-02-status-summary-queued-count.md | 94 ++++++++ .../task-03-attach-count-to-task-payloads.md | 95 ++++++++ .../task-04-sidebar-badge-ui.md | 86 +++++++ .../task-05-e2e-and-verification.md | 81 +++++++ docs/specs/INDEX.md | 1 + docs/specs/ui/sidebar-queued-prompt-count.md | 149 ++++++++++++ 51 files changed, 2262 insertions(+), 64 deletions(-) create mode 100644 apps/backend/internal/orchestrator/handlers/queue_handlers_publish_test.go create mode 100644 apps/backend/internal/orchestrator/messagequeue/count_pending_test.go create mode 100644 apps/backend/internal/orchestrator/session_task_id_test.go create mode 100644 apps/backend/internal/task/handlers/task_http_queued_test.go create mode 100644 apps/backend/internal/task/service/service_queued_prompt.go create mode 100644 apps/backend/internal/task/statussummary/projector_queued_test.go create mode 100644 apps/backend/internal/task/statussummary/rebuild_queued_test.go create mode 100644 apps/web/e2e/tests/task/mobile-sidebar-queued-count.spec.ts create mode 100644 apps/web/e2e/tests/task/sidebar-queued-count.spec.ts create mode 100644 docs/plans/sidebar-queued-prompt-count/plan.md create mode 100644 docs/plans/sidebar-queued-prompt-count/task-01-pending-prompt-count-queries.md create mode 100644 docs/plans/sidebar-queued-prompt-count/task-02-status-summary-queued-count.md create mode 100644 docs/plans/sidebar-queued-prompt-count/task-03-attach-count-to-task-payloads.md create mode 100644 docs/plans/sidebar-queued-prompt-count/task-04-sidebar-badge-ui.md create mode 100644 docs/plans/sidebar-queued-prompt-count/task-05-e2e-and-verification.md create mode 100644 docs/specs/ui/sidebar-queued-prompt-count.md diff --git a/apps/backend/internal/backendapp/boot_state.go b/apps/backend/internal/backendapp/boot_state.go index 28fc953323..7fa361ebe6 100644 --- a/apps/backend/internal/backendapp/boot_state.go +++ b/apps/backend/internal/backendapp/boot_state.go @@ -656,6 +656,14 @@ func (b bootStateBuilder) taskDTOsWithSessionInfo(ctx context.Context, tasks []* b.logBootError("repair missing task status summaries", err) } } + // Stamp the authoritative per-task queued prompt count onto every summary so + // the boot payload shows the sidebar badge on first paint, matching the + // shared list/snapshot assembly. Best-effort: a counter failure omits the + // badge instead of failing the boot payload. + queuedByTask, queuedErr := b.p.taskSvc.CountPendingQueuedByTaskIDs(ctx, taskIDs) + if queuedErr != nil { + b.logBootError("queued prompt counts", queuedErr) + } result := make([]taskdto.TaskDTO, 0, len(tasks)) for _, task := range tasks { if task == nil { @@ -698,6 +706,16 @@ func (b bootStateBuilder) taskDTOsWithSessionInfo(ctx context.Context, tasks []* taskdto.EnrichTaskForegroundActivity(&dto, sessions, b.p.orchestratorSvc) } dto.StatusSummary = statusSummaries[task.ID] + if dto.StatusSummary != nil { + switch { + case queuedErr != nil: + // Counter failed: honor the documented no-badge fallback in the + // boot payload without persisting the cleared value. + dto.StatusSummary.QueuedPromptCount = 0 + case queuedByTask != nil: + dto.StatusSummary.QueuedPromptCount = queuedByTask[task.ID] + } + } result = append(result, dto) } return result diff --git a/apps/backend/internal/backendapp/gateway.go b/apps/backend/internal/backendapp/gateway.go index e5147ef59c..2e73280645 100644 --- a/apps/backend/internal/backendapp/gateway.go +++ b/apps/backend/internal/backendapp/gateway.go @@ -157,6 +157,7 @@ func provideGateway( log, orchestratorSvc, taskSvc, + orchestratorSvc.SessionTaskID, referenceValidator, ) queueHandlers.SetAttachmentClaimer(taskSvc) @@ -294,6 +295,9 @@ func provideGateway( } return task.WorkspaceID, nil }, + CountQueuedPrompts: func(ctx context.Context, taskID string) (int, error) { + return orchestratorSvc.GetMessageQueue().CountPendingByTask(ctx, taskID) + }, Logger: log, }) if err := projector.Start(ctx); err != nil { diff --git a/apps/backend/internal/backendapp/orchestrator.go b/apps/backend/internal/backendapp/orchestrator.go index 42141acf53..106d2e5dd7 100644 --- a/apps/backend/internal/backendapp/orchestrator.go +++ b/apps/backend/internal/backendapp/orchestrator.go @@ -143,6 +143,11 @@ func provideOrchestrator( // boot payload and task.updated events. taskSvc.SetForegroundActivityProvider(orchestratorSvc) + // Let the task service stamp status_summary.queued_prompt_count on task + // list/snapshot payloads (initial-load backstop for the sidebar badge; the + // status-summary projector keeps the field live between loads). + taskSvc.SetQueuedPromptCounter(orchestratorSvc.GetMessageQueue()) + // Per-user scoping for the session-keyed WS actions. The orchestrator // resolves sessions through its own repo handle, so it does not inherit the // task service's authorize* checks. diff --git a/apps/backend/internal/backendapp/status_summary_boot_test.go b/apps/backend/internal/backendapp/status_summary_boot_test.go index 079492ac01..6b489f15ac 100644 --- a/apps/backend/internal/backendapp/status_summary_boot_test.go +++ b/apps/backend/internal/backendapp/status_summary_boot_test.go @@ -80,3 +80,74 @@ func TestBootKanbanSnapshotHydratesTaskStatusSummaryInBatch(t *testing.T) { } t.Fatalf("task %q missing from boot snapshot: %s", task.ID, raw) } + +type bootFakeQueuedCounter struct { + byTask map[string]int +} + +func (f bootFakeQueuedCounter) CountPendingByTaskIDs(_ context.Context, taskIDs []string) (map[string]int, error) { + out := make(map[string]int, len(taskIDs)) + for _, id := range taskIDs { + out[id] = f.byTask[id] + } + return out, nil +} + +func TestBootKanbanSnapshotStampsQueuedPromptCount(t *testing.T) { + harness := newBootStateTestHarness(t) + ctx := context.Background() + workspaces, err := harness.taskSvc.ListWorkspaces(ctx) + if err != nil { + t.Fatalf("list workspaces: %v", err) + } + workflows, err := harness.taskSvc.ListWorkflows(ctx, workspaces[0].ID, true) + if err != nil { + t.Fatalf("list workflows: %v", err) + } + steps, err := harness.workflowSvc.ListStepsByWorkflow(ctx, workflows[0].ID) + if err != nil { + t.Fatalf("list workflow steps: %v", err) + } + task, err := harness.taskSvc.CreateTask(ctx, &service.CreateTaskRequest{ + WorkspaceID: workspaces[0].ID, + WorkflowID: workflows[0].ID, + WorkflowStepID: steps[0].ID, + Title: "Queued count boot task", + }) + if err != nil { + t.Fatalf("create task: %v", err) + } + harness.taskSvc.SetQueuedPromptCounter(bootFakeQueuedCounter{byTask: map[string]int{task.ID: 2}}) + + state := bootInitialState(ctx, nil, routeParams{ + taskSvc: harness.taskSvc, + services: &Services{Workflow: harness.workflowSvc}, + }, webapp.ClassifyRoute("/")) + raw, err := json.Marshal(state) + if err != nil { + t.Fatalf("marshal boot state: %v", err) + } + var decoded struct { + KanbanMulti struct { + Snapshots map[string]struct { + Tasks []struct { + ID string `json:"id"` + StatusSummary *statussummary.TaskStatusSummary `json:"statusSummary"` + } `json:"tasks"` + } `json:"snapshots"` + } `json:"kanbanMulti"` + } + if err := json.Unmarshal(raw, &decoded); err != nil { + t.Fatalf("decode boot state: %v", err) + } + for _, item := range decoded.KanbanMulti.Snapshots[workflows[0].ID].Tasks { + if item.ID != task.ID { + continue + } + if item.StatusSummary == nil || item.StatusSummary.QueuedPromptCount != 2 { + t.Fatalf("boot status summary queued prompt count = %#v, want 2", item.StatusSummary) + } + return + } + t.Fatalf("task %q missing from boot snapshot", task.ID) +} diff --git a/apps/backend/internal/orchestrator/event_handlers_agent.go b/apps/backend/internal/orchestrator/event_handlers_agent.go index d43113307c..913e07e09b 100644 --- a/apps/backend/internal/orchestrator/event_handlers_agent.go +++ b/apps/backend/internal/orchestrator/event_handlers_agent.go @@ -87,6 +87,13 @@ func (s *Service) publishQueueStatusEvent(ctx context.Context, sessionID string) "count": queueStatus.Count, "max": queueStatus.Max, } + if taskID, err := s.SessionTaskID(ctx, sessionID); err != nil { + s.logger.Warn("resolve session task for queue status event", + zap.String("session_id", sessionID), + zap.Error(err)) + } else if taskID != "" { + eventData["task_id"] = taskID + } s.logger.Debug("publishing queue status changed event", zap.String("session_id", sessionID), diff --git a/apps/backend/internal/orchestrator/handlers/queue_handlers.go b/apps/backend/internal/orchestrator/handlers/queue_handlers.go index 1d0f7e083b..f3bd15a17c 100644 --- a/apps/backend/internal/orchestrator/handlers/queue_handlers.go +++ b/apps/backend/internal/orchestrator/handlers/queue_handlers.go @@ -74,6 +74,12 @@ type QueueAccessAuthorizer interface { AuthorizeTaskSessionAccess(ctx context.Context, taskID, sessionID string) error } +// SessionTaskResolver returns the task that owns a session. It enriches the +// message.queue.status_changed event with task_id so task-scoped consumers +// (e.g. the status summary projector) can refresh per-task queued counts. +// An empty result omits the field; an error is logged and also omits it. +type SessionTaskResolver func(ctx context.Context, sessionID string) (string, error) + // QueueAttachmentClaimer binds staged file descriptors to the task/session // after a queue entry has been durably accepted. type QueueAttachmentClaimer interface { @@ -90,14 +96,15 @@ type queueEntryTaker interface { // QueueHandlers handles WebSocket message-queue operations. type QueueHandlers struct { - queueService QueueService - queueDrainer QueueDrainer - queueDispatcher QueueSendNowDispatcher - accessAuthorizer QueueAccessAuthorizer - eventBus bus.EventBus - logger *logger.Logger - referenceValidator entityrefs.SubmissionValidator - attachmentClaimer QueueAttachmentClaimer + queueService QueueService + queueDrainer QueueDrainer + queueDispatcher QueueSendNowDispatcher + accessAuthorizer QueueAccessAuthorizer + sessionTaskResolver SessionTaskResolver + eventBus bus.EventBus + logger *logger.Logger + referenceValidator entityrefs.SubmissionValidator + attachmentClaimer QueueAttachmentClaimer } // SetAttachmentClaimer wires the task attachment registry into queue adds. @@ -106,13 +113,16 @@ func (h *QueueHandlers) SetAttachmentClaimer(claimer QueueAttachmentClaimer) { h.attachmentClaimer = claimer } -// NewQueueHandlers creates a new QueueHandlers instance. +// NewQueueHandlers creates a new QueueHandlers instance. sessionTaskResolver +// enriches published queue status events with the owning task_id; nil keeps +// the payload unchanged. func NewQueueHandlers( queueService QueueService, eventBus bus.EventBus, log *logger.Logger, queueDrainer QueueDrainer, accessAuthorizer QueueAccessAuthorizer, + sessionTaskResolver SessionTaskResolver, validators ...entityrefs.SubmissionValidator, ) *QueueHandlers { var referenceValidator entityrefs.SubmissionValidator @@ -120,12 +130,13 @@ func NewQueueHandlers( referenceValidator = validators[0] } handlers := &QueueHandlers{ - queueService: queueService, - queueDrainer: queueDrainer, - accessAuthorizer: accessAuthorizer, - eventBus: eventBus, - logger: log.WithFields(zap.String("component", "queue-handlers")), - referenceValidator: referenceValidator, + queueService: queueService, + queueDrainer: queueDrainer, + accessAuthorizer: accessAuthorizer, + sessionTaskResolver: sessionTaskResolver, + eventBus: eventBus, + logger: log.WithFields(zap.String("component", "queue-handlers")), + referenceValidator: referenceValidator, } if dispatcher, ok := queueDrainer.(QueueSendNowDispatcher); ok { handlers.queueDispatcher = dispatcher @@ -822,14 +833,24 @@ func (h *QueueHandlers) publishStatus(ctx context.Context, sessionID string) { return } status := h.queueService.GetStatus(ctx, sessionID) + eventData := map[string]interface{}{ + fieldSessionID: sessionID, + "entries": status.Entries, + "count": status.Count, + fieldMax: status.Max, + } + if h.sessionTaskResolver != nil { + if taskID, err := h.sessionTaskResolver(ctx, sessionID); err != nil { + h.logger.Warn("resolve session task for queue status event", + zap.String("session_id", sessionID), + zap.Error(err)) + } else if taskID != "" { + eventData["task_id"] = taskID + } + } _ = h.eventBus.Publish(ctx, events.MessageQueueStatusChanged, bus.NewEvent( events.MessageQueueStatusChanged, "queue-handlers", - map[string]interface{}{ - fieldSessionID: sessionID, - "entries": status.Entries, - "count": status.Count, - fieldMax: status.Max, - }, + eventData, )) } diff --git a/apps/backend/internal/orchestrator/handlers/queue_handlers_publish_test.go b/apps/backend/internal/orchestrator/handlers/queue_handlers_publish_test.go new file mode 100644 index 0000000000..394703a830 --- /dev/null +++ b/apps/backend/internal/orchestrator/handlers/queue_handlers_publish_test.go @@ -0,0 +1,111 @@ +package handlers + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/events/bus" + "github.com/kandev/kandev/internal/orchestrator/messagequeue" + "github.com/stretchr/testify/require" +) + +// capturingQueueEventBus records the last message.queue.status_changed payload +// so tests can assert the task_id enrichment on the published event. +type capturingQueueEventBus struct { + lastData map[string]interface{} +} + +func (m *capturingQueueEventBus) Publish(_ context.Context, _ string, event *bus.Event) error { + if data, ok := event.Data.(map[string]interface{}); ok { + m.lastData = data + } + return nil +} +func (m *capturingQueueEventBus) Subscribe(_ string, _ bus.EventHandler) (bus.Subscription, error) { + return nil, nil +} +func (m *capturingQueueEventBus) QueueSubscribe(_, _ string, _ bus.EventHandler) (bus.Subscription, error) { + return nil, nil +} +func (m *capturingQueueEventBus) Request(_ context.Context, _ string, _ *bus.Event, _ time.Duration) (*bus.Event, error) { + return nil, nil +} +func (m *capturingQueueEventBus) Close() {} +func (m *capturingQueueEventBus) IsConnected() bool { return true } + +func setupQueueHandlersWithResolver( + t *testing.T, + resolver SessionTaskResolver, +) (*QueueHandlers, *messagequeue.Service, *capturingQueueEventBus) { + t.Helper() + log, err := logger.NewLogger(logger.LoggingConfig{ + Level: "error", + Format: "console", + OutputPath: "stderr", + }) + require.NoError(t, err) + events := &capturingQueueEventBus{} + svc := messagequeue.NewServiceMemory(log) + handlers := NewQueueHandlers(svc, events, log, nil, allowQueueAccess{}, resolver) + return handlers, svc, events +} + +func TestPublishStatusIncludesTaskIDWhenResolvable(t *testing.T) { + handlers, svc, events := setupQueueHandlersWithResolver(t, func(_ context.Context, sessionID string) (string, error) { + if sessionID == "s1" { + return "task-1", nil + } + return "", nil + }) + ctx := context.Background() + + msg := &messagequeue.QueuedMessage{SessionID: "s1", TaskID: "task-1", Content: "x", QueuedBy: messagequeue.QueuedByUser} + _, err := svc.QueueMessageWithMetadata(ctx, msg.SessionID, msg.TaskID, msg.Content, "", msg.QueuedBy, false, nil, nil) + require.NoError(t, err) + + handlers.publishStatus(ctx, "s1") + + require.NotNil(t, events.lastData, "expected a published queue status event") + require.Equal(t, "task-1", events.lastData["task_id"]) + require.Equal(t, "s1", events.lastData["session_id"]) +} + +func TestPublishStatusOmitsTaskIDWhenResolutionFails(t *testing.T) { + handlers, svc, events := setupQueueHandlersWithResolver(t, func(context.Context, string) (string, error) { + return "", errors.New("session gone") + }) + ctx := context.Background() + + msg := &messagequeue.QueuedMessage{SessionID: "s1", TaskID: "task-1", Content: "x", QueuedBy: messagequeue.QueuedByUser} + _, err := svc.QueueMessageWithMetadata(ctx, msg.SessionID, msg.TaskID, msg.Content, "", msg.QueuedBy, false, nil, nil) + require.NoError(t, err) + + handlers.publishStatus(ctx, "s1") + + require.NotNil(t, events.lastData, "expected a published queue status event") + _, hasTaskID := events.lastData["task_id"] + require.False(t, hasTaskID, "failed resolution must not fabricate a task_id") + require.Equal(t, 1, events.lastData["count"]) +} + +func TestPublishStatusWithoutResolverPublishesWithoutTaskID(t *testing.T) { + log, err := logger.NewLogger(logger.LoggingConfig{ + Level: "error", + Format: "console", + OutputPath: "stderr", + }) + require.NoError(t, err) + events := &capturingQueueEventBus{} + svc := messagequeue.NewServiceMemory(log) + handlers := NewQueueHandlers(svc, events, log, nil, allowQueueAccess{}, nil) + ctx := context.Background() + + handlers.publishStatus(ctx, "s1") + + require.NotNil(t, events.lastData, "expected a published queue status event") + _, hasTaskID := events.lastData["task_id"] + require.False(t, hasTaskID, "nil resolver must not fabricate a task_id") +} diff --git a/apps/backend/internal/orchestrator/handlers/queue_handlers_send_now_test.go b/apps/backend/internal/orchestrator/handlers/queue_handlers_send_now_test.go index b284fbf911..cd64ef6a2e 100644 --- a/apps/backend/internal/orchestrator/handlers/queue_handlers_send_now_test.go +++ b/apps/backend/internal/orchestrator/handlers/queue_handlers_send_now_test.go @@ -38,7 +38,7 @@ func TestWsSendNowValidatesScopeAndReturnsSentCount(t *testing.T) { require.NoError(t, err) dispatcher := &mockQueueSendNow{sentCount: 3} events := &mockEventBus{} - handlers := NewQueueHandlers(messagequeue.NewServiceMemory(log), events, log, dispatcher, allowQueueAccess{}) + handlers := NewQueueHandlers(messagequeue.NewServiceMemory(log), events, log, dispatcher, allowQueueAccess{}, nil) response, err := handlers.wsSendNow(context.Background(), createTestMessage(t, ws.ActionMessageQueueSendNow, map[string]interface{}{ "session_id": "session-1", "scope": orchestrator.QueueSendNowScopeAll, @@ -100,7 +100,7 @@ func TestWsSendNowMapsStableErrors(t *testing.T) { log, logErr := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "console", OutputPath: "stderr"}) require.NoError(t, logErr) dispatcher := &mockQueueSendNow{err: tc.err} - handlers := NewQueueHandlers(messagequeue.NewServiceMemory(log), &mockEventBus{}, log, dispatcher, allowQueueAccess{}) + handlers := NewQueueHandlers(messagequeue.NewServiceMemory(log), &mockEventBus{}, log, dispatcher, allowQueueAccess{}, nil) response, callErr := handlers.wsSendNow(context.Background(), createTestMessage(t, ws.ActionMessageQueueSendNow, map[string]interface{}{ "session_id": "session-1", "scope": orchestrator.QueueSendNowScopeAll, })) diff --git a/apps/backend/internal/orchestrator/handlers/queue_handlers_test.go b/apps/backend/internal/orchestrator/handlers/queue_handlers_test.go index 257ce94730..a1fe1de187 100644 --- a/apps/backend/internal/orchestrator/handlers/queue_handlers_test.go +++ b/apps/backend/internal/orchestrator/handlers/queue_handlers_test.go @@ -134,7 +134,7 @@ func setupQueueHandlersWithDrainer(t *testing.T, drainer QueueDrainer) (*QueueHa }) require.NoError(t, err) svc := messagequeue.NewServiceMemory(log) - return NewQueueHandlers(svc, &mockEventBus{}, log, drainer, allowQueueAccess{}), svc + return NewQueueHandlers(svc, &mockEventBus{}, log, drainer, allowQueueAccess{}, nil), svc } func setupQueueHandlersWithValidator(t *testing.T, validator entityrefs.SubmissionValidator) (*QueueHandlers, *messagequeue.Service) { @@ -146,7 +146,7 @@ func setupQueueHandlersWithValidator(t *testing.T, validator entityrefs.Submissi }) require.NoError(t, err) svc := messagequeue.NewServiceMemory(log) - return NewQueueHandlers(svc, &mockEventBus{}, log, nil, allowQueueAccess{}, validator), svc + return NewQueueHandlers(svc, &mockEventBus{}, log, nil, allowQueueAccess{}, nil, validator), svc } func createTestMessage(t *testing.T, action string, payload interface{}) *ws.Message { @@ -204,7 +204,7 @@ func TestQueueHandlersDenyUnauthorizedSessionActions(t *testing.T) { access := &fakeQueueAccessAuthorizer{sessionErr: errors.New("secret denial")} events := &mockEventBus{} drainer := &mockQueueDrainer{drained: true} - handlers := NewQueueHandlers(svc, events, log, drainer, access) + handlers := NewQueueHandlers(svc, events, log, drainer, access, nil) response, err := tc.call(handlers, context.Background(), createTestMessage(t, tc.action, tc.body(second.ID))) require.NoError(t, err) @@ -235,7 +235,7 @@ func TestQueueHandlersDenyUnauthorizedTaskSessionPairActions(t *testing.T) { svc := messagequeue.NewServiceMemory(log) access := &fakeQueueAccessAuthorizer{pairErr: errors.New("secret denial")} events := &mockEventBus{} - handlers := NewQueueHandlers(svc, events, log, nil, access) + handlers := NewQueueHandlers(svc, events, log, nil, access, nil) response, err := tc.call(handlers, context.Background(), createTestMessage(t, tc.action, map[string]interface{}{ "session_id": "s", "task_id": "other-task", "content": "secret", diff --git a/apps/backend/internal/orchestrator/messagequeue/count_pending_test.go b/apps/backend/internal/orchestrator/messagequeue/count_pending_test.go new file mode 100644 index 0000000000..83b9cc265e --- /dev/null +++ b/apps/backend/internal/orchestrator/messagequeue/count_pending_test.go @@ -0,0 +1,141 @@ +package messagequeue + +import ( + "context" + "reflect" + "testing" + + "github.com/kandev/kandev/internal/common/logger" +) + +// repositoriesUnderTest returns one SQLite and one memory Repository so the +// pending-count contract is proven on both storage backends. +func repositoriesUnderTest(t *testing.T) map[string]Repository { + t.Helper() + return map[string]Repository{ + "sqlite": newTestSQLiteRepo(t), + "memory": NewMemoryRepository(), + } +} + +func TestCountPendingByTaskIDsAccumulatesAcrossSessions(t *testing.T) { + for name, repo := range repositoriesUnderTest(t) { + t.Run(name, func(t *testing.T) { + ctx := context.Background() + insert := func(sessionID, taskID string) { + t.Helper() + msg := &QueuedMessage{SessionID: sessionID, TaskID: taskID, Content: "x", QueuedBy: QueuedByUser} + if err := repo.Insert(ctx, msg, 0); err != nil { + t.Fatalf("insert %s/%s: %v", sessionID, taskID, err) + } + } + insert("s1", "t1") + insert("s1", "t1") + insert("s2", "t1") + insert("s2", "t2") + + got, err := repo.CountPendingByTaskIDs(ctx, []string{"t1", "t2"}) + if err != nil { + t.Fatalf("CountPendingByTaskIDs: %v", err) + } + want := map[string]int{"t1": 3, "t2": 1} + if !reflect.DeepEqual(got, want) { + t.Errorf("counts = %v, want %v", got, want) + } + }) + } +} + +func TestCountPendingByTaskIDsExcludesReservedInFlight(t *testing.T) { + for name, repo := range repositoriesUnderTest(t) { + t.Run(name, func(t *testing.T) { + ctx := context.Background() + insert := func(sessionID, taskID, content string, metadata map[string]interface{}) { + t.Helper() + msg := &QueuedMessage{SessionID: sessionID, TaskID: taskID, Content: content, QueuedBy: QueuedByWorkflow, Metadata: metadata} + if err := repo.Insert(ctx, msg, 0); err != nil { + t.Fatalf("insert: %v", err) + } + } + // The durable lifecycle row must be the head so ReserveHead marks IT + // in flight; the ordinary pending row stays pending below it. + insert("s1", "t1", "lifecycle", map[string]interface{}{MetadataLifecycleDurable: true}) + insert("s1", "t1", "pending", nil) + // Reserve the durable head: it becomes in-flight and must not count. + if _, err := repo.ReserveHead(ctx, "s1"); err != nil { + t.Fatalf("reserve: %v", err) + } + + got, err := repo.CountPendingByTaskIDs(ctx, []string{"t1"}) + if err != nil { + t.Fatalf("CountPendingByTaskIDs: %v", err) + } + if got["t1"] != 1 { + t.Errorf("pending count for t1 = %d, want 1 (reserved in-flight row excluded)", got["t1"]) + } + }) + } +} + +func TestCountPendingByTaskIDsReturnsZeroForEmptyAndMissing(t *testing.T) { + for name, repo := range repositoriesUnderTest(t) { + t.Run(name, func(t *testing.T) { + ctx := context.Background() + got, err := repo.CountPendingByTaskIDs(ctx, []string{"missing"}) + if err != nil { + t.Fatalf("CountPendingByTaskIDs: %v", err) + } + want := map[string]int{"missing": 0} + if !reflect.DeepEqual(got, want) { + t.Errorf("counts = %v, want %v", got, want) + } + }) + } +} + +func TestCountPendingByTaskIDsIsolatesTasks(t *testing.T) { + for name, repo := range repositoriesUnderTest(t) { + t.Run(name, func(t *testing.T) { + ctx := context.Background() + for _, taskID := range []string{"ta", "tb", "tc"} { + msg := &QueuedMessage{SessionID: "s1", TaskID: taskID, Content: "x", QueuedBy: QueuedByUser} + if err := repo.Insert(ctx, msg, 0); err != nil { + t.Fatalf("insert %s: %v", taskID, err) + } + } + got, err := repo.CountPendingByTaskIDs(ctx, []string{"ta", "tc"}) + if err != nil { + t.Fatalf("CountPendingByTaskIDs: %v", err) + } + want := map[string]int{"ta": 1, "tc": 1} + if !reflect.DeepEqual(got, want) { + t.Errorf("counts = %v, want %v", got, want) + } + }) + } +} + +func TestServiceCountPendingByTask(t *testing.T) { + svc := NewServiceMemory(logger.Default()) + ctx := context.Background() + msg := &QueuedMessage{SessionID: "s1", TaskID: "t1", Content: "x", QueuedBy: QueuedByUser} + if err := svc.repo.Insert(ctx, msg, 0); err != nil { + t.Fatalf("insert: %v", err) + } + + count, err := svc.CountPendingByTask(ctx, "t1") + if err != nil { + t.Fatalf("CountPendingByTask: %v", err) + } + if count != 1 { + t.Errorf("CountPendingByTask(t1) = %d, want 1", count) + } + + empty, err := svc.CountPendingByTask(ctx, "nope") + if err != nil { + t.Fatalf("CountPendingByTask empty: %v", err) + } + if empty != 0 { + t.Errorf("CountPendingByTask(nope) = %d, want 0", empty) + } +} diff --git a/apps/backend/internal/orchestrator/messagequeue/repository.go b/apps/backend/internal/orchestrator/messagequeue/repository.go index 663c7b1bb4..057419420f 100644 --- a/apps/backend/internal/orchestrator/messagequeue/repository.go +++ b/apps/backend/internal/orchestrator/messagequeue/repository.go @@ -45,6 +45,13 @@ type Repository interface { // CountBySession returns the number of entries for a session. CountBySession(ctx context.Context, sessionID string) (int, error) + // CountPendingByTaskIDs returns the number of pending entries per task, + // keyed by task_id, for every requested task ID (zero when a task has no + // pending entries). Pending excludes durable lifecycle rows already + // reserved in flight, matching GetStatus semantics. The reserved exclusion + // is applied in Go via IsReservedInFlight, never by matching JSON in SQL. + CountPendingByTaskIDs(ctx context.Context, taskIDs []string) (map[string]int, error) + // TakeHead atomically returns and deletes the lowest-position entry for the // session. Returns nil, nil if the queue is empty. TakeHead(ctx context.Context, sessionID string) (*QueuedMessage, error) diff --git a/apps/backend/internal/orchestrator/messagequeue/repository_memory.go b/apps/backend/internal/orchestrator/messagequeue/repository_memory.go index 6d52fd4140..3ed72bf405 100644 --- a/apps/backend/internal/orchestrator/messagequeue/repository_memory.go +++ b/apps/backend/internal/orchestrator/messagequeue/repository_memory.go @@ -60,6 +60,31 @@ func (r *memoryRepository) PurgeTask(_ context.Context, taskID string) (int, err return removed, nil } +// CountPendingByTaskIDs counts pending entries per task, excluding durable +// lifecycle rows reserved in flight (filtered via IsReservedInFlight). +func (r *memoryRepository) CountPendingByTaskIDs(_ context.Context, taskIDs []string) (map[string]int, error) { + r.mu.Lock() + defer r.mu.Unlock() + counts := make(map[string]int, len(taskIDs)) + for _, taskID := range taskIDs { + counts[taskID] = 0 + } + if len(taskIDs) == 0 { + return counts, nil + } + for _, list := range r.entries { + for _, msg := range list { + if msg.IsReservedInFlight() { + continue + } + if _, wanted := counts[msg.TaskID]; wanted { + counts[msg.TaskID]++ + } + } + } + return counts, nil +} + func (r *memoryRepository) Insert(_ context.Context, msg *QueuedMessage, maxPerSession int) error { r.mu.Lock() defer r.mu.Unlock() diff --git a/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go b/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go index 444f921adf..04bc86bc9c 100644 --- a/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go +++ b/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go @@ -537,6 +537,46 @@ func (r *sqliteRepository) CountBySession(ctx context.Context, sessionID string) return n, err } +// CountPendingByTaskIDs counts pending entries per task, excluding durable +// lifecycle rows reserved in flight (filtered in Go via IsReservedInFlight). +func (r *sqliteRepository) CountPendingByTaskIDs(ctx context.Context, taskIDs []string) (map[string]int, error) { + counts := make(map[string]int, len(taskIDs)) + for _, taskID := range taskIDs { + counts[taskID] = 0 + } + if len(taskIDs) == 0 { + return counts, nil + } + query, args, err := sqlx.In(` + SELECT id, session_id, task_id, position, content, model, plan_mode, + attachments_json, metadata_json, queued_at, queued_by + FROM queued_messages + WHERE task_id IN (?) + `, taskIDs) + if err != nil { + return nil, fmt.Errorf("count pending by task ids: %w", err) + } + rows, err := r.ro.QueryxContext(ctx, r.ro.Rebind(query), args...) + if err != nil { + return nil, fmt.Errorf("count pending by task ids: %w", err) + } + defer func() { _ = rows.Close() }() + for rows.Next() { + msg, err := scanQueuedRow(rows) + if err != nil { + return nil, fmt.Errorf("count pending by task ids scan: %w", err) + } + if msg.IsReservedInFlight() { + continue + } + counts[msg.TaskID]++ + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("count pending by task ids rows: %w", err) + } + return counts, nil +} + func (r *sqliteRepository) TakeHead(ctx context.Context, sessionID string) (*QueuedMessage, error) { // Share the per-session lock with MergeIntoAbove so a drain and a merge on // the same queue are serialized in-process, not just at the DB layer. diff --git a/apps/backend/internal/orchestrator/messagequeue/service.go b/apps/backend/internal/orchestrator/messagequeue/service.go index 95e393faca..239937968f 100644 --- a/apps/backend/internal/orchestrator/messagequeue/service.go +++ b/apps/backend/internal/orchestrator/messagequeue/service.go @@ -649,6 +649,30 @@ func (s *Service) GetStatus(ctx context.Context, sessionID string) *QueueStatus } } +// CountPendingByTaskIDs returns the pending prompt count per task, keyed by +// task_id, for every requested task. Reserved in-flight lifecycle rows are +// excluded, matching GetStatus. Used by task-list assembly and the status +// summary projector to render per-task queued-prompt badges. +func (s *Service) CountPendingByTaskIDs(ctx context.Context, taskIDs []string) (map[string]int, error) { + counts, err := s.repo.CountPendingByTaskIDs(ctx, taskIDs) + if err != nil { + s.logger.Error("count pending by task ids failed", + zap.Int("task_count", len(taskIDs)), + zap.Error(err)) + return nil, err + } + return counts, nil +} + +// CountPendingByTask returns the pending prompt count for one task. +func (s *Service) CountPendingByTask(ctx context.Context, taskID string) (int, error) { + counts, err := s.CountPendingByTaskIDs(ctx, []string{taskID}) + if err != nil { + return 0, err + } + return counts[taskID], nil +} + // SnapshotSession returns the complete persisted queue state for rollback. // Unlike GetStatus, it includes durable lifecycle rows reserved by an in-flight // delivery so a later restore cannot erase them before acknowledgement. diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index fd2086240a..bef3b29eb7 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -960,6 +960,24 @@ func (s *Service) authorizeSession(ctx context.Context, sessionID string) error return s.sessionAccessCheck(ctx, sessionID) } +// SessionTaskID returns the task that owns a session, or "" when the session +// is unknown. Used to enrich message.queue.status_changed events with the +// task_id so task-scoped consumers (e.g. the status summary projector) can +// refresh per-task queued-prompt counts. +func (s *Service) SessionTaskID(ctx context.Context, sessionID string) (string, error) { + if sessionID == "" || s.repo == nil { + return "", nil + } + session, err := s.repo.GetTaskSession(ctx, sessionID) + if err != nil { + return "", err + } + if session == nil { + return "", nil + } + return session.TaskID, nil +} + // authorizeTask applies the configured per-user task check. No-op when unwired. func (s *Service) authorizeTask(ctx context.Context, taskID string) error { if s.taskAccessCheck == nil || taskID == "" { diff --git a/apps/backend/internal/orchestrator/session_task_id_test.go b/apps/backend/internal/orchestrator/session_task_id_test.go new file mode 100644 index 0000000000..ef07de7952 --- /dev/null +++ b/apps/backend/internal/orchestrator/session_task_id_test.go @@ -0,0 +1,74 @@ +package orchestrator + +import ( + "context" + "errors" + "testing" + + "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/task/models" +) + +// sessionTaskIDStubRepo satisfies sessionExecutorStore with only GetTaskSession +// usable — SessionTaskID calls nothing else, so the embedded nil interface is +// never touched (same pattern as pairStubRepo). +type sessionTaskIDStubRepo struct { + sessionExecutorStore + session *models.TaskSession + err error +} + +func (r *sessionTaskIDStubRepo) GetTaskSession(context.Context, string) (*models.TaskSession, error) { + return r.session, r.err +} + +func TestSessionTaskIDResolvesOwningTask(t *testing.T) { + s := &Service{ + logger: logger.Default(), + repo: &sessionTaskIDStubRepo{ + session: &models.TaskSession{ID: "s1", TaskID: "task-1"}, + }, + } + got, err := s.SessionTaskID(context.Background(), "s1") + if err != nil { + t.Fatalf("SessionTaskID: %v", err) + } + if got != "task-1" { + t.Fatalf("SessionTaskID = %q, want %q", got, "task-1") + } +} + +func TestSessionTaskIDEmptyForUnknownSession(t *testing.T) { + s := &Service{logger: logger.Default(), repo: &sessionTaskIDStubRepo{session: nil}} + got, err := s.SessionTaskID(context.Background(), "ghost") + if err != nil { + t.Fatalf("SessionTaskID: %v", err) + } + if got != "" { + t.Fatalf("SessionTaskID = %q, want empty for unknown session", got) + } +} + +func TestSessionTaskIDEmptyForEmptyInput(t *testing.T) { + s := &Service{ + logger: logger.Default(), + repo: &sessionTaskIDStubRepo{ + session: &models.TaskSession{ID: "s1", TaskID: "task-1"}, + }, + } + got, err := s.SessionTaskID(context.Background(), "") + if err != nil { + t.Fatalf("SessionTaskID: %v", err) + } + if got != "" { + t.Fatalf("SessionTaskID = %q, want empty for empty session id", got) + } +} + +func TestSessionTaskIDPropagatesRepoError(t *testing.T) { + repoErr := errors.New("db down") + s := &Service{logger: logger.Default(), repo: &sessionTaskIDStubRepo{err: repoErr}} + if _, err := s.SessionTaskID(context.Background(), "s1"); !errors.Is(err, repoErr) { + t.Fatalf("SessionTaskID error = %v, want %v", err, repoErr) + } +} diff --git a/apps/backend/internal/task/handlers/task_http_handlers.go b/apps/backend/internal/task/handlers/task_http_handlers.go index c6e9fb33bb..c02c63ffcb 100644 --- a/apps/backend/internal/task/handlers/task_http_handlers.go +++ b/apps/backend/internal/task/handlers/task_http_handlers.go @@ -238,6 +238,16 @@ func buildTaskDTOsWithSessionInfo( log.Warn("failed to repair missing task status summaries", zap.Error(err)) } } + // Stamp the authoritative per-task queued prompt count onto every summary. + // The projector keeps the summary field fresh between list loads; this + // fresh batch read is the initial-load backstop (e.g. after a restart the + // projector may not have observed every queue mutation yet). Never + // fabricate a summary here — a synthetic summary would make the frontend + // treat summary fields as authoritative and hide the coarse fallbacks. + queuedByTask, queuedErr := svc.CountPendingQueuedByTaskIDs(ctx, taskIDs) + if queuedErr != nil { + log.Warn("failed to load queued prompt counts for task list, omitting badges", zap.Error(queuedErr)) + } result := make([]dto.TaskDTO, 0, len(tasks)) for _, task := range tasks { sessions := sessionsByTask[task.ID] @@ -270,6 +280,20 @@ func buildTaskDTOsWithSessionInfo( taskDTO.TaskPendingAction = taskPendingActionPtr(sessions, pendingActionsBySession) dto.EnrichTaskForegroundActivity(&taskDTO, sessions, activityProvider) taskDTO.StatusSummary = statusSummaries[task.ID] + if taskDTO.StatusSummary != nil { + switch { + case queuedErr != nil: + // Counter failed: honor the documented no-badge fallback in the + // response without persisting the cleared value. + taskDTO.StatusSummary.QueuedPromptCount = 0 + case queuedByTask != nil: + // Fresh per-task count from the queue store; the projector keeps + // this field live between list loads, this batch read is the + // authoritative initial-load backstop. See the comment above the + // batch load. + taskDTO.StatusSummary.QueuedPromptCount = queuedByTask[task.ID] + } + } result = append(result, taskDTO) } return result, nil diff --git a/apps/backend/internal/task/handlers/task_http_queued_test.go b/apps/backend/internal/task/handlers/task_http_queued_test.go new file mode 100644 index 0000000000..9dc5b9361f --- /dev/null +++ b/apps/backend/internal/task/handlers/task_http_queued_test.go @@ -0,0 +1,153 @@ +package handlers + +import ( + "context" + "encoding/json" + "errors" + "path/filepath" + "testing" + + "github.com/jmoiron/sqlx" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/db" + "github.com/kandev/kandev/internal/events/bus" + "github.com/kandev/kandev/internal/task/models" + "github.com/kandev/kandev/internal/task/repository" + taskrepo "github.com/kandev/kandev/internal/task/repository/sqlite" + "github.com/kandev/kandev/internal/task/service" + "github.com/kandev/kandev/internal/task/statussummary" +) + +type fakeQueuedPromptCounter struct { + byTask map[string]int + err error +} + +func (f fakeQueuedPromptCounter) CountPendingByTaskIDs(_ context.Context, taskIDs []string) (map[string]int, error) { + if f.err != nil { + return nil, f.err + } + out := make(map[string]int, len(taskIDs)) + for _, id := range taskIDs { + out[id] = f.byTask[id] + } + return out, nil +} + +func newQueuedTaskDTOBuilder(t *testing.T) (*service.Service, *TaskHandlers, *taskrepo.Repository) { + t.Helper() + dbConn, err := db.OpenSQLite(filepath.Join(t.TempDir(), "test.db")) + require.NoError(t, err) + sqlxDB := sqlx.NewDb(dbConn, "sqlite3") + t.Cleanup(func() { + _ = sqlxDB.Close() + }) + repo, cleanup, err := repository.Provide(sqlxDB, sqlxDB, nil) + require.NoError(t, err) + t.Cleanup(func() { + _ = cleanup() + }) + require.NoError(t, repo.CreateWorkspace(context.Background(), &models.Workspace{ID: "ws-1", Name: "Workspace"})) + log, err := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "json", OutputPath: "stdout"}) + require.NoError(t, err) + svc := service.NewService(service.Repos{ + Workspaces: repo, Tasks: repo, TaskRepos: repo, + Workflows: repo, Messages: repo, Turns: repo, + Sessions: repo, GitSnapshots: repo, RepoEntities: repo, + Executors: repo, Environments: repo, TaskEnvironments: repo, + Reviews: repo, StatusSummaries: repo, + }, bus.NewMemoryEventBus(log), log, service.RepositoryDiscoveryConfig{}) + h := &TaskHandlers{service: svc, logger: log} + return svc, h, repo +} + +func createQueuedTestTask( + t *testing.T, + svc *service.Service, + repo *taskrepo.Repository, + counter fakeQueuedPromptCounter, +) *models.Task { + t.Helper() + ctx := context.Background() + svc.SetQueuedPromptCounter(counter) + task := &models.Task{ID: "task-1", WorkspaceID: "ws-1", Title: "Queued badge task"} + require.NoError(t, repo.CreateTask(ctx, task)) + require.NoError(t, repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: "s1", TaskID: "task-1", State: models.TaskSessionStateIdle, IsPrimary: true, + })) + return task +} + +func TestTaskDTOBuilderStampsQueuedPromptCount(t *testing.T) { + svc, h, repo := newQueuedTaskDTOBuilder(t) + ctx := context.Background() + task := createQueuedTestTask(t, svc, repo, fakeQueuedPromptCounter{byTask: map[string]int{"task-1": 3}}) + + result, err := h.toTaskDTOsWithSessionInfo(ctx, []*models.Task{task}) + require.NoError(t, err) + require.Len(t, result, 1) + require.NotNil(t, result[0].StatusSummary, "expected a status summary on the DTO") + assert.Equal(t, 3, result[0].StatusSummary.QueuedPromptCount) + + payload, err := json.Marshal(result[0]) + require.NoError(t, err) + assert.Contains(t, string(payload), `"queued_prompt_count":3`) +} + +func TestTaskDTOBuilderOmitsZeroQueuedPromptCount(t *testing.T) { + svc, h, repo := newQueuedTaskDTOBuilder(t) + ctx := context.Background() + task := createQueuedTestTask(t, svc, repo, fakeQueuedPromptCounter{byTask: map[string]int{"task-1": 0}}) + + result, err := h.toTaskDTOsWithSessionInfo(ctx, []*models.Task{task}) + require.NoError(t, err) + require.Len(t, result, 1) + require.NotNil(t, result[0].StatusSummary) + assert.Zero(t, result[0].StatusSummary.QueuedPromptCount) + + payload, err := json.Marshal(result[0]) + require.NoError(t, err) + assert.NotContains(t, string(payload), "queued_prompt_count") +} + +func TestTaskDTOBuilderDegradesWhenQueuedCounterFails(t *testing.T) { + svc, h, repo := newQueuedTaskDTOBuilder(t) + ctx := context.Background() + task := createQueuedTestTask(t, svc, repo, fakeQueuedPromptCounter{err: errors.New("queue store down")}) + + result, err := h.toTaskDTOsWithSessionInfo(ctx, []*models.Task{task}) + require.NoError(t, err, "a failed queued counter must not fail the task list") + require.Len(t, result, 1) + require.NotNil(t, result[0].StatusSummary, "the summary must survive a queued counter failure") + assert.Zero(t, result[0].StatusSummary.QueuedPromptCount, + "a failed counter must clear the badge in the response (documented fallback)") +} + +func TestTaskDTOBuilderPreservesProjectedCountWhenCounterUnwired(t *testing.T) { + _, h, repo := newQueuedTaskDTOBuilder(t) + ctx := context.Background() + task := &models.Task{ID: "task-1", WorkspaceID: "ws-1", Title: "Unwired counter task"} + require.NoError(t, repo.CreateTask(ctx, task)) + require.NoError(t, repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: "s1", TaskID: "task-1", State: models.TaskSessionStateIdle, IsPrimary: true, + })) + // A projected summary with a positive count, as if the projector had + // persisted it. With no counter wired, the assembly must NOT overwrite it. + changed, err := repo.CompareAndUpdateTaskStatusSummary(ctx, &statussummary.StoredTaskStatusSummary{ + TaskID: task.ID, + WorkspaceID: "ws-1", + Summary: statussummary.TaskStatusSummary{Revision: 3, QueuedPromptCount: 5}, + }) + require.NoError(t, err) + require.True(t, changed) + + result, err := h.toTaskDTOsWithSessionInfo(ctx, []*models.Task{task}) + require.NoError(t, err) + require.Len(t, result, 1) + require.NotNil(t, result[0].StatusSummary) + assert.Equal(t, 5, result[0].StatusSummary.QueuedPromptCount, + "an unwired counter must preserve the projected count, not zero it") +} diff --git a/apps/backend/internal/task/service/service.go b/apps/backend/internal/task/service/service.go index e5102176d2..cb55a3f596 100644 --- a/apps/backend/internal/task/service/service.go +++ b/apps/backend/internal/task/service/service.go @@ -272,6 +272,7 @@ type Service struct { statusSummaries repository.TaskStatusSummaryRepository attachmentSvc *AttachmentService statusSummaryPRs TaskStatusSummaryPRReader + queuedPromptCounter QueuedPromptCounter eventBus bus.EventBus logger *logger.Logger discoveryConfig RepositoryDiscoveryConfig diff --git a/apps/backend/internal/task/service/service_queued_prompt.go b/apps/backend/internal/task/service/service_queued_prompt.go new file mode 100644 index 0000000000..454ab7e2ef --- /dev/null +++ b/apps/backend/internal/task/service/service_queued_prompt.go @@ -0,0 +1,32 @@ +package service + +import "context" + +// QueuedPromptCounter supplies the pending prompt count per task (all +// sessions, pending semantics identical to message.queue.get). Satisfied at +// the composition root by an adapter over the messagequeue service; the task +// service keeps the interface narrow so it takes no hard orchestrator +// dependency and can be faked in tests. +type QueuedPromptCounter interface { + CountPendingByTaskIDs(ctx context.Context, taskIDs []string) (map[string]int, error) +} + +// SetQueuedPromptCounter wires the per-task queued-prompt counter used to +// stamp status_summary.queued_prompt_count on task list/snapshot payloads. +// Optional; when unset the field is left absent. +func (s *Service) SetQueuedPromptCounter(counter QueuedPromptCounter) { + if s != nil { + s.queuedPromptCounter = counter + } +} + +// CountPendingQueuedByTaskIDs returns the pending prompt count for each +// requested task, keyed by task_id. Returns nil (no error) when no counter is +// wired or no tasks were requested, so callers can distinguish "provider +// unavailable — preserve the projected count" from a successful zero lookup. +func (s *Service) CountPendingQueuedByTaskIDs(ctx context.Context, taskIDs []string) (map[string]int, error) { + if s == nil || s.queuedPromptCounter == nil || len(taskIDs) == 0 { + return nil, nil + } + return s.queuedPromptCounter.CountPendingByTaskIDs(ctx, taskIDs) +} diff --git a/apps/backend/internal/task/service/service_status_summary_rebuild.go b/apps/backend/internal/task/service/service_status_summary_rebuild.go index 19c01e5c62..32fe802440 100644 --- a/apps/backend/internal/task/service/service_status_summary_rebuild.go +++ b/apps/backend/internal/task/service/service_status_summary_rebuild.go @@ -55,6 +55,13 @@ func (s *Service) HydrateMissingTaskStatusSummaries( } prByTask, prObserved := s.loadSummaryPRs(ctx, taskIDs(missing)) gitBySession, gitObserved := s.loadSummaryGit(ctx, sessionIDsForTasks(missing, sessionsByTask)) + queuedByTask, queuedErr := s.CountPendingQueuedByTaskIDs(ctx, taskIDs(missing)) + if queuedErr != nil { + if s.logger != nil { + s.logger.Warn("failed to load queued prompt counts for status summary repair", zap.Error(queuedErr)) + } + queuedByTask = map[string]int{} + } now := time.Now().UTC() for _, task := range missing { if task == nil || task.ID == "" { @@ -67,6 +74,7 @@ func (s *Service) HydrateMissingTaskStatusSummaries( gitObserved, prByTask[task.ID], prObserved, + queuedByTask[task.ID], now, ) next := statussummary.BuildFromAuthoritative(input) @@ -192,16 +200,18 @@ func (s *Service) rebuildInput( gitObserved bool, prs []statussummary.PullRequestInput, prObserved bool, + queuedPromptCount int, now time.Time, ) statussummary.RebuildInput { input := statussummary.RebuildInput{ - Sessions: make([]statussummary.RebuildSession, 0, len(sessions)), - PendingActions: make(map[string]string), - ActivityObserved: s.foregroundActivity != nil, - PullRequests: prs, - PRObserved: prObserved, - GitObserved: gitObserved, - Now: now, + Sessions: make([]statussummary.RebuildSession, 0, len(sessions)), + PendingActions: make(map[string]string), + ActivityObserved: s.foregroundActivity != nil, + PullRequests: prs, + PRObserved: prObserved, + GitObserved: gitObserved, + QueuedPromptCount: maxInt(queuedPromptCount, 0), + Now: now, } countProvider, hasCountProvider := s.foregroundActivity.(activeSubagentCountProvider) for _, session := range sessions { diff --git a/apps/backend/internal/task/service/service_status_summary_rebuild_test.go b/apps/backend/internal/task/service/service_status_summary_rebuild_test.go index c403993ca9..05d8afd823 100644 --- a/apps/backend/internal/task/service/service_status_summary_rebuild_test.go +++ b/apps/backend/internal/task/service/service_status_summary_rebuild_test.go @@ -39,6 +39,18 @@ func (statusSummaryRebuildActivityProvider) ForegroundActivity(string) v1.Foregr func (statusSummaryRebuildActivityProvider) ActiveSubagentCount(string) int { return 3 } +type statusSummaryQueuedPromptCounter struct { + counts map[string]int +} + +func (c statusSummaryQueuedPromptCounter) CountPendingByTaskIDs(_ context.Context, taskIDs []string) (map[string]int, error) { + out := make(map[string]int, len(taskIDs)) + for _, id := range taskIDs { + out[id] = c.counts[id] + } + return out, nil +} + func TestHydrateMissingTaskStatusSummariesRepairsExistingTaskOnce(t *testing.T) { svc, _, repo := createTestService(t) ctx := context.Background() @@ -84,6 +96,7 @@ func TestHydrateMissingTaskStatusSummariesRepairsExistingTaskOnce(t *testing.T) svc.SetForegroundActivityProvider(statusSummaryRebuildActivityProvider{}) prReader := &statusSummaryRebuildPRReader{} svc.SetTaskStatusSummaryPRReader(prReader) + svc.SetQueuedPromptCounter(statusSummaryQueuedPromptCounter{counts: map[string]int{"task-1": 2}}) task := &models.Task{ID: "task-1", WorkspaceID: "ws-1"} sessions := map[string][]*models.TaskSession{task.ID: {session}} pending := map[string]models.TaskPendingAction{session.ID: models.TaskPendingActionPermission} @@ -116,6 +129,9 @@ func TestHydrateMissingTaskStatusSummariesRepairsExistingTaskOnce(t *testing.T) if summary.PullRequest == nil || summary.PullRequest.Number != 42 || summary.PullRequest.AggregateState != "ready" { t.Fatalf("pull request summary = %+v", summary.PullRequest) } + if summary.QueuedPromptCount != 2 { + t.Fatalf("queued prompt count = %d, want 2 from the queued counter", summary.QueuedPromptCount) + } if prReader.calls != 1 { t.Fatalf("PR reader calls = %d, want one batch read", prReader.calls) } diff --git a/apps/backend/internal/task/statussummary/model.go b/apps/backend/internal/task/statussummary/model.go index 11e939fbaa..73da925a8f 100644 --- a/apps/backend/internal/task/statussummary/model.go +++ b/apps/backend/internal/task/statussummary/model.go @@ -35,6 +35,11 @@ type TaskStatusSummary struct { ActiveError *ActiveErrorSummary `json:"active_error,omitempty"` Git *GitSummary `json:"git,omitempty"` PullRequest *PullRequestSummary `json:"pull_request,omitempty"` + // QueuedPromptCount is the number of prompts currently en-queued for the + // task across all of its sessions (pending semantics identical to + // message.queue.get). Omitted when zero so task rows without queued work + // stay byte-identical to earlier builds. + QueuedPromptCount int `json:"queued_prompt_count,omitempty"` } type PrimarySessionSummary struct { @@ -93,6 +98,9 @@ func (s TaskStatusSummary) Validate() error { if s.ActiveSubagentCount < 0 { return fmt.Errorf("active subagent count cannot be negative") } + if s.QueuedPromptCount < 0 { + return fmt.Errorf("queued prompt count cannot be negative") + } if err := validatePrimarySession(s.PrimarySession); err != nil { return err } @@ -185,6 +193,7 @@ func (s TaskStatusSummary) SemanticJSON() ([]byte, error) { ActiveError: s.ActiveError, Git: s.Git, PullRequest: s.PullRequest, + QueuedPromptCount: s.QueuedPromptCount, }) } @@ -196,4 +205,5 @@ type semanticPayload struct { ActiveError *ActiveErrorSummary `json:"active_error,omitempty"` Git *GitSummary `json:"git,omitempty"` PullRequest *PullRequestSummary `json:"pull_request,omitempty"` + QueuedPromptCount int `json:"queued_prompt_count,omitempty"` } diff --git a/apps/backend/internal/task/statussummary/model_test.go b/apps/backend/internal/task/statussummary/model_test.go index e2f26e512a..35c31ac2e8 100644 --- a/apps/backend/internal/task/statussummary/model_test.go +++ b/apps/backend/internal/task/statussummary/model_test.go @@ -56,6 +56,40 @@ func TestTaskStatusSummarySemanticJSONIsBoundedAndOmitsTransportMetadata(t *test } } +func TestTaskStatusSummaryQueuedPromptCountRoundTripsThroughSemanticJSON(t *testing.T) { + summary := TaskStatusSummary{QueuedPromptCount: 4} + payload, err := summary.SemanticJSON() + if err != nil { + t.Fatalf("semantic JSON: %v", err) + } + var decoded TaskStatusSummary + if err := json.Unmarshal(payload, &decoded); err != nil { + t.Fatalf("decode semantic JSON: %v", err) + } + if decoded.QueuedPromptCount != 4 { + t.Fatalf("queued prompt count after round trip = %d, want 4", decoded.QueuedPromptCount) + } + if !summary.SemanticEqual(decoded) { + t.Fatalf("semantic round trip changed value: %#v", decoded) + } +} + +func TestTaskStatusSummaryQueuedPromptCountValidateRejectsNegative(t *testing.T) { + if err := (TaskStatusSummary{QueuedPromptCount: -1}).Validate(); err == nil { + t.Fatal("negative queued prompt count should be rejected") + } + if err := (TaskStatusSummary{QueuedPromptCount: 0}).Validate(); err != nil { + t.Fatalf("zero queued prompt count rejected: %v", err) + } +} + +func TestTaskStatusSummaryQueuedPromptCountAffectsSemanticEquality(t *testing.T) { + base := TaskStatusSummary{} + if base.SemanticEqual(TaskStatusSummary{QueuedPromptCount: 2}) { + t.Fatal("different queued prompt counts must not compare equal") + } +} + func TestTaskStatusSummaryValidateBoundsErrorPreview(t *testing.T) { valid := TaskStatusSummary{ActiveError: &ActiveErrorSummary{Preview: strings.Repeat("é", MaxActiveErrorPreviewBytes/2)}} if err := valid.Validate(); err != nil { diff --git a/apps/backend/internal/task/statussummary/projector.go b/apps/backend/internal/task/statussummary/projector.go index 459e52bab2..58bd55d9c9 100644 --- a/apps/backend/internal/task/statussummary/projector.go +++ b/apps/backend/internal/task/statussummary/projector.go @@ -56,8 +56,13 @@ type ProjectorConfig struct { EventBus bus.EventBus ResolveWorkspace WorkspaceResolver LoadGitObservations GitObservationLoader - Logger *logger.Logger - Now func() time.Time + // CountQueuedPrompts returns the number of prompts currently en-queued for + // a task across all of its sessions (pending semantics identical to + // message.queue.get). Wired from the messagequeue service at the + // composition root; nil disables queued-prompt projection. + CountQueuedPrompts func(context.Context, string) (int, error) + Logger *logger.Logger + Now func() time.Time } // Projector converts authoritative, bounded occurrences into one complete @@ -69,6 +74,7 @@ type Projector struct { eventBus bus.EventBus resolveWorkspace WorkspaceResolver loadGitObservations GitObservationLoader + countQueuedPrompts func(context.Context, string) (int, error) logger *logger.Logger now func() time.Time @@ -88,6 +94,7 @@ type projectionState struct { workspaceID string revision uint64 current *TaskStatusSummary + queuedCount int sessions map[string]sessionObservation pending map[string]string taskPending string @@ -139,6 +146,7 @@ func NewProjector(cfg ProjectorConfig) *Projector { eventBus: cfg.EventBus, resolveWorkspace: cfg.ResolveWorkspace, loadGitObservations: cfg.LoadGitObservations, + countQueuedPrompts: cfg.CountQueuedPrompts, logger: log.WithFields(zap.String("component", "task-status-summary-projector")), now: now, state: make(map[string]*projectionState), @@ -168,6 +176,7 @@ func (p *Projector) Start(ctx context.Context) error { events.BuildPermissionRequestWildcardSubject(), events.BuildGitEventWildcardSubject(), events.GitHubTaskPRUpdated, + events.MessageQueueStatusChanged, } for _, pattern := range patterns { sub, err := p.eventBus.Subscribe(pattern, p.handleEvent) @@ -238,11 +247,57 @@ func (p *Projector) handleEvent(ctx context.Context, event *bus.Event) error { return fmt.Errorf("task status summary %q has no workspace", taskID) } + if event.Type == events.MessageQueueStatusChanged { + return p.applyQueueStatusEvent(ctx, state, taskID) + } + changed := p.applySourceEventLocked(state, event.Type, data) if !changed { return nil } - return p.persistAndPublishLocked(ctx, taskID, state) + _, err = p.persistAndPublishLocked(ctx, taskID, state) + return err +} + +// maxQueueCountPersistAttempts bounds the retry loop when a competing writer +// keeps winning the compare-and-set. Each attempt re-reads the stored summary +// (which persistAndPublishLocked syncs into state on rejection) and re-queries +// the authoritative count, so a rejection can never suppress the update by +// leaving state.queuedCount ahead of the persisted value. +const maxQueueCountPersistAttempts = 3 + +// applyQueueStatusEvent refreshes the task's queued prompt count from the +// authoritative queue store. The event payload's per-session count is not +// reused: the badge is per-task across all sessions, and the queue may have +// changed between the status snapshot and this projection. +func (p *Projector) applyQueueStatusEvent(ctx context.Context, state *projectionState, taskID string) error { + if p.countQueuedPrompts == nil { + return nil + } + for attempt := 0; attempt < maxQueueCountPersistAttempts; attempt++ { + count, err := p.countQueuedPrompts(ctx, taskID) + if err != nil { + return fmt.Errorf("count queued prompts for task %q: %w", taskID, err) + } + if count == state.queuedCount { + return nil + } + state.queuedCount = count + accepted, err := p.persistAndPublishLocked(ctx, taskID, state) + if err != nil { + return err + } + if accepted { + return nil + } + } + // The count self-corrects on the next queue event or list load, but a + // sustained contention run is worth surfacing so a repeated rejector is not + // silently starved. + p.logger.Warn("exhausted CAS retries updating queued prompt count", + zap.String("task_id", taskID), + zap.Int("attempts", maxQueueCountPersistAttempts)) + return nil } func (p *Projector) applySourceEventLocked(state *projectionState, eventType string, data map[string]interface{}) bool { @@ -455,6 +510,7 @@ func (p *Projector) restorePersistedState(ctx context.Context, taskID string, st } state.current = cloneSummary(summary) state.revision = summary.Revision + state.queuedCount = summary.QueuedPromptCount state.taskPending = summary.PendingAction if summary.PrimarySession != nil && summary.PrimarySession.ID != "" { state.sessions[summary.PrimarySession.ID] = sessionObservation{ @@ -726,21 +782,26 @@ func pullRequestObservationKey(data map[string]interface{}) string { return "" } -func (p *Projector) persistAndPublishLocked(ctx context.Context, taskID string, state *projectionState) error { +// persistAndPublishLocked persists the projected summary and publishes the +// replacement event. The boolean reports whether the write was accepted (or a +// no-op because the derived summary is already current); false means a +// competing writer won and the stored summary was reloaded into state, +// including queuedCount, so callers can retry against the refreshed revision. +func (p *Projector) persistAndPublishLocked(ctx context.Context, taskID string, state *projectionState) (bool, error) { next := deriveSummary(state) if state.current != nil && state.current.SemanticEqual(next) { - return nil + return true, nil } if state.revision == ^uint64(0) { - return fmt.Errorf("task status summary %q revision overflow", taskID) + return false, fmt.Errorf("task status summary %q revision overflow", taskID) } next.Revision = state.revision + 1 next.UpdatedAt = p.now().UTC() if err := next.Validate(); err != nil { - return fmt.Errorf("validate projected task status summary %q: %w", taskID, err) + return false, fmt.Errorf("validate projected task status summary %q: %w", taskID, err) } if p.store == nil { - return fmt.Errorf("task status summary store is unavailable") + return false, fmt.Errorf("task status summary store is unavailable") } accepted, err := p.store.CompareAndUpdateTaskStatusSummary(ctx, &StoredTaskStatusSummary{ TaskID: taskID, @@ -748,18 +809,19 @@ func (p *Projector) persistAndPublishLocked(ctx context.Context, taskID string, Summary: next, }) if err != nil { - return fmt.Errorf("persist projected task status summary %q: %w", taskID, err) + return false, fmt.Errorf("persist projected task status summary %q: %w", taskID, err) } if !accepted { rows, loadErr := p.store.LoadTaskStatusSummaries(ctx, []string{taskID}) if loadErr != nil { - return fmt.Errorf("reload projected task status summary %q: %w", taskID, loadErr) + return false, fmt.Errorf("reload projected task status summary %q: %w", taskID, loadErr) } if stored := rows[taskID]; stored != nil { state.current = cloneSummary(stored) state.revision = stored.Revision + state.queuedCount = stored.QueuedPromptCount } - return nil + return false, nil } state.current = cloneSummary(&next) state.revision = next.Revision @@ -767,10 +829,10 @@ func (p *Projector) persistAndPublishLocked(ctx context.Context, taskID string, payload := SummaryUpdated{TaskID: taskID, WorkspaceID: state.workspaceID, Summary: next} if err := p.eventBus.Publish(ctx, events.TaskStatusSummaryUpdated, bus.NewEvent(events.TaskStatusSummaryUpdated, "task-status-summary", payload)); err != nil { - return fmt.Errorf("publish task status summary %q: %w", taskID, err) + return false, fmt.Errorf("publish task status summary %q: %w", taskID, err) } } - return nil + return true, nil } func deriveSummary(state *projectionState) TaskStatusSummary { @@ -783,6 +845,7 @@ func deriveSummary(state *projectionState) TaskStatusSummary { ActiveError: deriveActiveError(state), Git: deriveGitSummary(state), PullRequest: derivePullRequestSummary(state), + QueuedPromptCount: state.queuedCount, } } diff --git a/apps/backend/internal/task/statussummary/projector_queued_test.go b/apps/backend/internal/task/statussummary/projector_queued_test.go new file mode 100644 index 0000000000..8d0b8ba285 --- /dev/null +++ b/apps/backend/internal/task/statussummary/projector_queued_test.go @@ -0,0 +1,212 @@ +package statussummary + +import ( + "context" + "sync/atomic" + "testing" + "time" + + "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/events" + "github.com/kandev/kandev/internal/events/bus" +) + +func TestProjectorQueueEventUpdatesQueuedPromptCount(t *testing.T) { + projector, store, eventBus, _, _ := newProjectorTest(t) + const taskID = "task-queued-count" + + publishSessionState(t, eventBus, taskID, "session-1", nil) + + projector.countQueuedPrompts = func(_ context.Context, id string) (int, error) { + if id != taskID { + return 0, nil + } + return 3, nil + } + + publishProjectorEvent(t, eventBus, events.MessageQueueStatusChanged, events.MessageQueueStatusChanged, map[string]interface{}{ + "task_id": taskID, + "session_id": "session-1", + }) + + summary := store.summary(taskID) + if summary == nil { + t.Fatal("expected a summary after the queue event") + } + if summary.QueuedPromptCount != 3 { + t.Fatalf("queued prompt count = %d, want 3", summary.QueuedPromptCount) + } +} + +func TestProjectorQueueEventWithUnchangedCountDoesNotRepublish(t *testing.T) { + projector, store, eventBus, updates, _ := newProjectorTest(t) + const taskID = "task-queued-unchanged" + + publishSessionState(t, eventBus, taskID, "session-1", nil) + baseline := updates.Load() + + projector.countQueuedPrompts = func(_ context.Context, id string) (int, error) { + return 2, nil + } + publishProjectorEvent(t, eventBus, events.MessageQueueStatusChanged, events.MessageQueueStatusChanged, map[string]interface{}{ + "task_id": taskID, + "session_id": "session-1", + }) + first := updates.Load() + if first == baseline { + t.Fatalf("first queue event should have published a summary update (baseline %d)", baseline) + } + + // Same count again: the projector must not bump the revision or republish. + publishProjectorEvent(t, eventBus, events.MessageQueueStatusChanged, events.MessageQueueStatusChanged, map[string]interface{}{ + "task_id": taskID, + "session_id": "session-1", + }) + if got := updates.Load(); got != first { + t.Fatalf("unchanged queued count republished: updates %d -> %d", first, got) + } + if summary := store.summary(taskID); summary.QueuedPromptCount != 2 { + t.Fatalf("queued prompt count = %d, want 2", summary.QueuedPromptCount) + } +} + +func TestProjectorRestoresQueuedPromptCountFromPersistedSummary(t *testing.T) { + _, store, eventBus, _, _ := newProjectorTest(t) + const taskID = "task-queued-restore" + + // Seed a persisted summary carrying a queued count from a previous run. + store.rows[taskID] = &StoredTaskStatusSummary{ + TaskID: taskID, + WorkspaceID: "workspace-1", + Summary: TaskStatusSummary{ + Revision: 7, + QueuedPromptCount: 5, + }, + } + + publishSessionState(t, eventBus, taskID, "session-1", nil) + + summary := store.summary(taskID) + if summary == nil { + t.Fatal("expected a summary") + } + if summary.QueuedPromptCount != 5 { + t.Fatalf("restored queued prompt count = %d, want 5", summary.QueuedPromptCount) + } +} + +func TestProjectorQueueEventWithoutTaskIDIsIgnored(t *testing.T) { + projector, store, eventBus, updates, _ := newProjectorTest(t) + + calls := 0 + projector.countQueuedPrompts = func(_ context.Context, id string) (int, error) { + calls++ + return 1, nil + } + + publishProjectorEvent(t, eventBus, events.MessageQueueStatusChanged, events.MessageQueueStatusChanged, map[string]interface{}{ + "session_id": "session-1", + }) + + if calls != 0 { + t.Fatalf("counter calls = %d, want 0 for an event without task_id", calls) + } + if got := updates.Load(); got != 0 { + t.Fatalf("summary publishes = %d, want 0", got) + } + if len(store.rows) != 0 { + t.Fatalf("queue event without task_id persisted summaries: %d rows", len(store.rows)) + } +} + +// competingWriterStore rejects the projector's first compare-and-update by +// landing a higher revision carrying the STALE queued count (simulating +// another writer persisting between the projector's count query and its +// write), then accepts subsequent writes. +type competingWriterStore struct { + base *projectorTestStore + rejected bool +} + +func (s *competingWriterStore) LoadTaskStatusSummaries( + ctx context.Context, + taskIDs []string, +) (map[string]*TaskStatusSummary, error) { + return s.base.LoadTaskStatusSummaries(ctx, taskIDs) +} + +func (s *competingWriterStore) CompareAndUpdateTaskStatusSummary( + ctx context.Context, + stored *StoredTaskStatusSummary, +) (bool, error) { + if !s.rejected { + s.rejected = true + rows, _ := s.base.LoadTaskStatusSummaries(ctx, []string{stored.TaskID}) + if row := rows[stored.TaskID]; row != nil { + competing := *row + competing.Revision = stored.Summary.Revision + 1 // beat the projector's attempt + _, _ = s.base.CompareAndUpdateTaskStatusSummary(ctx, &StoredTaskStatusSummary{ + TaskID: stored.TaskID, + WorkspaceID: stored.WorkspaceID, + Summary: competing, + }) + } + } + return s.base.CompareAndUpdateTaskStatusSummary(ctx, stored) +} + +func TestProjectorQueueEventRetriesAfterRejectedWrite(t *testing.T) { + const taskID = "task-queued-retry" + store := &competingWriterStore{base: newProjectorTestStore()} + store.base.rows[taskID] = &StoredTaskStatusSummary{ + TaskID: taskID, + WorkspaceID: "workspace-1", + Summary: TaskStatusSummary{Revision: 5, QueuedPromptCount: 0}, + } + eventBus := bus.NewMemoryEventBus(logger.Default()) + updates := new(atomic.Int64) + if _, err := eventBus.Subscribe(events.TaskStatusSummaryUpdated, func(_ context.Context, event *bus.Event) error { + updates.Add(1) + return nil + }); err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + projector := NewProjector(ProjectorConfig{ + Store: store, + EventBus: eventBus, + ResolveWorkspace: func(context.Context, string) (string, error) { + return "workspace-1", nil + }, + CountQueuedPrompts: func(context.Context, string) (int, error) { + return 3, nil + }, + Now: func() time.Time { return time.Date(2026, 8, 1, 18, 0, 0, 0, time.UTC) }, + }) + if err := projector.Start(ctx); err != nil { + cancel() + eventBus.Close() + t.Fatal(err) + } + t.Cleanup(func() { + cancel() + projector.Close() + eventBus.Close() + }) + + publishProjectorEvent(t, eventBus, events.MessageQueueStatusChanged, events.MessageQueueStatusChanged, map[string]interface{}{ + "task_id": taskID, + "session_id": "session-1", + }) + + summary := store.base.summary(taskID) + if summary == nil || summary.QueuedPromptCount != 3 { + t.Fatalf("stored queued prompt count = %+v, want 3 after rejected-write retry", summary) + } + if summary.Revision != 8 { + t.Fatalf("revision = %d, want 8 (5 -> competing 7 -> accepted 8)", summary.Revision) + } + if got := updates.Load(); got != 1 { + t.Fatalf("publishes = %d, want exactly 1 (the retried write)", got) + } +} diff --git a/apps/backend/internal/task/statussummary/rebuild.go b/apps/backend/internal/task/statussummary/rebuild.go index eec231ee36..e7ac9e5e39 100644 --- a/apps/backend/internal/task/statussummary/rebuild.go +++ b/apps/backend/internal/task/statussummary/rebuild.go @@ -55,7 +55,10 @@ type RebuildInput struct { GitObserved bool PullRequests []PullRequestInput PRObserved bool - Now time.Time + // QueuedPromptCount is the authoritative pending prompt count for the task + // (all sessions). Supplied by the caller; 0 means nothing is queued. + QueuedPromptCount int + Now time.Time } // BuildFromAuthoritative derives the same summary semantics used by the live @@ -131,6 +134,7 @@ func BuildFromAuthoritative(input RebuildInput) TaskStatusSummary { checksPassing: maxInt(pullRequest.ChecksPassing, 0), } } + state.queuedCount = maxInt(input.QueuedPromptCount, 0) return deriveSummary(state) } diff --git a/apps/backend/internal/task/statussummary/rebuild_queued_test.go b/apps/backend/internal/task/statussummary/rebuild_queued_test.go new file mode 100644 index 0000000000..79c634920e --- /dev/null +++ b/apps/backend/internal/task/statussummary/rebuild_queued_test.go @@ -0,0 +1,27 @@ +package statussummary + +import ( + "testing" + "time" +) + +func TestBuildFromAuthoritativeIncludesQueuedPromptCount(t *testing.T) { + now := time.Date(2026, 8, 1, 18, 0, 0, 0, time.UTC) + got := BuildFromAuthoritative(RebuildInput{ + Now: now, + QueuedPromptCount: 7, + }) + if got.QueuedPromptCount != 7 { + t.Fatalf("queued prompt count = %d, want 7", got.QueuedPromptCount) + } + if got.QueuedPromptCount != 0 && got.PrimarySession != nil { + t.Fatalf("unexpected session derivation: %+v", got) + } +} + +func TestBuildFromAuthoritativeDefaultsQueuedPromptCountToZero(t *testing.T) { + got := BuildFromAuthoritative(RebuildInput{Now: time.Now().UTC()}) + if got.QueuedPromptCount != 0 { + t.Fatalf("queued prompt count = %d, want 0", got.QueuedPromptCount) + } +} diff --git a/apps/web/components/task/mobile/session-task-switcher-sheet-hooks.test.ts b/apps/web/components/task/mobile/session-task-switcher-sheet-hooks.test.ts index ef908bacb3..ebe87c9906 100644 --- a/apps/web/components/task/mobile/session-task-switcher-sheet-hooks.test.ts +++ b/apps/web/components/task/mobile/session-task-switcher-sheet-hooks.test.ts @@ -134,3 +134,21 @@ describe("toSheetItem", () => { ).toBe(ERROR_PREVIEW); }); }); + +describe("toSheetItem queued prompt count", () => { + it("carries the queued prompt count from the task status summary", () => { + const item = toSheetItem( + task({ statusSummary: { revision: 3, updated_at: UPDATED_AT, queued_prompt_count: 2 } }), + emptyCtx(), + ); + expect(item.queuedCount).toBe(2); + }); + + it("leaves queuedCount undefined when nothing is queued", () => { + const item = toSheetItem( + task({ statusSummary: { revision: 4, updated_at: UPDATED_AT } }), + emptyCtx(), + ); + expect(item.queuedCount).toBeUndefined(); + }); +}); diff --git a/apps/web/components/task/mobile/session-task-switcher-sheet-hooks.ts b/apps/web/components/task/mobile/session-task-switcher-sheet-hooks.ts index 787939e540..505464a3d1 100644 --- a/apps/web/components/task/mobile/session-task-switcher-sheet-hooks.ts +++ b/apps/web/components/task/mobile/session-task-switcher-sheet-hooks.ts @@ -126,6 +126,9 @@ export function toSheetItem( isRemoteExecutor: task.isRemoteExecutor, remoteExecutorType: task.primaryExecutorType ?? undefined, remoteExecutorName: task.primaryExecutorName ?? undefined, + // Queued prompt count badge — same status summary source as the desktop + // sidebar mapper (buildSidebarItem) so both surfaces agree. + queuedCount: task.statusSummary?.queued_prompt_count, }; } diff --git a/apps/web/components/task/task-item-stats-row.tsx b/apps/web/components/task/task-item-stats-row.tsx index 7fd52ce430..dae4930740 100644 --- a/apps/web/components/task/task-item-stats-row.tsx +++ b/apps/web/components/task/task-item-stats-row.tsx @@ -1,31 +1,42 @@ "use client"; -import { useAppStore } from "@/components/state-provider"; +import { IconMail } from "@tabler/icons-react"; import { Tooltip, TooltipContent, TooltipTrigger } from "@kandev/ui/tooltip"; -import { cn, formatRelativeTime } from "@/lib/utils"; +import { useTranslation } from "react-i18next"; +import { useAppStore } from "@/components/state-provider"; import { isDebugUI } from "@/lib/config"; import type { SessionPollMode } from "@/lib/state/slices/session-runtime/types"; -import { useTranslation } from "react-i18next"; +import { cn, formatRelativeTime } from "@/lib/utils"; -/** Debug-overlay-only poll-mode legend. `label` stays English on purpose: the - * row is gated behind `isDebugUI()`, so these are developer diagnostics rather - * than user copy — and a SCREAMING_CASE const is invisible to - * `i18next/no-literal-string`, so nothing would have flagged it either way. */ -const POLL_MODE_CONFIG: Record = - { - fast: { letter: "F", color: "text-emerald-500", label: "focused, 2s polling" }, - slow: { letter: "S", color: "text-yellow-500", label: "subscribed, 30s polling" }, - paused: { letter: "P", color: "text-muted-foreground/40", label: "no subscribers" }, - }; +/** Debug-overlay-only poll-mode legend. Labels resolve through `t()` so the + * pseudo-locale stays complete; the row itself is gated behind `isDebugUI()`. */ +const POLL_MODE_CONFIG: Record< + SessionPollMode, + { letter: string; color: string; labelKey: string } +> = { + fast: { letter: "F", color: "text-emerald-500", labelKey: "sidebar:pollModeFastLabel" }, + slow: { letter: "S", color: "text-yellow-500", labelKey: "sidebar:pollModeSlowLabel" }, + paused: { + letter: "P", + color: "text-muted-foreground/40", + labelKey: "sidebar:pollModePausedLabel", + }, +}; +/** + * The sidebar row's metadata line: relative last-update time, PR number, + * queued-prompt mail badge, and (debug UI only) the session poll-mode letter. + */ export function TaskItemStatsRow({ updatedAt, prInfo, primarySessionId, + queuedCount, }: { updatedAt?: string; prInfo?: { number: number; state: string; aggregateState?: string }; primarySessionId?: string | null; + queuedCount?: number; }) { const { t } = useTranslation(); const pollMode = useAppStore((s) => @@ -34,9 +45,10 @@ export function TaskItemStatsRow({ : null, ); - if (!updatedAt && !prInfo && !pollMode) return null; + if (!updatedAt && !prInfo && !pollMode && !queuedCount) return null; const modeConfig = pollMode ? POLL_MODE_CONFIG[pollMode] : null; + const modeLabel = modeConfig ? t(modeConfig.labelKey) : ""; return ( @@ -44,6 +56,17 @@ export function TaskItemStatsRow({ {formatRelativeTime(updatedAt)} )} {prInfo && #{prInfo.number}} + {queuedCount ? ( + + + ) : null} {modeConfig && ( @@ -52,7 +75,7 @@ export function TaskItemStatsRow({ - {t("task:gitPollMode", { mode: pollMode, label: modeConfig.label })} + {t("task:gitPollMode", { mode: pollMode, label: modeLabel })} )} diff --git a/apps/web/components/task/task-item.test.tsx b/apps/web/components/task/task-item.test.tsx index a4e762a615..07f0cbe83a 100644 --- a/apps/web/components/task/task-item.test.tsx +++ b/apps/web/components/task/task-item.test.tsx @@ -391,3 +391,38 @@ describe("TaskItem background-running lifecycle", () => { expect(screen.queryByTestId(RUNNING_ICON_TEST_ID)).toBeNull(); }); }); + +describe("TaskItem queued prompt count badge", () => { + const QUEUED_BADGE_TEST_ID = "sidebar-task-queued-count"; + + it("shows a mail badge with the count when prompts are queued", () => { + renderTaskItem({ queuedCount: 3 }); + + const badge = screen.getByTestId(QUEUED_BADGE_TEST_ID); + expect(badge.textContent).toBe("3"); + expect(badge.querySelector("svg")).not.toBeNull(); + }); + + it("hides the badge when nothing is queued", () => { + renderTaskItem({ queuedCount: 0 }); + expect(screen.queryByTestId(QUEUED_BADGE_TEST_ID)).toBeNull(); + + renderTaskItem({ queuedCount: undefined }); + expect(screen.queryByTestId(QUEUED_BADGE_TEST_ID)).toBeNull(); + }); + + it("renders the metadata line for a row whose only metadata is the queued count", () => { + renderTaskItem({ queuedCount: 2, updatedAt: undefined }); + + const badge = screen.getByTestId(QUEUED_BADGE_TEST_ID); + expect(badge.textContent).toBe("2"); + }); + + it("localizes the accessible label with the count", () => { + renderTaskItem({ queuedCount: 1 }); + + expect(screen.getByTestId(QUEUED_BADGE_TEST_ID).getAttribute("aria-label")).toContain( + "queued prompt", + ); + }); +}); diff --git a/apps/web/components/task/task-item.tsx b/apps/web/components/task/task-item.tsx index 76bcfc456c..5ce98fa03d 100644 --- a/apps/web/components/task/task-item.tsx +++ b/apps/web/components/task/task-item.tsx @@ -85,6 +85,8 @@ type TaskItemProps = { onToggleSubtasks?: () => void; repositories?: string[]; prInfo?: { number: number; state: string; aggregateState?: string }; + /** Number of prompts currently en-queued for this task (mail badge). */ + queuedCount?: number; issueInfo?: { url: string; number: number }; isPinned?: boolean; agentErrorMessage?: string | null; @@ -324,6 +326,7 @@ function TaskItemContent({ repositories, updatedAt, prInfo, + queuedCount, issueInfo, agentErrorMessage, }: { @@ -338,6 +341,7 @@ function TaskItemContent({ repositories?: string[]; updatedAt?: string; prInfo?: { number: number; state: string; aggregateState?: string }; + queuedCount?: number; issueInfo?: { url: string; number: number }; agentErrorMessage?: string | null; }) { @@ -375,7 +379,12 @@ function TaskItemContent({ {repositories.join(" · ")} )} - + ); } @@ -428,6 +437,7 @@ export const TaskItem = memo(function TaskItem({ onToggleSubtasks, repositories, prInfo, + queuedCount, issueInfo, isPinned, agentErrorMessage, @@ -474,6 +484,7 @@ export const TaskItem = memo(function TaskItem({ repositories={repositories} updatedAt={updatedAt} prInfo={prInfo} + queuedCount={queuedCount} issueInfo={issueInfo} agentErrorMessage={agentErrorMessage} /> diff --git a/apps/web/components/task/task-session-sidebar-item.test.ts b/apps/web/components/task/task-session-sidebar-item.test.ts index 3d7cb78f4d..7d331fefc0 100644 --- a/apps/web/components/task/task-session-sidebar-item.test.ts +++ b/apps/web/components/task/task-session-sidebar-item.test.ts @@ -92,4 +92,22 @@ describe("buildSidebarItem", () => { expect(item.isArchived).toBe(true); }); + + it("carries the queued prompt count from the status summary", () => { + const item = buildSidebarItem( + task({ statusSummary: { revision: 4, updated_at: UPDATED_AT, queued_prompt_count: 3 } }), + emptyContext(), + ); + + expect(item.queuedCount).toBe(3); + }); + + it("leaves queuedCount undefined when the summary has no queued prompts", () => { + const item = buildSidebarItem( + task({ statusSummary: { revision: 5, updated_at: UPDATED_AT } }), + emptyContext(), + ); + + expect(item.queuedCount).toBeUndefined(); + }); }); diff --git a/apps/web/components/task/task-session-sidebar-item.ts b/apps/web/components/task/task-session-sidebar-item.ts index d7ef953681..a5a5f6b2fc 100644 --- a/apps/web/components/task/task-session-sidebar-item.ts +++ b/apps/web/components/task/task-session-sidebar-item.ts @@ -104,6 +104,7 @@ function sidebarStatus( hasPendingPermission: pending.permission, prInfo: summaryPRInfo(summary), issueInfo: issueInfoForTask(task), + queuedCount: summary?.queued_prompt_count, }; } diff --git a/apps/web/components/task/task-switcher-row.tsx b/apps/web/components/task/task-switcher-row.tsx index b933e807b5..b9f4f58cb2 100644 --- a/apps/web/components/task/task-switcher-row.tsx +++ b/apps/web/components/task/task-switcher-row.tsx @@ -154,6 +154,7 @@ function TaskRowItem({ updatedAt={task.updatedAt} repositories={task.repositories} prInfo={task.prInfo} + queuedCount={task.queuedCount} issueInfo={task.issueInfo} agentErrorMessage={task.agentErrorMessage} isSubTask={isSubTask} diff --git a/apps/web/components/task/task-switcher-types.ts b/apps/web/components/task/task-switcher-types.ts index 80d3d35767..75884f4f16 100644 --- a/apps/web/components/task/task-switcher-types.ts +++ b/apps/web/components/task/task-switcher-types.ts @@ -39,6 +39,8 @@ export type TaskSwitcherItem = { parentTaskId?: string; workspaceMode?: "inherit_parent" | "new_workspace" | "shared_group"; prInfo?: { number: number; state: string; aggregateState?: string }; + /** Number of prompts currently en-queued for this task (mail badge). */ + queuedCount?: number; isPRReview?: boolean; isIssueWatch?: boolean; issueInfo?: { url: string; number: number }; diff --git a/apps/web/e2e/helpers/api-client.ts b/apps/web/e2e/helpers/api-client.ts index 1c69c7368c..7c03d11596 100644 --- a/apps/web/e2e/helpers/api-client.ts +++ b/apps/web/e2e/helpers/api-client.ts @@ -2141,6 +2141,11 @@ export class ApiClient { }); } + /** Removes every pending queued message for a session (message.queue.cancel). */ + async clearQueue(sessionId: string): Promise { + await this.wsRequest("message.queue.cancel", { session_id: sessionId }); + } + // --- Integration config seeding (real API, not mock) --- async setJiraConfig(payload: { diff --git a/apps/web/e2e/tests/task/mobile-sidebar-queued-count.spec.ts b/apps/web/e2e/tests/task/mobile-sidebar-queued-count.spec.ts new file mode 100644 index 0000000000..a25c669d2d --- /dev/null +++ b/apps/web/e2e/tests/task/mobile-sidebar-queued-count.spec.ts @@ -0,0 +1,57 @@ +/** + * Mobile parity for the sidebar queued prompt count badge. + * + * The mobile task-switcher sheet renders through the same TaskItem rows as + * the desktop sidebar, so the mail badge must appear there too, without any + * hover dependency. Lives in `mobile-*.spec.ts` so the `mobile-chrome` + * Playwright project applies the mobile device automatically. + */ +import { test, expect } from "../../fixtures/test-base"; +import { SessionPage } from "../../pages/session-page"; + +test.describe("Mobile sidebar — queued prompt count", () => { + test("shows the queued count badge in the task switcher sheet", async ({ + testPage, + apiClient, + seedData, + }) => { + const task = await apiClient.createTask(seedData.workspaceId, "Mobile Queued Badge Task", { + workflow_id: seedData.workflowId, + workflow_step_id: seedData.startStepId, + repository_ids: [seedData.repositoryId], + }); + const { session_id: sessionId } = await apiClient.seedTaskSession(task.id, { + state: "IDLE", + agentProfileId: seedData.agentProfileId, + }); + await apiClient.queueMessage(task.id, sessionId, "Mobile queued prompt"); + + await testPage.goto(`/t/${task.id}`); + const session = new SessionPage(testPage); + await session.waitForLoad(); + + // Open the task switcher sheet from the mobile session top bar. + await testPage.getByTestId("mobile-session-menu").click(); + const sheet = testPage.getByRole("dialog"); + await expect(sheet.getByText("Mobile Queued Badge Task")).toBeVisible({ timeout: 10_000 }); + + const badge = sheet + .locator(`[data-testid='sortable-task-block'][data-task-id='${task.id}']`) + .getByTestId("sidebar-task-queued-count"); + await expect(badge).toBeVisible({ timeout: 10_000 }); + await expect(badge).toHaveText("1"); + + await testPage.screenshot({ + path: "e2e/test-results/screenshots/sidebar-queued-count-mobile.png", + fullPage: false, + }); + + // No horizontal overflow from the badge on a phone viewport: the badge + // stays within the sheet's horizontal bounds. + const [sheetBox, badgeBox] = await Promise.all([sheet.boundingBox(), badge.boundingBox()]); + expect(sheetBox).not.toBeNull(); + expect(badgeBox).not.toBeNull(); + expect(badgeBox!.x).toBeGreaterThanOrEqual(sheetBox!.x); + expect(badgeBox!.x + badgeBox!.width).toBeLessThanOrEqual(sheetBox!.x + sheetBox!.width); + }); +}); diff --git a/apps/web/e2e/tests/task/sidebar-queued-count.spec.ts b/apps/web/e2e/tests/task/sidebar-queued-count.spec.ts new file mode 100644 index 0000000000..637387532e --- /dev/null +++ b/apps/web/e2e/tests/task/sidebar-queued-count.spec.ts @@ -0,0 +1,105 @@ +/** + * E2E for the sidebar queued prompt count badge. + * + * Every task and subtask row in the left task view shows a mail badge with + * the number of prompts en-queued for that task. Covers: + * - Badge appears with the right count after prompts are queued + * - Badge disappears live (no reload) when the queue is cleared + * - No badge on a task with an empty queue + * - Subtask rows get the same badge treatment + */ +import { test, expect } from "../../fixtures/test-base"; +import { SessionPage } from "../../pages/session-page"; + +const stepOpts = (seedData: { workflowId: string; startStepId: string }) => ({ + workflow_id: seedData.workflowId, + workflow_step_id: seedData.startStepId, +}); + +function queuedBadge(page: SessionPage, taskId: string) { + return page.sidebar + .locator(`[data-testid='sortable-task-block'][data-task-id='${taskId}']`) + .getByTestId("sidebar-task-queued-count"); +} + +test.describe("Sidebar queued prompt count", () => { + test("shows a mail badge with the queued count and clears live", async ({ + testPage, + apiClient, + seedData, + }) => { + const task = await apiClient.createTask(seedData.workspaceId, "Queued Badge Task", { + ...stepOpts(seedData), + repository_ids: [seedData.repositoryId], + }); + const { session_id: sessionId } = await apiClient.seedTaskSession(task.id, { + state: "IDLE", + agentProfileId: seedData.agentProfileId, + }); + for (let index = 0; index < 3; index++) { + await apiClient.queueMessage(task.id, sessionId, `Queued prompt ${index + 1}`); + } + + await testPage.goto(`/t/${task.id}`); + const session = new SessionPage(testPage); + await session.waitForLoad(); + await expect(session.sidebar).toBeVisible({ timeout: 10_000 }); + + const badge = queuedBadge(session, task.id); + await expect(badge).toBeVisible({ timeout: 10_000 }); + await expect(badge).toHaveText("3"); + await expect(badge.locator("svg")).toHaveCount(1); + + await testPage.screenshot({ + path: "e2e/test-results/screenshots/sidebar-queued-count-desktop.png", + fullPage: false, + }); + + // Clearing the queue must remove the badge without a reload (live path + // through the status-summary broadcast). + await apiClient.clearQueue(sessionId); + await expect(badge).not.toBeVisible({ timeout: 10_000 }); + }); + + test("shows no badge for a task with an empty queue", async ({ + testPage, + apiClient, + seedData, + }) => { + const task = await apiClient.createTask(seedData.workspaceId, "Empty Queue Task", { + ...stepOpts(seedData), + repository_ids: [seedData.repositoryId], + }); + + await testPage.goto(`/t/${task.id}`); + const session = new SessionPage(testPage); + await session.waitForLoad(); + + await expect(session.sidebar.getByText("Empty Queue Task")).toBeVisible({ timeout: 10_000 }); + await expect(queuedBadge(session, task.id)).toHaveCount(0); + }); + + test("shows the badge on a subtask row", async ({ testPage, apiClient, seedData }) => { + const parent = await apiClient.createTask(seedData.workspaceId, "Badge Parent Task", { + ...stepOpts(seedData), + repository_ids: [seedData.repositoryId], + }); + const child = await apiClient.createTask(seedData.workspaceId, "Badge Child Task", { + ...stepOpts(seedData), + parent_id: parent.id, + }); + const { session_id: sessionId } = await apiClient.seedTaskSession(child.id, { + state: "IDLE", + agentProfileId: seedData.agentProfileId, + }); + await apiClient.queueMessage(child.id, sessionId, "Subtask queued prompt"); + + await testPage.goto(`/t/${parent.id}`); + const session = new SessionPage(testPage); + await session.waitForLoad(); + + const badge = queuedBadge(session, child.id); + await expect(badge).toBeVisible({ timeout: 10_000 }); + await expect(badge).toHaveText("1"); + }); +}); diff --git a/apps/web/lib/types/task-status-summary.ts b/apps/web/lib/types/task-status-summary.ts index 41377c7f17..8f02e33e17 100644 --- a/apps/web/lib/types/task-status-summary.ts +++ b/apps/web/lib/types/task-status-summary.ts @@ -10,6 +10,8 @@ export type TaskStatusSummary = { foreground_activity?: ForegroundActivity; active_subagent_count?: number; pending_action?: TaskPendingAction; + /** Number of prompts currently en-queued for the task (all sessions). */ + queued_prompt_count?: number; active_error?: { session_id: string; stamp: string; diff --git a/apps/web/src/locales/en/sidebar.json b/apps/web/src/locales/en/sidebar.json index 84ac1e9fd0..488a9d4f28 100644 --- a/apps/web/src/locales/en/sidebar.json +++ b/apps/web/src/locales/en/sidebar.json @@ -54,9 +54,14 @@ "offline": "Offline", "openStatus": "Open status", "pausedBadge": "paused", + "pollModeFastLabel": "focused, 2s polling", + "pollModePausedLabel": "no subscribers", + "pollModeSlowLabel": "subscribed, 30s polling", "preferences": "Preferences", "profileAndPassword": "Profile & Password", "projects": "Projects", + "queuedPromptCount_one": "{{count}} queued prompt", + "queuedPromptCount_other": "{{count}} queued prompts", "quickChat": "Quick Chat", "quickTerminal": "Quick terminal", "quickChatTerminals": "Terminals", diff --git a/apps/web/src/locales/pseudo/sidebar.json b/apps/web/src/locales/pseudo/sidebar.json index 165edfddd1..50314dfcf2 100644 --- a/apps/web/src/locales/pseudo/sidebar.json +++ b/apps/web/src/locales/pseudo/sidebar.json @@ -54,9 +54,14 @@ "offline": "Ōƒƒĺĩńē", "openStatus": "Ōƥēń śţàţũś", "pausedBadge": "ƥàũśēď", + "pollModeFastLabel": "ƒōćũśēď, 2ś ƥōĺĺĩńĝ", + "pollModePausedLabel": "ńō śũƀśćŕĩƀēŕś", + "pollModeSlowLabel": "śũƀśćŕĩƀēď, 30ś ƥōĺĺĩńĝ", "preferences": "Ƥŕēƒēŕēńćēś", "profileAndPassword": "Ƥŕōƒĩĺē & Ƥàśśŵōŕď", "projects": "Ƥŕōĵēćţś", + "queuedPromptCount_one": "{{count}} qũēũēď ƥŕōḿƥţ", + "queuedPromptCount_other": "{{count}} qũēũēď ƥŕōḿƥţś", "quickChat": "Qũĩćķ Ćĥàţ", "quickTerminal": "Qũĩćķ ţēŕḿĩńàĺ", "quickChatTerminals": "Ţēŕḿĩńàĺś", diff --git a/docs/plans/sidebar-queued-prompt-count/plan.md b/docs/plans/sidebar-queued-prompt-count/plan.md new file mode 100644 index 0000000000..e043d9d2c0 --- /dev/null +++ b/docs/plans/sidebar-queued-prompt-count/plan.md @@ -0,0 +1,182 @@ +--- +spec: docs/specs/ui/sidebar-queued-prompt-count.md +created: 2026-08-04 +status: implemented +--- + +# Implementation Plan: Sidebar Queued Prompt Count Badge + +## Overview + +Show, on every task and subtask row in the left task view, a mail-icon badge +with the number of prompts currently en-queued for that task (all sessions, +pending semantics identical to `message.queue.get`). The count rides the +existing per-task status summary: computed fresh at list/snapshot assembly for +initial load, and refreshed live by the status-summary projector subscribing +to the internal queue-status event and reusing the workspace-wide +`task.status_summary.updated` broadcast. Zero renders nothing. + +## Confirmed Current Behavior + +- The queue is per-session in `queued_messages` (SQLite table created in + `internal/orchestrator/messagequeue/repository_sqlite.go:initSchema`); each + row carries a denormalized `task_id` written at insert. Pending = any row + except durable lifecycle rows marked `lifecycle_reserved_in_flight` + (`Service.GetStatus` filters in memory via `QueuedMessage.IsReservedInFlight`). +- Queue status reaches clients only as the session-scoped WS event + `message.queue.status_changed` (`{session_id, entries, count, max}`), routed + by `TaskEventBroadcaster` via `BroadcastToSession`. The payload has no + `task_id`, and the event never reaches task-list clients for non-focused + sessions. +- The sidebar (`TaskSessionSidebar` → `TaskSwitcher` → `TaskItem` → + `TaskItemStatsRow`) consumes `KanbanTask.statusSummary` (type + `TaskStatusSummary`, snake_case JSON), populated from the HTTP workflow + snapshot and kept live by `task.status_summary.updated` (workspace-wide). +- No per-task queued count exists anywhere. There is no index on + `queued_messages.task_id`; the table is per-install small (bounded by + `max_per_session` per session), so a batched scan per list assembly is + acceptable. + +## Behavioral Contract + +- Badge = mail icon + pending prompt count on the row metadata line; hidden at + 0; same treatment for subtasks and tasks. +- Count = pending `queued_messages` rows with `task_id` = task (excluding + reserved-in-flight), across all of the task's sessions. +- Initial list/snapshot loads are authoritative (fresh count at assembly). +- Live updates flow through the existing status-summary broadcast; no new WS + action, no workspace-wide leak of queue contents. +- `message.queue.status_changed` keeps its session-scoped routing; its payload + gains an informational `task_id`. + +## Backend Design + +### Pending count query + +- Add `CountPendingByTaskIDs(ctx, taskIDs []string) (map[string]int, error)` + to `messagequeue.Repository` (interface, SQLite, memory). SQLite: one + `SELECT task_id, metadata_json FROM queued_messages WHERE task_id IN (...)` + + in-memory reserved filter (reuse `scanQueuedRow`/`IsReservedInFlight` + semantics; do not filter JSON in SQL). Memory: iterate per-session lists. +- Add `Service.CountPendingByTaskIDs` and single-task + `Service.CountPendingByTask(ctx, taskID) (int, error)` delegating to it. + Export a narrow `QueuedPromptCounter`-style adapter at the composition root + (backendapp) so neither the task service nor the status-summary projector + imports `messagequeue`. + +### Status summary live projection + +- `TaskStatusSummary` gains `QueuedPromptCount int \`json:"queued_prompt_count,omitempty"\``; + `Validate` rejects negatives; `SemanticJSON`/`semanticPayload` include it + (persistence round-trip); `SemanticEqual` picks it up via `reflect.DeepEqual`. +- Projector: add `CountQueuedPrompts func(ctx, taskID) (int, error)` to + `ProjectorConfig`; subscribe to `events.MessageQueueStatusChanged`; add + `queuedCount` to `projectionState`; new `applyQueueStatusEventLocked` case + that queries the count and returns changed only on delta; restore the count + in `restorePersistedState`; include it in `deriveSummary`. +- Publish sites add `task_id` to the internal event payload: + - `orchestrator/event_handlers_agent.go:publishQueueStatusEvent` — resolve + session→task via a new small `SessionTaskID(ctx, sessionID)` helper on the + orchestrator service (its repo owns sessions). + - `orchestrator/handlers/queue_handlers.go:publishStatus` — resolve via an + injected `SessionTaskResolver` (wire `orchestratorSvc.SessionTaskID` at + `backendapp/gateway.go:155`). On resolution failure, log and publish + without `task_id` (projector skips; next event or list load reconciles). + +### List/snapshot assembly + +- Task service: add optional `QueuedPromptCounter` interface + `SetQueuedPromptCounter` + (pattern of `ForegroundActivityProvider`); batch method + `CountPendingQueuedByTaskIDs`. +- `buildTaskDTOsWithSessionInfo` (all list/snapshot paths funnel through it: + HTTP list, workspace list, single task, WS list, workflow/workspace + snapshot): load counts for the listed tasks and stamp + `taskDTO.StatusSummary.QueuedPromptCount` (only when a summary exists — never + fabricate a summary, because `sidebarSessionStatus` treats a non-nil summary + as authoritative for other fields). +- `HydrateMissingTaskStatusSummaries`: include the count in the rebuilt + summaries for tasks it repairs (extend `statussummary.RebuildInput`). +- backendapp wires the adapter into the task service and the projector config. + +## Frontend Design + +- `lib/types/task-status-summary.ts`: add `queued_prompt_count?: number`. + (`toKanbanTask` already carries `status_summary` wholesale; WS summary + updates replace it wholesale, so no mapper change is needed.) +- `buildSidebarItem`: `queuedCount: summary?.queued_prompt_count` on the item. +- `TaskSwitcherItem`: new `queuedCount?: number`; `TaskRow` forwards it. +- `TaskItem`/`TaskItemStatsRow`: render + `IconMail` + count when `queuedCount > 0`; include it in the row's + early-return condition; `data-testid="sidebar-task-queued-count"`; accessible + name via `t("sidebar:queuedPromptCount", { count })` + (`_one`/`_other` plurals in `apps/web/src/locales/en/sidebar.json` + + pseudo-locale). Badge is inline-flex, shrink-safe, non-interactive. +- i18n ratchet: new copy uses `t()` (sidebar namespace); do not migrate + untouched literals in the file. + +## Responsive and Mobile Contract + +The badge is a static inline-flex text+icon pair inside the existing metadata +row. No hover dependency, no hit target, no new scroll owner; row overflow +behavior is unchanged (title truncates, badge does not wrap). Desktop and +mobile render identically because both use `TaskItem`. Mobile E2E asserts the +badge through the mobile sidebar flow. + +## TDD and Verification Strategy + +1. Backend repo/service tests for the count query (multi-session accumulation, + reserved-in-flight exclusion, empty, batch). RED → implement → GREEN. + `cd apps/backend && go test ./internal/orchestrator/messagequeue/...` +2. Model + projector tests: field round-trip, negative validation, queue event + updates count, no-change suppresses publish, restore keeps count. RED → + implement → GREEN. + `cd apps/backend && go test ./internal/task/statussummary/...` +3. DTO assembly + hydration tests (count stamped on existing summaries, + included in repaired summaries, absent when no summary). RED → implement → + GREEN. `cd apps/backend && go test ./internal/task/...` +4. Frontend unit tests: `buildSidebarItem` mapping, `TaskItem` badge + render/hide, plural/aria. RED → implement → GREEN. + `cd apps && pnpm --filter @kandev/web exec vitest run components/task/task-item.test.tsx components/task/task-session-sidebar-item.test.ts` +5. Full gates: `make -C apps/backend test lint build`, then from `apps`: + web typecheck, lint, `i18n:check`, `i18n:ratchet`. +6. E2E (desktop + mobile) as Task 05: + `cd apps/web && pnpm e2e -- tests/task/sidebar-queued-count.spec.ts tests/task/mobile-sidebar-queued-count.spec.ts`, + preserving/restoring queue state in teardown. + +## Implementation Waves + +- [x] [Task 01: Add pending prompt count queries to the message queue](task-01-pending-prompt-count-queries.md) — wave 1 +- [x] [Task 02: Add queued_prompt_count to the live status summary](task-02-status-summary-queued-count.md) — wave 2, after Task 01 +- [x] [Task 03: Attach the count to task list/snapshot payloads](task-03-attach-count-to-task-payloads.md) — wave 3, after Task 02 +- [x] [Task 04: Render the badge in the task sidebar](task-04-sidebar-badge-ui.md) — wave 3, after Task 02 (file-disjoint from Task 03; parallel-safe) +- [x] [Task 05: E2E coverage and verification](task-05-e2e-and-verification.md) — wave 4, after Tasks 03 and 04 + +This package does not authorize subagents; implementation remains in the +user-controlled primary session. + +## Results + +All five tasks implemented and verified: backend package tests, vet and golangci-lint on touched packages +green; 50 frontend unit tests plus typecheck/lint/i18n gates green; E2E desktop + mobile 4/4 on the built +artifact (badge visible, live clear, hidden at 0, subtasks). Full per-task command results are recorded in +each task file's Results section. Pre-existing VM-environment test failures (launcher/agentctl/local-repo) +reproduce at HEAD and are unrelated. + +## Environment Prerequisite + +If `apps/node_modules` is absent, run `pnpm install --frozen-lockfile` from +`apps/` before frontend RED tests. Do not change the lockfile. + +## Risks and Boundaries + +- Fabricating a status summary for a task that has none would flip + `sidebarSessionStatus` to summary-authoritative and hide the relative time; + assembly must stamp only existing summaries and let hydration create the + rest. +- JSON-in-SQL filtering of the reserved marker is fragile; keep the reserved + exclusion in Go (mirrors `GetStatus`). +- `task_id` has no index; list assembly does one batched scan of a bounded + table. If queue volumes ever grow, add a replayable migration for an index. +- WIP overflow queues (`queued_for_step_id`) are unrelated and out of scope. +- The projector may briefly lag on startup; fresh assembly at list time is the + correctness backstop, and the next queue event repairs the live path. diff --git a/docs/plans/sidebar-queued-prompt-count/task-01-pending-prompt-count-queries.md b/docs/plans/sidebar-queued-prompt-count/task-01-pending-prompt-count-queries.md new file mode 100644 index 0000000000..48e622ceef --- /dev/null +++ b/docs/plans/sidebar-queued-prompt-count/task-01-pending-prompt-count-queries.md @@ -0,0 +1,67 @@ +--- +id: "01-pending-prompt-count-queries" +title: "Add pending prompt count queries to the message queue" +status: done +wave: 1 +depends_on: [] +plan: "plan.md" +spec: "../../specs/ui/sidebar-queued-prompt-count.md" +--- + +# Task 01: Add Pending Prompt Count Queries + +## Acceptance + +- `messagequeue.Repository` (interface, SQLite, memory) exposes + `CountPendingByTaskIDs(ctx, taskIDs []string) (map[string]int, error)` + returning pending rows grouped by `task_id`. Pending excludes rows whose + metadata marks `lifecycle_reserved_in_flight` (same semantics as + `Service.GetStatus`); durable lifecycle rows not yet reserved still count. + The reserved exclusion happens in Go, never via JSON matching in SQL. +- `messagequeue.Service` exposes `CountPendingByTaskIDs` and + `CountPendingByTask(ctx, taskID string) (int, error)`. +- Unit tests prove: multi-session accumulation for one task, reserved-in-flight + exclusion, empty result maps to zero, batch with mixed found/missing task + IDs, and cross-task isolation. Memory and SQLite repositories both covered. + +## TDD Sequence + +1. Write repository tests (SQLite + memory) and service tests asserting the + behavior above. Run RED. +2. Implement `CountPendingByTaskIDs` on the interface, `sqliteRepository`, and + `memoryRepository`, plus the two service methods. +3. Run the package tests GREEN and refactor. + +## Verification + +```bash +cd apps/backend +go test ./internal/orchestrator/messagequeue/... +``` + +## Files Likely Touched + +- `apps/backend/internal/orchestrator/messagequeue/repository.go` +- `apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go` +- `apps/backend/internal/orchestrator/messagequeue/repository_memory.go` +- `apps/backend/internal/orchestrator/messagequeue/service.go` +- `apps/backend/internal/orchestrator/messagequeue/repository_sqlite_test.go` +- `apps/backend/internal/orchestrator/messagequeue/repository_memory_test.go` +- `apps/backend/internal/orchestrator/messagequeue/service_test.go` + +## Dependencies + +None. + +## Output Contract + +Report RED/GREEN evidence, the final method signatures, and changed files. +Update this task and `plan.md` status in the same implementation conversation. + +## Results + +- RED: `TestCountPendingByTaskIDs*` and `TestServiceCountPendingByTask` failed to compile (methods absent). +- GREEN: `go test -tags fts5 ./internal/orchestrator/messagequeue/` passes — SQLite and memory both cover + multi-session accumulation, reserved-in-flight exclusion, empty/missing task IDs, and cross-task isolation. +- Final signatures: `Repository.CountPendingByTaskIDs(ctx, taskIDs []string) (map[string]int, error)` + (SQLite + memory), `Service.CountPendingByTaskIDs`, `Service.CountPendingByTask(ctx, taskID) (int, error)`. diff --git a/docs/plans/sidebar-queued-prompt-count/task-02-status-summary-queued-count.md b/docs/plans/sidebar-queued-prompt-count/task-02-status-summary-queued-count.md new file mode 100644 index 0000000000..77db5b0046 --- /dev/null +++ b/docs/plans/sidebar-queued-prompt-count/task-02-status-summary-queued-count.md @@ -0,0 +1,94 @@ +--- +id: "02-status-summary-queued-count" +title: "Add queued_prompt_count to the live status summary" +status: done +wave: 2 +depends_on: ["01-pending-prompt-count-queries"] +plan: "plan.md" +spec: "../../specs/ui/sidebar-queued-prompt-count.md" +--- + +# Task 02: Add `queued_prompt_count` to the Live Status Summary + +## Acceptance + +- `statussummary.TaskStatusSummary` gains a new field; `Validate` + rejects negative values; `SemanticJSON`/`semanticPayload` include the field + so the persisted `task_status_summaries` payload round-trips it; old stored + rows decode with the field absent (0). + + ```go + QueuedPromptCount int `json:"queued_prompt_count,omitempty"` + ``` +- The projector (`internal/task/statussummary/projector.go`) subscribes to + `events.MessageQueueStatusChanged` and refreshes the affected task's count + through a new `CountQueuedPrompts func(ctx, taskID string) (int, error)` in + `ProjectorConfig`. A queue event whose recomputed count equals the current + value does not bump the revision or republish. The count is restored from + the persisted summary in `restorePersistedState` and included in + `deriveSummary`. +- The internal `message.queue.status_changed` event payload gains `task_id` at + both publish sites: + - `orchestrator.Service.publishQueueStatusEvent` resolves session→task via a + new `SessionTaskID(ctx, sessionID)` helper (orchestrator repo owns + sessions); missing/unknown session → log and publish without `task_id`. + - `handlers.QueueHandlers.publishStatus` resolves via an injected + `SessionTaskResolver` field (wired to `orchestratorSvc.SessionTaskID` in + `backendapp/gateway.go`); resolution failure → log, publish without + `task_id`. + The WS-facing `message.queue.status_changed` payload therefore also carries + `task_id` (informational); its session-scoped routing is unchanged. +- The projector is wired in `backendapp/gateway.go` with the count adapter from + `orchestratorSvc.GetMessageQueue()`. +- Unit tests prove: model round-trip + negative validation; projector updates + the summary on a queue event, suppresses republish on unchanged count, + restores the count from persisted state, and ignores events without + `task_id`; handlers publish `task_id` when resolvable and skip it when not. + +## TDD Sequence + +1. Model tests (round-trip, validation) + projector tests (event → count + update, no-change suppression, restore, missing task_id) + publish-site + tests. Run RED. +2. Implement the field, projector subscription/state/derive/restore, the + orchestrator `SessionTaskID` helper, the handler resolver, and gateway + wiring. +3. Run package tests GREEN, then refactor. + +## Verification + +```bash +cd apps/backend +go test ./internal/task/statussummary/... ./internal/orchestrator/... ./internal/backendapp/... +``` + +## Files Likely Touched + +- `apps/backend/internal/task/statussummary/model.go` +- `apps/backend/internal/task/statussummary/model_test.go` +- `apps/backend/internal/task/statussummary/projector.go` +- `apps/backend/internal/task/statussummary/projector_test.go` +- `apps/backend/internal/orchestrator/event_handlers_agent.go` (publishQueueStatusEvent) +- `apps/backend/internal/orchestrator/service.go` (SessionTaskID helper) +- `apps/backend/internal/orchestrator/handlers/queue_handlers.go` (publishStatus + resolver field) +- `apps/backend/internal/orchestrator/handlers/queue_handlers_test.go` +- `apps/backend/internal/backendapp/gateway.go` + +## Dependencies + +Task 01 supplies `messagequeue.Service.CountPendingByTask`. + +## Output Contract + +Report RED/GREEN evidence, the new event payload shape, and changed files. +Update this task and `plan.md` status in the same implementation conversation. + +## Results + +- RED: model/projector/publish-site tests failed against the unimplemented field and payload. +- GREEN: `go test -tags fts5 ./internal/task/statussummary/ ./internal/orchestrator/... ./internal/backendapp/...` — + round-trip + negative validation, queue-event count updates, no-change suppression, restore, missing-task_id ignore. +- The internal and WS `message.queue.status_changed` payload now carries `task_id` alongside + `{session_id, entries, count, max}`; session-scoped routing unchanged. +- Follow-up: `TestProjectorQueueEventRetriesAfterRejectedWrite` — competing writer lands a higher revision with + the stale count; the retried write wins (final stored count 3, revision 8, exactly one publish). diff --git a/docs/plans/sidebar-queued-prompt-count/task-03-attach-count-to-task-payloads.md b/docs/plans/sidebar-queued-prompt-count/task-03-attach-count-to-task-payloads.md new file mode 100644 index 0000000000..b67ea3e85b --- /dev/null +++ b/docs/plans/sidebar-queued-prompt-count/task-03-attach-count-to-task-payloads.md @@ -0,0 +1,95 @@ +--- +id: "03-attach-count-to-task-payloads" +title: "Attach the count to task list/snapshot payloads" +status: done +wave: 3 +depends_on: ["02-status-summary-queued-count"] +plan: "plan.md" +spec: "../../specs/ui/sidebar-queued-prompt-count.md" +--- + +# Task 03: Attach the Count to Task List/Snapshot Payloads + +## Acceptance + +- The task service gains an optional `QueuedPromptCounter` interface (batched: + `CountPendingByTaskIDs(ctx, taskIDs []string) (map[string]int, error)`) plus + `SetQueuedPromptCounter` (pattern of `ForegroundActivityProvider`), and a + batch helper `CountPendingQueuedByTaskIDs`. The task service must not import + `internal/orchestrator/messagequeue`. +- `buildTaskDTOsWithSessionInfo` (the single assembly used by HTTP list, + workspace list, single task, WS list, and workflow/workspace snapshot) + loads fresh counts for the listed tasks and stamps + `StatusSummary.QueuedPromptCount` onto each task's summary. It never + fabricates a summary for a task that has none — a synthetic summary would + flip `sidebarSessionStatus` to summary-authoritative and hide the relative + time (see Risks). Tasks without a summary keep no count field. +- `HydrateMissingTaskStatusSummaries` includes the count in summaries it + repairs: `statussummary.RebuildInput` gains the count and the hydrate path + fills it from the counter. Repaired rows therefore carry the badge on the + next load even before the projector touches them. +- backendapp wires the messagequeue-backed adapter into both the task service + and (already done by Task 02) the projector. +- Tests prove: assembly stamps fresh counts on tasks with summaries; repaired + summaries include counts; tasks without summaries omit the field; counter + failure degrades to no badge without failing the list; the JSON payload of a + list/snapshot response carries `status_summary.queued_prompt_count`. +- Counter-failure DTO contract: when the counter errors, the response DTO + clears `queued_prompt_count` to `0` for that request only — never persisted. + When the counter is unwired (provider unavailable), the DTO preserves the + projected count instead of stamping a zero. Regression tests cover both: an + errored counter yields a non-nil summary with count `0`; an unwired counter + preserves a positive projected count (`TestTaskDTOBuilderPreservesProjectedCountWhenCounterUnwired`). + +## TDD Sequence + +1. Write service/assembly/hydration tests for the acceptance cases. Run RED. +2. Implement the counter interface + setter, the batch helper, the + `RebuildInput` extension, the assembly stamping, and the gateway wiring. +3. Run the task-package and backendapp tests GREEN, then refactor. + +## Verification + +```bash +cd apps/backend +go test ./internal/task/... ./internal/backendapp/... +``` + +## Files Likely Touched + +- `apps/backend/internal/task/service/service.go` +- `apps/backend/internal/task/service/service_status_summary_rebuild.go` +- `apps/backend/internal/task/service/service_status_summary_rebuild_test.go` +- `apps/backend/internal/task/statussummary/rebuild.go` (RebuildInput) +- `apps/backend/internal/task/handlers/task_http_handlers.go` +- `apps/backend/internal/task/handlers/task_http_handlers_test.go` +- `apps/backend/internal/backendapp/gateway.go` + +## Dependencies + +Task 02 defines the summary field this task stamps; Task 01 supplies the +underlying count query. + +## Risks + +- Never construct a `TaskStatusSummary` solely to carry the count; the + frontend treats a non-nil summary as authoritative for session/updatedAt + fields. Stamp only existing summaries; hydration is the only creator. + +## Output Contract + +Report RED/GREEN evidence, the final DTO JSON for a task with and without a +summary, and changed files. Update this task and `plan.md` status in the same +implementation conversation. + +## Results + +- RED: assembly/hydration/rebuild tests failed against the missing count plumbing. +- GREEN: `go test -tags fts5 ./internal/task/... ./internal/backendapp/...` — assembly stamps fresh counts on + existing summaries, hydration includes counts in repaired rows, tasks without summaries omit the field, + and a counter failure clears the response count without persisting. +- `TestBootKanbanSnapshotStampsQueuedPromptCount` proves the boot kanban snapshot carries `queued_prompt_count`; + `TestTaskDTOBuilderPreservesProjectedCountWhenCounterUnwired` proves an unwired counter leaves a projected + positive count intact. +- DTO JSON for a queued task carries `"queued_prompt_count":3` inside `status_summary`; a task with no summary + carries no count field. diff --git a/docs/plans/sidebar-queued-prompt-count/task-04-sidebar-badge-ui.md b/docs/plans/sidebar-queued-prompt-count/task-04-sidebar-badge-ui.md new file mode 100644 index 0000000000..ed334cb69f --- /dev/null +++ b/docs/plans/sidebar-queued-prompt-count/task-04-sidebar-badge-ui.md @@ -0,0 +1,86 @@ +--- +id: "04-sidebar-badge-ui" +title: "Render the queued prompt count badge in the task sidebar" +status: done +wave: 3 +depends_on: ["02-status-summary-queued-count"] +plan: "plan.md" +spec: "../../specs/ui/sidebar-queued-prompt-count.md" +--- + +# Task 04: Render the Badge in the Task Sidebar + +## Acceptance + +- `TaskStatusSummary` (web type) gains `queued_prompt_count?: number`. + `toKanbanTask` and the WS `task.status_summary.updated` handler already move + the summary wholesale, so no mapper change is required. +- `buildSidebarItem` maps `summary.queued_prompt_count` to a new + `queuedCount?: number` on `TaskSwitcherItem`; `TaskRow` forwards it to + `TaskItem`. +- `TaskItemStatsRow` renders, on the metadata line, an inline-flex badge with + `IconMail` (from `@tabler/icons-react`) followed by the count whenever + `queuedCount > 0`. Count 0 (or undefined) renders nothing, and the row's + early-return condition includes the badge so a row with only a queue badge + still renders. The badge is `data-testid="sidebar-task-queued-count"`, + non-interactive, shrink-safe, and must not wrap or overflow the row. +- Accessible name is localized via `useTranslation`: + `sidebar:queuedPromptCount_one` (`{{count}} queued prompt`) / + `sidebar:queuedPromptCount_other` (`{{count}} queued prompts`) added to + `apps/web/src/locales/en/sidebar.json`, with the pseudo-locale regenerated. + No other literals in the file are migrated (i18n ratchet judges only added + lines). +- Subtasks and tasks share `TaskItem`, so both get identical treatment. +- Unit tests prove: `buildSidebarItem` carries the count; the badge renders + for counts > 0 on task and subtask rows, is absent at 0/undefined, carries + the pluralized accessible name, and does not render when the row would + otherwise be empty. + +## TDD Sequence + +1. Write `buildSidebarItem` and `TaskItem` tests for the cases above. Run RED. +2. Implement the type field, item plumbing, row rendering, and i18n keys. +3. Run focused tests GREEN, then typecheck, lint, `i18n:check`, and + `i18n:ratchet` (see Verification). + +## Verification + +```bash +cd apps +pnpm --filter @kandev/web exec vitest run \ + components/task/task-item.test.tsx \ + components/task/task-session-sidebar-item.test.ts +pnpm --filter @kandev/web run typecheck +pnpm --filter @kandev/web run lint +pnpm --filter @kandev/web run i18n:check +pnpm --filter @kandev/web run i18n:ratchet +``` + +## Files Likely Touched + +- `apps/web/lib/types/task-status-summary.ts` +- `apps/web/components/task/task-session-sidebar-item.ts` +- `apps/web/components/task/task-switcher.tsx` +- `apps/web/components/task/task-item.tsx` +- `apps/web/components/task/task-item.test.tsx` +- `apps/web/components/task/task-session-sidebar-item.test.ts` +- `apps/web/src/locales/en/sidebar.json` +- generated pseudo-locale catalog + +## Dependencies + +Task 02 fixes the summary JSON shape (`queued_prompt_count`) this task reads. + +## Output Contract + +Report RED/GREEN unit evidence, the badge markup, i18n status, typecheck/lint +status, and changed files. Update this task and `plan.md` status in the same +implementation conversation. + +## Results + +- RED: buildSidebarItem/TaskItem badge tests failed against the missing prop and rendering. +- GREEN: 50 focused unit tests pass (task-item, task-session-sidebar-item, mobile sheet-hooks, task-switcher); + web typecheck, lint, `i18n:check`, and `i18n:ratchet` all pass. +- The badge renders `IconMail` + count on the metadata line when > 0 (desktop and mobile sheet share + `TaskItem`), with localized `sidebar:queuedPromptCount_one/_other` aria; count 0 renders nothing. diff --git a/docs/plans/sidebar-queued-prompt-count/task-05-e2e-and-verification.md b/docs/plans/sidebar-queued-prompt-count/task-05-e2e-and-verification.md new file mode 100644 index 0000000000..ec3a853c08 --- /dev/null +++ b/docs/plans/sidebar-queued-prompt-count/task-05-e2e-and-verification.md @@ -0,0 +1,81 @@ +--- +id: "05-e2e-and-verification" +title: "E2E coverage and full verification" +status: done +wave: 4 +depends_on: ["03-attach-count-to-task-payloads", "04-sidebar-badge-ui"] +plan: "plan.md" +spec: "../../specs/ui/sidebar-queued-prompt-count.md" +--- + +# Task 05: E2E Coverage and Full Verification + +## Acceptance + +- Desktop Playwright spec (`apps/web/e2e/tests/task/sidebar-queued-count.spec.ts`): + - Create a task, seed an IDLE session, queue 3 prompts via the existing + `apiClient.queueMessage` helper, open the workspace, and assert the task + row in `SidebarTasksPage` shows the badge (`sidebar-task-queued-count`) + with text `3`. + - Clear the queue (add an `apiClient.clearQueue(sessionId)` helper backed by + `message.queue.cancel` if none exists) and assert the badge disappears + without a reload (live path through the status-summary broadcast). + - Assert a task with an empty queue never shows the badge. + - Teardown restores queue state so parallel specs cannot inherit counts. +- Mobile flow: same enqueue assertion driven through the mobile sidebar (per + `mobile-parity`), confirming no hover dependency and no horizontal overflow. +- Backend Go tests, frontend unit tests, typecheck, lint, and both i18n gates + all GREEN (commands below). + +## TDD Sequence + +1. Write the desktop spec (badge visible → live clear → hidden) and run it + RED against the pre-feature build expectations if feasible, else after the + unit work with a deliberately failing selector first. +2. Implement any missing E2E helper (`clearQueue`) and finish the spec. +3. Run the specs GREEN; run the full verification set. + +## Verification + +Every command below runs from the repository root (repo root is the parent of +`apps/`); subshells keep each block independent. + +```bash +# Desktop + mobile E2E (exact specs; both run in one Playwright invocation) +( cd apps/web && pnpm e2e -- tests/task/sidebar-queued-count.spec.ts tests/task/mobile-sidebar-queued-count.spec.ts ) + +# Backend gates +( cd apps/backend && make test lint build ) + +# Frontend gates +( cd apps && pnpm --filter @kandev/web run typecheck && pnpm --filter @kandev/web run lint && pnpm --filter @kandev/web run i18n:check && pnpm --filter @kandev/web run i18n:ratchet ) +``` + +## Files Likely Touched + +- `apps/web/e2e/tests/task/sidebar-queued-count.spec.ts` +- `apps/web/e2e/tests/task/mobile-sidebar-queued-count.spec.ts` (or a shared + helper reusing the `SidebarTasksPage` page object) +- `apps/web/e2e/helpers/api-client.ts` (clearQueue helper if absent) + +## Dependencies + +Tasks 03 and 04 deliver the payload and the badge; this task proves them end +to end on desktop and mobile. + +## Output Contract + +Report the E2E run results (badge visible, live clear, hidden at 0), the full +verification command results, and changed files. Update this task and +`plan.md` status in the same implementation conversation. + +## Results + +- E2E: `sidebar-queued-count.spec.ts` (3 tests) + `mobile-sidebar-queued-count.spec.ts` (1 test) — 4/4 pass on + the built artifact: badge shows the queued count with the mail icon, clears live with no reload, is absent at 0, + and covers subtask rows; mobile asserts the badge in the task-switcher sheet. +- Full gates: backend `go build ./...`, package tests, vet and golangci-lint on touched packages green; + web typecheck, lint, `i18n:check`, `i18n:ratchet` green. (Pre-existing VM-environment failures in + launcher/agentctl/local-repo packages reproduce at HEAD and are unrelated.) +- Manual: secondary dev instance on the LAN — badge rendered `📧 3` after enqueuing prompts and disappeared + without reload after `message.queue.cancel`. diff --git a/docs/specs/INDEX.md b/docs/specs/INDEX.md index 8bfdd84532..805dcbe9fc 100644 --- a/docs/specs/INDEX.md +++ b/docs/specs/INDEX.md @@ -154,6 +154,7 @@ Subscription quota tracking and per-agent cheap-model profile routing. | [sidebar-view-creation](ui/sidebar-view-creation.md) | shipped | | [command-panel sidebar task reveal](ui/command-panel-sidebar-task-reveal.md) | draft | | [sidebar-task-completion-icons](ui/sidebar-task-completion-icons.md) | shipped | +| [sidebar-queued-prompt-count](ui/sidebar-queued-prompt-count.md) | shipped | | [session-tab-delete-feedback](ui/session-tab-delete-feedback.md) | shipped | | [slash-command-composer](ui/slash-command-composer.md) | shipped | | [subagent-observability](ui/subagent-observability.md) | building | diff --git a/docs/specs/ui/sidebar-queued-prompt-count.md b/docs/specs/ui/sidebar-queued-prompt-count.md new file mode 100644 index 0000000000..1a516b08aa --- /dev/null +++ b/docs/specs/ui/sidebar-queued-prompt-count.md @@ -0,0 +1,149 @@ +--- +status: shipped +created: 2026-08-04 +owner: kandev +--- + +# Sidebar Queued Prompt Count Badge + +## Why + +A task with en-queued prompts looks identical to an idle one in the left +task view: nothing distinguishes "the agent is busy and three follow-ups are +waiting" from "nothing is waiting". Users bounce between the sidebar and each +task's queue panel to find where their typed work is stuck. A compact per-row +count on the metadata line (the same line that already carries the last +update time and PR number) makes queued work visible at a glance, for tasks +and subtasks alike. + +## What + +### The badge + +- Every task row and subtask row in the left task view (`TaskSessionSidebar`) + shows, on its metadata line, a badge with a mail icon followed by the number + of prompts currently en-queued for that task. +- The count covers all of the task's sessions. It counts persisted queue + entries whose `task_id` is the task, excluding rows already reserved for + in-flight delivery (the same pending semantics as `message.queue.get`). +- A count of `0` renders no badge: the metadata line is unchanged from today + for tasks with nothing queued. +- The badge is display-only: it is not a button, has no hover dependency, and + never pushes the row to a second line. + +### Data path + +- **Initial load:** every task list/snapshot endpoint (HTTP list, workspace + list, workflow/workspace snapshot, WS list, single task) carries the count in + the task's status summary (`status_summary.queued_prompt_count`). The count + is computed fresh from the queue table at assembly time so the sidebar is + correct on first paint regardless of projector state. +- **Live updates:** the task status summary projector subscribes to the + internal `message.queue.status_changed` event, recomputes the affected + task's pending count, and publishes the existing workspace-wide + `task.status_summary.updated` broadcast. The sidebar therefore updates for + every task in the workspace without any new per-session subscription. +- **Queue panel payload:** the WebSocket `message.queue.status_changed` event + (session-scoped, unchanged routing) gains an informational `task_id` field. + Its `entries` content remains session-scoped; it is never broadcast + workspace-wide. + +## API Surface + +No new actions or endpoints. Payload additions only: + +```text +task status summary (HTTP DTO, WS task.status_summary.updated) + status_summary.queued_prompt_count: int // omitted when 0 + +message.queue.status_changed (WebSocket, session-scoped as today) + task_id: string // informational addition +``` + +## Data Model + +No table migration. The count derives from the existing `queued_messages` +table (`task_id` column, written at insert from the session's owning task). +Pending semantics match `messagequeue.Service.GetStatus`: a row carrying +`lifecycle_reserved_in_flight` metadata is excluded; durable lifecycle rows +not yet reserved remain pending and count. The persisted +`task_status_summaries` payload gains `queued_prompt_count` through the +existing `SemanticJSON` envelope; old rows decode with the field absent +(0). + +## State and Concurrency + +- The projector refreshes the count only when the queue event's computed count + differs from its current value; unchanged counts do not bump the summary + revision or republish. +- The projector restores the last persisted count on restart and corrects it on + the next queue event. List/snapshot assembly always reads the count fresh, + so a missed event or projector startup ordering cannot leave a stale badge on + initial load. +- A queue mutation and a list assembly racing read the same queue table; the + list response may carry a count from either side of the mutation, both + authoritative snapshots. +- Reserved-in-flight rows are excluded exactly as in `GetStatus`; a prompt + already handed to a dispatch attempt is not "queued". + +## Responsive and Mobile Behavior + +- The badge renders inside the existing metadata row on both desktop and + mobile sidebar layouts; it is a static text+icon pair, so it adds no hit + target, no hover behavior, and no scroll owner. +- The row's ellipsis/overflow behavior is unchanged: the badge is inline-flex + and shrink-safe, so narrow viewports truncate the title rather than wrap the + badge. +- The accessible name is localized; the visible text is only the count. + +## Failure Modes + +- **Queue event missed (startup ordering, projector down):** initial list + loads still show the correct count (fresh assembly); the next queue event + repairs the live path. +- **Session→task resolution fails at a publish site:** the event is published + without `task_id`; the projector skips it and the next successful event (or + the next list load) reconciles. +- **Summary row missing and count is 0:** nothing to show; the row behaves as + today. +- **Summary row missing and count > 0 (hydration failed):** the badge is + absent on that row until a summary exists; the queue panel remains the + fallback surface. Same degradation class as other summary-only fields. +- **Count query failure at assembly:** the list endpoint logs and serves the + rest of the row without the badge: the response DTO clears any existing + `queued_prompt_count` to `0` for that request only. The cleared value is + never persisted back to the summary store, so a later successful count or + projector event restores the badge. Matching the existing status-summary + load behavior. +- **Queued-counter provider unavailable (unwired):** the DTO leaves the + projected `queued_prompt_count` untouched instead of stamping a zero over + it, so a previously projected positive count survives list loads. + +## Scenarios + +- **GIVEN** a task with three prompts queued across its sessions, **WHEN** the + sidebar loads, **THEN** its row shows a mail icon and `3` on the metadata + line. +- **GIVEN** a task with an empty queue, **WHEN** the sidebar loads, **THEN** + its row shows no mail badge and the metadata line is unchanged. +- **GIVEN** a task with two queued prompts, **WHEN** one is drained, **THEN** + the badge updates to `1` without a reload, and to nothing when the last one + drains. +- **GIVEN** a prompt reserved for in-flight delivery while a later prompt + remains pending, **WHEN** the sidebar renders, **THEN** only the pending + prompt counts. +- **GIVEN** a subtask with one queued prompt, **WHEN** the sidebar renders the + subtree, **THEN** the subtask row shows the same badge treatment as a + top-level task. +- **GIVEN** a phone viewport, **WHEN** the sidebar renders a row with queued + prompts, **THEN** the badge is visible without hover and the row does not + overflow horizontally. + +## Out of Scope + +- Clicking the badge to open the queue panel (the row click still navigates). +- Per-session breakdown of the count. +- WIP overflow queues (`queued_for_step_id`) — a different queue concept with + its own UI. +- Queue capacity display, queue contents, or provenance in the sidebar. +- Changing the session-scoped routing of `message.queue.status_changed`.