diff --git a/apps/backend/internal/events/bus/memory.go b/apps/backend/internal/events/bus/memory.go index 113f0d20eb..fe65de9157 100644 --- a/apps/backend/internal/events/bus/memory.go +++ b/apps/backend/internal/events/bus/memory.go @@ -332,8 +332,9 @@ func compilePattern(pattern string) *regexp.Regexp { // Replace escaped \* with regex for single token (anything except .) escaped = strings.ReplaceAll(escaped, `\*`, `[^.]+`) - // Replace escaped \> with regex for remaining tokens (anything) - escaped = strings.ReplaceAll(escaped, `\>`, `.+`) + // regexp.QuoteMeta does not escape > because it has no special meaning in + // regular expressions. Replace it directly with a multi-token matcher. + escaped = strings.ReplaceAll(escaped, `>`, `.+`) // Anchor the pattern escaped = "^" + escaped + "$" diff --git a/apps/backend/internal/events/bus/memory_test.go b/apps/backend/internal/events/bus/memory_test.go index e9d6f078fb..2ed03188f9 100644 --- a/apps/backend/internal/events/bus/memory_test.go +++ b/apps/backend/internal/events/bus/memory_test.go @@ -178,11 +178,6 @@ func TestMemoryEventBus_SingleTokenWildcard(t *testing.T) { } func TestMemoryEventBus_MultiTokenWildcard(t *testing.T) { - // Note: The current implementation has a bug where > wildcard doesn't work correctly - // because regexp.QuoteMeta doesn't escape > (it's not a special regex char). - // This test documents the current behavior. When the bug is fixed, update this test. - t.Skip("Skipping: > wildcard has a known bug in compilePattern - regexp.QuoteMeta doesn't escape >") - log := newTestLogger(t) bus := NewMemoryEventBus(log) defer bus.Close() diff --git a/apps/backend/internal/gateway/websocket/task_notifications.go b/apps/backend/internal/gateway/websocket/task_notifications.go index b4197d4db2..e03c827fd3 100644 --- a/apps/backend/internal/gateway/websocket/task_notifications.go +++ b/apps/backend/internal/gateway/websocket/task_notifications.go @@ -41,7 +41,7 @@ func RegisterTaskNotifications(ctx context.Context, eventBus bus.EventBus, hub * b.subscribe(eventBus, events.TaskUpdated, ws.ActionTaskUpdated) b.subscribe(eventBus, events.SessionWorkspaceSourcesUpdated, ws.ActionSessionWorkspaceSourcesUpdated) b.subscribe(eventBus, events.TaskDeleted, ws.ActionTaskDeleted) - b.subscribe(eventBus, events.TaskStateChanged, ws.ActionTaskStateChanged) + b.subscribeLifecycleStateEvents(eventBus) b.subscribe(eventBus, events.TaskPlanCreated, ws.ActionTaskPlanCreated) b.subscribe(eventBus, events.TaskPlanUpdated, ws.ActionTaskPlanUpdated) b.subscribe(eventBus, events.TaskPlanDeleted, ws.ActionTaskPlanDeleted) @@ -71,7 +71,6 @@ func RegisterTaskNotifications(ctx context.Context, eventBus bus.EventBus, hub * b.subscribe(eventBus, events.EnvironmentCreated, ws.ActionEnvironmentCreated) b.subscribe(eventBus, events.EnvironmentUpdated, ws.ActionEnvironmentUpdated) b.subscribe(eventBus, events.EnvironmentDeleted, ws.ActionEnvironmentDeleted) - b.subscribe(eventBus, events.TaskSessionStateChanged, ws.ActionSessionStateChanged) b.subscribe(eventBus, events.TaskSessionActivityChanged, ws.ActionSessionActivityChanged) b.subscribe(eventBus, events.MessageAdded, ws.ActionSessionMessageAdded) b.subscribe(eventBus, events.MessageUpdated, ws.ActionSessionMessageUpdated) @@ -105,85 +104,35 @@ func (b *TaskEventBroadcaster) Close() { } func (b *TaskEventBroadcaster) subscribe(eventBus bus.EventBus, subject, action string) { + b.subscribeWithResolver(eventBus, subject, func(*bus.Event) string { return action }) +} + +// subscribeLifecycleStateEvents uses one NATS subscription so task and session +// state notifications remain ordered for clients when the event bus is remote. +func (b *TaskEventBroadcaster) subscribeLifecycleStateEvents(eventBus bus.EventBus) { + b.subscribeWithResolver(eventBus, ">", func(event *bus.Event) string { + switch event.Type { + case events.TaskStateChanged: + return ws.ActionTaskStateChanged + case events.TaskSessionStateChanged: + return ws.ActionSessionStateChanged + default: + return "" + } + }) +} + +func (b *TaskEventBroadcaster) subscribeWithResolver( + eventBus bus.EventBus, + subject string, + resolveAction func(*bus.Event) string, +) { sub, err := eventBus.Subscribe(subject, func(ctx context.Context, event *bus.Event) error { - msg, err := ws.NewNotification(action, event.Data) - if err != nil { - b.logger.Error("failed to build websocket notification", zap.String("action", action), zap.Error(err)) + action := resolveAction(event) + if action == "" { return nil } - // Try to extract session_id from event data (works for both map and struct types) - var sessionID string - if data, ok := event.Data.(map[string]interface{}); ok { - sessionID, _ = data["session_id"].(string) - } else if data, ok := event.Data.(interface{ GetSessionID() string }); ok { - sessionID = data.GetSessionID() - } - - // Debug logging for session state changes with metadata - if action == ws.ActionSessionStateChanged { - if data, ok := event.Data.(map[string]interface{}); ok { - if metadata, ok := data["metadata"]; ok { - b.logger.Debug("received session.state_changed with metadata", - zap.String("action", action), - zap.String("session_id", sessionID), - zap.Any("metadata", metadata)) - } - } - } - if data, ok := event.Data.(map[string]interface{}); ok { - b.logLifecycleBroadcast(action, data, sessionID) - } - - workspaceID := extractWorkspaceID(event.Data) - switch action { - case ws.ActionWorkspaceCreated, ws.ActionWorkspaceUpdated, ws.ActionWorkspaceDeleted: - // Workspace event payloads are the workspace DTO itself: the - // workspace ID lives under "id", not "workspace_id". Without this - // the extract comes back empty and the event would broadcast to - // every user (caught by the auth E2E segregation spec). - if workspaceID == "" { - workspaceID = extractStringField(event.Data, "id") - } - b.hub.BroadcastToWorkspace(workspaceID, msg) - return nil - case ws.ActionSessionAgentctlStarting, ws.ActionSessionAgentctlReady, ws.ActionSessionAgentctlError: - if sessionID != "" { - b.hub.BroadcastToSession(sessionID, msg) - return nil - } - case ws.ActionSessionStateChanged: - // Broadcast beyond the session subscribers so the sidebar task - // switcher can track state changes for all tasks — but scoped to - // the owning workspace's user when auth is enabled. - b.hub.BroadcastToWorkspace(workspaceID, msg) - return nil - case ws.ActionSessionMessageAdded, ws.ActionSessionMessageUpdated, ws.ActionSessionMessageDeleted: - if sessionID != "" { - b.hub.BroadcastToSession(sessionID, msg) - return nil - } - case ws.ActionSessionWorkspaceSourcesUpdated: - if sessionID != "" { - b.hub.BroadcastToSession(sessionID, msg) - return nil - } - case ws.ActionMessageQueueStatusChanged: - if sessionID != "" { - b.hub.BroadcastToSession(sessionID, msg) - return nil - } - case ws.ActionExecutorPrepareProgress, ws.ActionExecutorPrepareCompleted: - // Broadcast to the owning workspace's clients so prepare - // progress/warnings are available when the user navigates to the - // session page after task creation. - b.hub.BroadcastToWorkspace(workspaceID, msg) - return nil - } - // Workspace-carrying events (task/workflow/repository/…) route to the - // owner's clients; events without workspace context (executors, - // environments, agent profiles — instance-wide resources) stay global. - b.hub.BroadcastToWorkspace(workspaceID, msg) - return nil + return b.broadcastEvent(ctx, event, action) }) if err != nil { b.logger.Error("failed to subscribe to events", zap.String("subject", subject), zap.Error(err)) @@ -192,6 +141,97 @@ func (b *TaskEventBroadcaster) subscribe(eventBus bus.EventBus, subject, action b.subscriptions = append(b.subscriptions, sub) } +func (b *TaskEventBroadcaster) broadcastEvent(ctx context.Context, event *bus.Event, action string) error { + msg, err := ws.NewNotification(action, event.Data) + if err != nil { + b.logger.Error("failed to build websocket notification", zap.String("action", action), zap.Error(err)) + return nil + } + sessionID := extractSessionID(event.Data) + b.logSessionStateMetadata(action, sessionID, event.Data) + if data, ok := event.Data.(map[string]interface{}); ok { + b.logLifecycleBroadcast(action, data, sessionID) + } + + return b.routeBroadcast(action, event.Data, sessionID, extractWorkspaceID(event.Data), msg) +} + +func (b *TaskEventBroadcaster) logSessionStateMetadata(action, sessionID string, data interface{}) { + if action != ws.ActionSessionStateChanged { + return + } + payload, ok := data.(map[string]interface{}) + if !ok { + return + } + metadata, ok := payload["metadata"] + if !ok { + return + } + b.logger.Debug("received session.state_changed with metadata", + zap.String("action", action), + zap.String("session_id", sessionID), + zap.Any("metadata", metadata)) +} + +func (b *TaskEventBroadcaster) routeBroadcast( + action string, + data interface{}, + sessionID string, + workspaceID string, + msg *ws.Message, +) error { + switch action { + case ws.ActionWorkspaceCreated, ws.ActionWorkspaceUpdated, ws.ActionWorkspaceDeleted: + // Workspace event payloads are the workspace DTO itself: the + // workspace ID lives under "id", not "workspace_id". Without this + // the extract comes back empty and the event would broadcast to + // every user (caught by the auth E2E segregation spec). + if workspaceID == "" { + workspaceID = extractStringField(data, "id") + } + b.hub.BroadcastToWorkspace(workspaceID, msg) + return nil + case ws.ActionSessionAgentctlStarting, ws.ActionSessionAgentctlReady, ws.ActionSessionAgentctlError: + if sessionID != "" { + b.hub.BroadcastToSession(sessionID, msg) + return nil + } + case ws.ActionSessionStateChanged: + // Broadcast beyond the session subscribers so the sidebar task + // switcher can track state changes for all tasks — but scoped to + // the owning workspace's user when auth is enabled. + b.hub.BroadcastToWorkspace(workspaceID, msg) + return nil + case ws.ActionSessionMessageAdded, ws.ActionSessionMessageUpdated, ws.ActionSessionMessageDeleted: + if sessionID != "" { + b.hub.BroadcastToSession(sessionID, msg) + return nil + } + case ws.ActionSessionWorkspaceSourcesUpdated: + if sessionID != "" { + b.hub.BroadcastToSession(sessionID, msg) + return nil + } + case ws.ActionMessageQueueStatusChanged: + if sessionID != "" { + b.hub.BroadcastToSession(sessionID, msg) + return nil + } + case ws.ActionExecutorPrepareProgress, ws.ActionExecutorPrepareCompleted: + // Broadcast to the owning workspace's clients so prepare + // progress/warnings are available when the user navigates to the + // session page after task creation. + b.hub.BroadcastToWorkspace(workspaceID, msg) + return nil + } + // Workspace-carrying events (task/workflow/repository/…) route to the + // owner's clients; events without workspace context (executors, + // environments, agent profiles — instance-wide resources) stay global. + b.hub.BroadcastToWorkspace(workspaceID, msg) + return nil +} + // extractWorkspaceID pulls a workspace ID from event payloads (map- or // struct-shaped). Empty means "no workspace context" — the event is treated // as instance-wide and broadcast to everyone. diff --git a/apps/backend/internal/gateway/websocket/task_notifications_test.go b/apps/backend/internal/gateway/websocket/task_notifications_test.go index cdf79066a0..9499bb0dda 100644 --- a/apps/backend/internal/gateway/websocket/task_notifications_test.go +++ b/apps/backend/internal/gateway/websocket/task_notifications_test.go @@ -8,6 +8,7 @@ import ( "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/events" "github.com/kandev/kandev/internal/events/bus" + ws "github.com/kandev/kandev/pkg/websocket" ) func testLogger() *logger.Logger { @@ -15,8 +16,76 @@ func testLogger() *logger.Logger { return log } +type subscriptionRecordingEventBus struct { + *bus.MemoryEventBus + subjects []string +} + +type recordingSubscription struct{} + +func (recordingSubscription) Unsubscribe() error { return nil } +func (recordingSubscription) IsValid() bool { return true } + +func (b *subscriptionRecordingEventBus) Subscribe(subject string, _ bus.EventHandler) (bus.Subscription, error) { + b.subjects = append(b.subjects, subject) + return recordingSubscription{}, nil +} + +func TestTaskEventBroadcaster_UsesOrderedWildcardForLifecycleStates(t *testing.T) { + log := testLogger() + eventBus := &subscriptionRecordingEventBus{MemoryEventBus: bus.NewMemoryEventBus(log)} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + hub := NewHub(nil, log) + b := RegisterTaskNotifications(ctx, eventBus, hub, log) + + seenWildcard := false + for _, subject := range eventBus.subjects { + if subject == ">" { + seenWildcard = true + } + if subject == events.TaskStateChanged || subject == events.TaskSessionStateChanged { + t.Fatalf("lifecycle state subject %q must use the ordered wildcard subscription", subject) + } + } + if !seenWildcard { + t.Fatal("lifecycle state notifications must share a NATS-style wildcard subscription") + } + + _ = b +} + +func TestTaskEventBroadcaster_OrdersLifecycleStateNotifications(t *testing.T) { + log := testLogger() + eventBus := bus.NewMemoryEventBus(log) + hub := NewHub(nil, log) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + _ = RegisterTaskNotifications(ctx, eventBus, hub, log) + + workspaceID := "workspace-1" + _ = eventBus.Publish(ctx, events.TaskStateChanged, bus.NewEvent( + events.TaskStateChanged, + "test", + map[string]interface{}{"task_id": "task-1", "workspace_id": workspaceID, "state": "IN_PROGRESS"}, + )) + _ = eventBus.Publish(ctx, events.TaskSessionStateChanged, bus.NewEvent( + events.TaskSessionStateChanged, + "test", + map[string]interface{}{"task_id": "task-1", "session_id": "session-1", "workspace_id": workspaceID, "new_state": "RUNNING"}, + )) + + first := <-hub.broadcast + second := <-hub.broadcast + if first.Action != ws.ActionTaskStateChanged || second.Action != ws.ActionSessionStateChanged { + t.Fatalf("lifecycle notification order = %q, %q; want task state then session state", first.Action, second.Action) + } +} + // TestTaskEventBroadcaster_NoDuplicateSubscriptions verifies that -// RegisterTaskNotifications creates exactly one subscription per event type. +// RegisterTaskNotifications creates one subscription per routed subject (with +// lifecycle state events intentionally sharing one ordered wildcard). // // The old code had a second subscription system (subscribeEventBusHandlers in // cmd/kandev/helpers.go) that subscribed to the same four events, causing @@ -33,13 +102,12 @@ func TestTaskEventBroadcaster_NoDuplicateSubscriptions(t *testing.T) { b := RegisterTaskNotifications(ctx, eventBus, hub, log) - // Count how many b.subscribe() calls are in RegisterTaskNotifications. - // Each call creates exactly one subscription. If any event is subscribed - // twice, this count will increase and the test will fail. + // Count how many subscriptions RegisterTaskNotifications creates. If any + // event is subscribed twice, this count will increase and the test will fail. // // Update this number when adding or removing event subscriptions in // RegisterTaskNotifications — it is intentionally exact. - const wantSubscriptions = 61 + const wantSubscriptions = 60 if got := len(b.subscriptions); got != wantSubscriptions { t.Errorf("RegisterTaskNotifications created %d subscriptions, want %d — "+ "did an event get subscribed twice?", got, wantSubscriptions) diff --git a/apps/backend/internal/orchestrator/event_handlers_streaming.go b/apps/backend/internal/orchestrator/event_handlers_streaming.go index 66b276437a..4541085306 100644 --- a/apps/backend/internal/orchestrator/event_handlers_streaming.go +++ b/apps/backend/internal/orchestrator/event_handlers_streaming.go @@ -1652,7 +1652,18 @@ func (s *Service) setSessionRunningForExecution(ctx context.Context, taskID, ses // (2,000+ redundant writes observed on long-running turns). wasAlreadyRunning := session.State == models.TaskSessionStateRunning - if updatedSession := s.updateTaskSessionState(ctx, taskID, sessionID, models.TaskSessionStateRunning, "", true, session); updatedSession != nil { + if updatedSession, _ := s.updateTaskSessionStateWithHook( + ctx, + taskID, + sessionID, + models.TaskSessionStateRunning, + "", + true, + func() { + s.reconcileRunningTaskStateLocked(ctx, taskID, sessionID) + }, + session, + ); updatedSession != nil { if len(preloadedSession) > 0 && preloadedSession[0] != nil && preloadedSession[0] != updatedSession { *preloadedSession[0] = *updatedSession } @@ -1661,7 +1672,10 @@ func (s *Service) setSessionRunningForExecution(ctx context.Context, taskID, ses if wasAlreadyRunning { return } +} +// reconcileRunningTaskStateLocked requires taskRuntimeStateMu to be held. +func (s *Service) reconcileRunningTaskStateLocked(ctx context.Context, taskID, sessionID string) { if err := s.reconcileTaskStateForRuntimeLocked( ctx, taskID, diff --git a/apps/backend/internal/orchestrator/event_handlers_streaming_test.go b/apps/backend/internal/orchestrator/event_handlers_streaming_test.go index 6c996c863f..c5b4824a1b 100644 --- a/apps/backend/internal/orchestrator/event_handlers_streaming_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_streaming_test.go @@ -33,6 +33,84 @@ type recordedEvent struct { event *bus.Event } +// taskServiceStateRepository exercises runtime task-state reconciliation +// through the same task-service publication path used in production. +type taskServiceStateRepository struct { + repo *sqliterepo.Repository + service *taskservice.Service +} + +func newTaskServiceStateRepository( + repo *sqliterepo.Repository, + eventBus bus.EventBus, +) *taskServiceStateRepository { + service := taskservice.NewService(taskservice.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, + }, eventBus, testLogger(), taskservice.RepositoryDiscoveryConfig{}) + return &taskServiceStateRepository{repo: repo, service: service} +} + +func (r *taskServiceStateRepository) GetTask(ctx context.Context, taskID string) (*v1.Task, error) { + task, err := r.repo.GetTask(ctx, taskID) + if err != nil { + return nil, err + } + return task.ToAPI(), nil +} + +func (r *taskServiceStateRepository) UpdateTaskState( + ctx context.Context, + taskID string, + state v1.TaskState, +) error { + _, err := r.service.UpdateTaskState(ctx, taskID, state) + return err +} + +func (r *taskServiceStateRepository) UpdateTaskStateIfCurrentIn( + ctx context.Context, + taskID string, + state v1.TaskState, + allowed []v1.TaskState, +) (bool, error) { + return r.service.UpdateTaskStateIfCurrentIn(ctx, taskID, state, allowed) +} + +func (r *taskServiceStateRepository) UpdateTaskStateIfNotArchived( + ctx context.Context, + taskID string, + state v1.TaskState, +) (bool, error) { + return r.service.UpdateTaskStateIfNotArchived(ctx, taskID, state) +} + +func (r *taskServiceStateRepository) UpdateTaskStateIfSessionState( + ctx context.Context, + taskID, sessionID string, + expectedSessionState models.TaskSessionState, + state v1.TaskState, +) (bool, error) { + return r.service.UpdateTaskStateIfSessionState( + ctx, + taskID, + sessionID, + expectedSessionState, + state, + ) +} + type recordingClarificationCanceller struct { sessions []string } @@ -1852,6 +1930,37 @@ func TestSetSessionWaitingForInput_NoRedundantTaskWrites(t *testing.T) { // TestSetSessionRunning_WritesOnTransition guards against an over-eager dedup // regression: when the session was NOT already RUNNING, the task state write // MUST still happen. +func TestSetSessionRunning_PublishesTaskStateBeforeRunningSession(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + require.NoError(t, repo.UpdateTaskState(ctx, "t1", v1.TaskStateReview)) + require.NoError(t, repo.UpdateTaskSessionState( + ctx, + "s1", + models.TaskSessionStateWaitingForInput, + "", + )) + + eventBus := &recordingEventBus{} + taskRepo := newTaskServiceStateRepository(repo, eventBus) + svc := &Service{ + logger: testLogger(), + eventBus: eventBus, + repo: repo, + taskRepo: taskRepo, + } + + svc.setSessionRunning(ctx, "t1", "s1") + + task, err := repo.GetTask(ctx, "t1") + require.NoError(t, err) + require.Equal(t, v1.TaskStateInProgress, task.State) + require.Len(t, eventBus.events, 2) + require.Equal(t, events.TaskStateChanged, eventBus.events[0].subject) + require.Equal(t, events.TaskSessionStateChanged, eventBus.events[1].subject) +} + func TestSetSessionRunning_WritesOnTransition(t *testing.T) { ctx := context.Background() repo := setupTestRepo(t) diff --git a/docs/decisions/2026-07-30-runtime-task-state-before-running-event.md b/docs/decisions/2026-07-30-runtime-task-state-before-running-event.md new file mode 100644 index 0000000000..14afdddc68 --- /dev/null +++ b/docs/decisions/2026-07-30-runtime-task-state-before-running-event.md @@ -0,0 +1,67 @@ +# ADR-2026-07-30-runtime-task-state-before-running-event: Publish Task State Before Running Session State + +**Status:** accepted +**Date:** 2026-07-30 +**Area:** backend, frontend, protocol, workflow + +## Context + +Task surfaces intentionally use different lifecycle facts: workflow position, +persisted `tasks.state`, and live `task_sessions.state`. During runtime start, +the orchestrator persisted and published `RUNNING` before reconciling the task +to `IN_PROGRESS`. A client could therefore render a truthful running indicator +inside a stale `Review` state group until the later task event arrived. + +## Decision + +When an owning non-Office session changes to `RUNNING`, the orchestrator +reconciles `tasks.state` through the existing session-state-guarded task update +before publishing `session.state_changed(RUNNING)`. + +The reconciliation runs only after the session compare-and-set succeeds and +before its event publication. If it changes the task, the existing +`task.state_changed` event is therefore published first. Repeated stream events +for an already-`RUNNING` session retain the existing no-write fast path. +Archive, terminal-session, clarification, cancellation, and Office guards +remain authoritative. The executor-success reconciliation remains as a healing +fallback. + +The WebSocket gateway subscribes to both lifecycle subjects through one +ordered NATS-style wildcard subscription and resolves the WebSocket action +from the event type. This preserves the producer's ordering contract across a +remote event bus, where separate subscriptions could otherwise race at the +gateway. + +## Consequences + +Clients can continue treating persisted task state as authoritative for State +grouping without synthesizing task state from session events. Desktop and +mobile task surfaces converge through their existing shared store and row +rendering. + +The task event may now precede the running session event by a small interval, +so a client may briefly show an `IN_PROGRESS` task without a spinner. That is +preferable to claiming active runtime inside `Review`, and the following +session event completes the display. A task-state persistence error cannot hide +a truly running session; the error is logged and the later executor +reconciliation may heal it. + +## Alternatives Considered + +- **Group the sidebar by live session activity.** Rejected because Group by + State deliberately exposes persisted task states such as `REVIEW`, + `COMPLETED`, and `FAILED`; collapsing those into activity buckets changes the + feature's meaning. +- **Patch `tasks.state` optimistically in the frontend on every running session + event.** Rejected because the frontend lacks the backend's Office, archive, + sibling-session, and race guards and would create a second task-state owner. +- **Publish the session event first and rely on eventual task reconciliation.** + Rejected because it is the observed inconsistency and makes correct UI depend + on event latency. +- **Keep separate gateway subscriptions for the two lifecycle subjects.** + Rejected because NATS delivers each subscription independently; their + callbacks can race and reverse the ordered events before WebSocket delivery. +- **Remove the no-write fast path and reconcile on every stream event.** + Rejected because long turns generate thousands of duplicate events; the fix + can preserve deduplication by reconciling only inside the successful + session-transition hook. diff --git a/docs/decisions/INDEX.md b/docs/decisions/INDEX.md index 42c376b029..54c1d90fce 100644 --- a/docs/decisions/INDEX.md +++ b/docs/decisions/INDEX.md @@ -90,3 +90,4 @@ Read individual ADRs for full context. Create new ones via `/record decision` or | 2026-07-29-quarantine-retention-override | [Make Quarantine Retention Overridable but Visible](2026-07-29-quarantine-retention-override.md) | accepted | backend, frontend | 2026-07-29 | | 2026-07-30-embedded-editor-executor-capabilities | [Derive Embedded Editor Availability from the Active Executor](2026-07-30-embedded-editor-executor-capabilities.md) | accepted | backend, frontend, protocol | 2026-07-30 | | 2026-07-30-session-owned-mcp-observability | [Keep MCP Attachment Evidence Session Owned](2026-07-30-session-owned-mcp-observability.md) | accepted | backend, frontend, protocol, security | 2026-07-30 | +| 2026-07-30-runtime-task-state-before-running-event | [Publish Task State Before Running Session State](2026-07-30-runtime-task-state-before-running-event.md) | accepted | backend, frontend, protocol, workflow | 2026-07-30 | diff --git a/docs/plans/runtime-task-state-publication-order/plan.md b/docs/plans/runtime-task-state-publication-order/plan.md new file mode 100644 index 0000000000..7c064fd347 --- /dev/null +++ b/docs/plans/runtime-task-state-publication-order/plan.md @@ -0,0 +1,143 @@ +--- +spec: docs/specs/tasks/runtime-state-publication-order.md +created: 2026-07-30 +status: complete +--- + +# Implementation Plan: Runtime Task-State Publication Order + +## Overview + +Close the confirmed event-ordering window without changing task-state meaning +or frontend grouping. The orchestrator will reconcile a successfully +transitioned session's owning task inside the existing pre-publication hook, +then publish the running session event. Focused Go tests will prove the event +order, persistence result, race guards, and no-redundant-write behavior. + +The confirmed root cause is that `setSessionRunningForExecution` currently +publishes `session.state_changed(RUNNING)` from `updateTaskSessionState` and only +then calls `reconcileTaskStateForRuntimeLocked`. The sidebar immediately paints +the running session indicator while Group by State still reads persisted +`tasks.state = REVIEW`. + +The remote-delivery path also requires the gateway to preserve this ordering: +task and session lifecycle notifications share one NATS-style subscription so +separate callback scheduling cannot reintroduce the race before WebSocket +delivery. + +## Backend + +### Runtime transition ordering + +- Update `setSessionRunningForExecution` in + `apps/backend/internal/orchestrator/event_handlers_streaming.go` to use + `updateTaskSessionStateWithHook`. +- Run `reconcileTaskStateForRuntimeLocked` from the hook after the session + compare-and-set succeeds but before `publishTaskSessionStateChanged`. +- Capture and log reconciliation errors without rolling back or suppressing the + truthful session transition. +- Remove the post-publication reconciliation call for the changed-state path. +- Preserve the `wasAlreadyRunning` fast path so repeated tool/stream events do + not perform task-state reads or writes. +- Keep `writeTaskInProgressForRuntime` and the executor-success callback as the + eventual healing path. + +### Gateway delivery ordering + +- Subscribe the WebSocket task broadcaster to one NATS-style wildcard for task + and session state events, resolving the WebSocket action from `event.Type`. +- Keep all other event subscriptions and routing behavior unchanged. +- Ensure the in-memory event bus implements the same `>` wildcard semantics so + local development and regression tests exercise the production contract. + +### Event and persistence contract + +- Continue using `UpdateTaskStateIfSessionState` so the session state, archive + status, terminal transitions, clarification/cancellation races, and Office + exclusion remain guarded by existing code. +- Do not add a schema, API field, WebSocket event, frontend-derived task state, + or new in-memory lifecycle cache. + +## Frontend + +No frontend production change is planned. Group by State must continue to use +persisted `task.state`, while the shared desktop/mobile task row continues to +render live session activity. Correct backend event order makes those existing +contracts converge. + +Mobile parity is state-only: the desktop sidebar and phone task drawer use the +same sidebar task aggregation and `TaskItem` rendering. There is no composition, +navigation, scroll, safe-area, pointer, or touch-target change. + +## Tests + +- **What:** a real `WAITING_FOR_INPUT -> RUNNING` transition persists + `tasks.state = IN_PROGRESS` before observers receive the running session + event. + **File:** + `apps/backend/internal/orchestrator/event_handlers_streaming_test.go`. + **How:** use the SQLite test repository, the real task service, and one + recording/memory event bus; subscribe to task/session state events and assert + both durable state and event order. +- **What:** the regression test fails on the current implementation because + `session.state_changed(RUNNING)` is observed while the task is still + `REVIEW`. + **File:** + `apps/backend/internal/orchestrator/event_handlers_streaming_test.go`. + **How:** record task and session state events through the real task service + rather than relying on sleeps or browser rendering. +- **What:** same-state stream churn remains deduplicated. + **File:** + `apps/backend/internal/orchestrator/event_handlers_streaming_test.go`. + **How:** retain and run `TestSetSessionRunning_NoRedundantTaskWrites`; extend + it only if needed to assert no state event. +- **What:** clarification/cancellation and archive races cannot be overwritten. + **Files:** + `apps/backend/internal/orchestrator/event_handlers_runtime_state_race_test.go` + and `apps/backend/internal/task/repository/sqlite/task_state_cas_test.go`. + **How:** run the existing guarded-CAS regression tests alongside the new + ordering test, including `-race`. +- **What:** gateway delivery keeps task-state notification ahead of the + running-session notification for a shared wildcard subscription. + **Files:** `apps/backend/internal/gateway/websocket/task_notifications.go`, + `apps/backend/internal/gateway/websocket/task_notifications_test.go`, and + `apps/backend/internal/events/bus/memory_test.go`. + **How:** assert lifecycle subjects use one wildcard, publish both events + through the in-memory NATS-compatible bus, and assert WebSocket action order. + +## E2E Tests + +No permanent browser test is planned. The repaired behavior is the absence of +an intermediate render between two WebSocket events; a Playwright +`MutationObserver` or polling assertion would be scheduler-dependent and could +pass on the broken implementation when React batches both events. The +producer-level integration test deterministically observes the same contract +at the event boundary. + +Existing frontend tests already pin the unchanged display semantics: + +- `apps/web/lib/sidebar/apply-view.test.ts` proves State grouping uses persisted + task state. +- `apps/web/components/task/task-item.test.tsx` proves running session activity + selects the spinner. +- `apps/web/e2e/tests/task/mobile-sidebar-workflow-completion-icon.spec.ts` + proves the phone drawer uses the shared task-row status rendering. + +## Implementation Waves And Parallel Candidates + +Wave 1: + +- [ ] [Task 01: Order runtime state publication](task-01-order-runtime-state-publication.md) + +The task is sequential because production ordering and its regression test +touch the same orchestrator lifecycle seam. No subagent authorization is +implied. + +## Risks + +- Publishing task state first changes event order for clients that observe both + subjects. Payloads and final states are unchanged. +- Reconciliation errors must not suppress a truthful running-session event. +- The pre-publication hook runs while the task runtime-state mutex is held; + implementation must call the existing locked reconciliation helper and must + not reacquire that mutex. diff --git a/docs/plans/runtime-task-state-publication-order/task-01-order-runtime-state-publication.md b/docs/plans/runtime-task-state-publication-order/task-01-order-runtime-state-publication.md new file mode 100644 index 0000000000..cd1b9a86ff --- /dev/null +++ b/docs/plans/runtime-task-state-publication-order/task-01-order-runtime-state-publication.md @@ -0,0 +1,78 @@ +--- +id: "01-order-runtime-state-publication" +title: "Order runtime state publication" +status: completed +wave: 1 +depends_on: [] +plan: "plan.md" +spec: "../../specs/tasks/runtime-state-publication-order.md" +--- + +# Task 01: Order runtime state publication + +## Acceptance + +- A successful `WAITING_FOR_INPUT -> RUNNING` session transition reconciles an + eligible task to `IN_PROGRESS` before publishing + `session.state_changed(RUNNING)`. +- Existing archive, Office, clarification/cancellation, terminal-state, and + session-state CAS guards remain intact; reconciliation failure does not hide + the running session. +- Already-`RUNNING` stream churn performs no redundant task-state writes or + state events, and the executor-success reconciliation remains available. +- The WebSocket gateway forwards task and session lifecycle notifications from + one ordered wildcard subscription, preserving their order for remote event + buses. + +## Verification + +```bash +cd apps/backend && go test \ + -run 'TestSetSessionRunning_(PublishesTaskStateBeforeRunningSession|NoRedundantTaskWrites|WritesOnTransition)' \ + ./internal/orchestrator +cd apps/backend && go test -race \ + -run 'Test(SetSessionRunning_PublishesTaskStateBeforeRunningSession|ReconcileTaskStateForRuntime_)' \ + ./internal/orchestrator +cd apps/backend && go test \ + -run 'TestUpdateTaskStateIfSessionState_' \ + ./internal/task/repository/sqlite +cd apps/backend && go test \ + -run 'TestTaskEventBroadcaster_(UsesOrderedWildcardForLifecycleStates|OrdersLifecycleStateNotifications|NoDuplicateSubscriptions|PreservesAllFields)' \ + ./internal/gateway/websocket +cd apps/backend && go test \ + -run TestMemoryEventBus_MultiTokenWildcard \ + ./internal/events/bus +``` + +## Files likely touched + +- `apps/backend/internal/orchestrator/event_handlers_streaming.go` +- `apps/backend/internal/orchestrator/event_handlers_streaming_test.go` +- `apps/backend/internal/gateway/websocket/task_notifications.go` +- `apps/backend/internal/gateway/websocket/task_notifications_test.go` +- `apps/backend/internal/events/bus/memory.go` +- `apps/backend/internal/events/bus/memory_test.go` + +## Dependencies + +None. + +## Parallelism + +Sequential. The production ordering and its regression test share the same +orchestrator lifecycle seam. + +## Inputs + +- Spec `What`, `Failure modes`, and `Scenarios`. +- ADR + `docs/decisions/2026-07-30-runtime-task-state-before-running-event.md`. +- Existing `updateTaskSessionStateWithHook` pre-publication seam. +- Existing `reconcileTaskStateForRuntimeLocked` guarded task-state writer. +- Existing no-redundant-write and runtime-state race tests. + +## Output contract + +Report the RED failure reason, GREEN results, files changed, exact commands and +results, remaining risks, and the task/plan status updates. Do not change +frontend grouping or synthesize task state from session events. diff --git a/docs/specs/INDEX.md b/docs/specs/INDEX.md index b65bdd9606..c538198f7f 100644 --- a/docs/specs/INDEX.md +++ b/docs/specs/INDEX.md @@ -69,6 +69,7 @@ Kandev's task model: documents, execution stages, labels, blocker escalation, su | [parent-child-message-interrupt](tasks/parent-child-message-interrupt.md) | shipped | | [parent-child-task-stop](tasks/parent-child-task-stop.md) | shipped | | [mcp-task-agent-profile-default](tasks/mcp-task-agent-profile-default/spec.md) | shipped | +| [runtime-state-publication-order](tasks/runtime-state-publication-order.md) | shipped | | [explicit-completion-signal](workflow/explicit-completion-signal/spec.md) | shipped | ## agents/ — agent governance diff --git a/docs/specs/tasks/runtime-state-publication-order.md b/docs/specs/tasks/runtime-state-publication-order.md new file mode 100644 index 0000000000..44ea899b71 --- /dev/null +++ b/docs/specs/tasks/runtime-state-publication-order.md @@ -0,0 +1,105 @@ +--- +status: shipped +created: 2026-07-30 +owner: kandev +decision: docs/decisions/2026-07-30-runtime-task-state-before-running-event.md +--- + +# Runtime Task-State Publication Order + +## Why + +Task surfaces combine the persisted task state with live session state. If a +session becomes `RUNNING` before its task is observably `IN_PROGRESS`, the +sidebar can briefly show a running spinner inside the `Review` state group even +though the task's workflow step and runtime are already active. + +## What + +- For a non-Office, unarchived task whose session transitions into `RUNNING`, + Kandev reconciles the persisted task state to `IN_PROGRESS` before publishing + that session's `session.state_changed` event. +- When reconciliation changes the task state, observers receive + `task.state_changed` before `session.state_changed` announces `RUNNING`. +- The WebSocket gateway consumes both lifecycle notifications through one + ordered NATS-style subscription, preserving that order when the event bus is + remote; the in-memory event bus supports the same wildcard semantics. +- The task-state write remains guarded by the owning session's authoritative + state. A concurrent clarification, cancellation, terminal transition, or + archive prevents a stale runtime writer from promoting the task. +- Repeated stream activity for a session already in `RUNNING` does not cause + repeated task-state writes or events. +- The executor's post-start task-state reconciliation remains an eventual + recovery path when the earlier reconciliation fails or is skipped. +- Existing sidebar state grouping, workflow-step rendering, status icons, and + desktop/mobile composition remain unchanged. + +## Data model + +This repair uses the existing persisted fields: + +- `tasks.state`, including `REVIEW` and `IN_PROGRESS` +- `task_sessions.state`, including `WAITING_FOR_INPUT`, `STARTING`, and + `RUNNING` + +No schema, enum, or persistence-format changes are introduced. + +## API surface + +The existing WebSocket event names and payloads remain unchanged: + +- `task.state_changed` +- `session.state_changed` + +The ordering contract is defined by +[ADR-2026-07-30-runtime-task-state-before-running-event](../../decisions/2026-07-30-runtime-task-state-before-running-event.md). + +## State machine + +- A workflow transition may persist the target workflow step and use `REVIEW` + as a safe intermediate task state while no session owns active work. +- When a prompt or runtime transition successfully changes an owning session to + `RUNNING`, the same lifecycle operation reconciles the task to + `IN_PROGRESS` before exposing the new session state to clients. +- When the turn settles and no sibling session is working, existing behavior + returns the task to `REVIEW`. + +## Failure modes + +- If guarded task reconciliation loses to a newer session/task transition, it + does not overwrite the winner. +- If task reconciliation returns an error after the session transition is + persisted, Kandev logs the error and still reports the truthful session + state. The existing executor-success reconciliation remains available to + heal the task state. +- Office task status remains controlled by the Office lifecycle and is not + promoted by this runtime reconciliation. + +## Scenarios + +- **GIVEN** a non-Office task in `REVIEW` with an owning session in + `WAITING_FOR_INPUT`, **WHEN** the session accepts a new prompt and transitions + to `RUNNING`, **THEN** `tasks.state` is `IN_PROGRESS` and + `task.state_changed` is published before `session.state_changed(RUNNING)`. +- **GIVEN** the sidebar is grouped by State, **WHEN** it observes a session + transition to `RUNNING`, **THEN** the task is not rendered with a running + spinner inside the `Review` group. +- **GIVEN** a remote event bus delivers the two lifecycle events, **WHEN** the + gateway forwards them to WebSocket clients, **THEN** the task-state + notification is delivered before the running-session notification. +- **GIVEN** a session and task already in `RUNNING`/`IN_PROGRESS`, **WHEN** + additional tool or stream events arrive, **THEN** no redundant task-state + write or state-change event is produced. +- **GIVEN** a clarification or cancellation changes the session before guarded + task reconciliation commits, **WHEN** the stale runtime transition resumes, + **THEN** the task is not promoted to `IN_PROGRESS`. +- **GIVEN** an archived or Office task, **WHEN** a session reports runtime + activity, **THEN** this repair does not rewrite its task state. + +## Out of scope + +- Changing how sidebar State groups are derived or labeled. +- Inferring persisted task state in the frontend from session events. +- Changing workflow-step transitions, session-state enums, WebSocket payloads, + Office task-state ownership, or task status icons. +- Adding new desktop or mobile layout, navigation, or touch behavior.