fix(orchestrator): order runtime task state events (#2082)

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 + "$"

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()

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.

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)

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,

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)

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.

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 |

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.

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.

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

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.