From 6734b64b54c76f8e46aa5de869d237f7886e2db2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20Flor=C3=AAncio?= Date: Mon, 1 Jun 2026 10:35:07 +0100 Subject: [PATCH] fix(e2e): ignore stale session state snapshots blocking idle input (#1208) --- apps/backend/cmd/kandev/helpers.go | 11 +- apps/backend/cmd/kandev/helpers_test.go | 45 ++++++ .../backend/internal/mcp/handlers/handlers.go | 24 ++++ .../internal/mcp/handlers/handlers_test.go | 84 +++++++++++ .../internal/office/testharness/routes.go | 1 + apps/backend/internal/orchestrator/dto/dto.go | 1 + .../orchestrator/event_handlers_streaming.go | 134 +++++++++++------- .../event_handlers_streaming_test.go | 20 +++ .../orchestrator/event_handlers_transient.go | 9 +- .../orchestrator/event_handlers_workflow.go | 18 ++- .../event_handlers_workflow_test.go | 38 +++++ .../internal/orchestrator/task_operations.go | 4 + .../orchestrator/task_operations_test.go | 3 + apps/web/e2e/helpers/session.ts | 2 +- apps/web/e2e/pages/session-page.ts | 56 +++++--- apps/web/e2e/tests/search/shared.ts | 4 +- .../tests/session/executor-not-found.spec.ts | 4 +- .../session/session-resume-commands.spec.ts | 6 +- .../session/session-resume-dedup.spec.ts | 31 +--- .../e2e/tests/session/session-resume.spec.ts | 12 +- .../session/use-session-resumption.test.ts | 46 ++++++ .../domains/session/use-session-resumption.ts | 3 +- apps/web/lib/types/backend.ts | 2 + .../ws/handlers/agent-session-pure.test.ts | 31 ++++ .../web/lib/ws/handlers/agent-session.test.ts | 35 +++++ apps/web/lib/ws/handlers/agent-session.ts | 29 +++- 26 files changed, 523 insertions(+), 130 deletions(-) diff --git a/apps/backend/cmd/kandev/helpers.go b/apps/backend/cmd/kandev/helpers.go index 3d3b55acf9..48301efdb9 100644 --- a/apps/backend/cmd/kandev/helpers.go +++ b/apps/backend/cmd/kandev/helpers.go @@ -100,6 +100,9 @@ func buildSessionDataProvider(taskRepo *sqliterepo.Repository, lifecycleMgr *lif } const sessionIDPayloadKey = "session_id" +const taskIDPayloadKey = "task_id" +const newStatePayloadKey = "new_state" +const sessionUpdatedAtPayloadKey = "updated_at" // appendAgentctlStatusMessage snapshots the current agentctl readiness for a // session so late-subscribing clients (page reload, task switch, WS reconnect) @@ -158,9 +161,10 @@ func appendAgentctlStatusMessage( // — without it, env-routed shell terminals stall on "Connecting terminal...". func appendSessionStateMessage(sessionID string, session *models.TaskSession, result []*ws.Message) []*ws.Message { payload := map[string]interface{}{ - "session_id": sessionID, - "task_id": session.TaskID, - "new_state": string(session.State), + sessionIDPayloadKey: sessionID, + taskIDPayloadKey: session.TaskID, + newStatePayloadKey: string(session.State), + sessionUpdatedAtPayloadKey: session.UpdatedAt.UTC().Format(time.RFC3339Nano), } if session.ReviewStatus != models.ReviewStatusNone { payload["review_status"] = string(session.ReviewStatus) @@ -342,7 +346,6 @@ func appendContextWindowMessage(sessionID string, session *models.TaskSession, r notification, err := ws.NewNotification(ws.ActionSessionStateChanged, map[string]interface{}{ "session_id": sessionID, "task_id": session.TaskID, - "new_state": string(session.State), "metadata": map[string]interface{}{ "context_window": contextWindow, }, diff --git a/apps/backend/cmd/kandev/helpers_test.go b/apps/backend/cmd/kandev/helpers_test.go index 492f901efb..2f556d318b 100644 --- a/apps/backend/cmd/kandev/helpers_test.go +++ b/apps/backend/cmd/kandev/helpers_test.go @@ -3,6 +3,7 @@ package main import ( "encoding/json" "testing" + "time" "github.com/kandev/kandev/internal/task/models" ws "github.com/kandev/kandev/pkg/websocket" @@ -52,6 +53,50 @@ func TestAppendSessionStateMessage_IncludesTaskEnvironmentID(t *testing.T) { } } +func TestAppendSessionStateMessage_IncludesUpdatedAt(t *testing.T) { + updatedAt := time.Date(2026, 1, 2, 12, 0, 0, 0, time.UTC) + session := &models.TaskSession{ + ID: "sess-3", + TaskID: "task-1", + State: models.TaskSessionStateWaitingForInput, + UpdatedAt: updatedAt, + } + + msgs := appendSessionStateMessage(session.ID, session, nil) + if len(msgs) != 1 { + t.Fatalf("expected 1 message, got %d", len(msgs)) + } + payload := decodePayload(t, msgs[0].Payload) + got, present := payload["updated_at"] + if !present { + t.Fatal("payload missing updated_at — stale subscribe snapshots cannot be ignored") + } + if got != updatedAt.Format(time.RFC3339Nano) { + t.Fatalf("expected updated_at=%q, got %v", updatedAt.Format(time.RFC3339Nano), got) + } +} + +func TestAppendContextWindowMessage_DoesNotEmitStateSnapshot(t *testing.T) { + session := &models.TaskSession{ + ID: "sess-4", + TaskID: "task-1", + State: models.TaskSessionStateRunning, + UpdatedAt: time.Date(2026, 1, 2, 12, 0, 0, 0, time.UTC), + Metadata: map[string]interface{}{ + "context_window": map[string]interface{}{"size": 100}, + }, + } + + msgs := appendContextWindowMessage(session.ID, session, nil) + if len(msgs) != 1 { + t.Fatalf("expected 1 message, got %d", len(msgs)) + } + payload := decodePayload(t, msgs[0].Payload) + if _, present := payload["new_state"]; present { + t.Fatal("context-window snapshot must not carry new_state and overwrite fresher session state") + } +} + // TestAppendSessionStateMessage_OmitsEmptyTaskEnvironmentID — sessions without // an environment (legacy rows, archived sessions) must not emit an empty // task_environment_id field that would clobber a populated frontend map. diff --git a/apps/backend/internal/mcp/handlers/handlers.go b/apps/backend/internal/mcp/handlers/handlers.go index a5468774b7..fc7cd05762 100644 --- a/apps/backend/internal/mcp/handlers/handlers.go +++ b/apps/backend/internal/mcp/handlers/handlers.go @@ -8,6 +8,7 @@ import ( "encoding/json" "errors" "fmt" + "time" "github.com/kandev/kandev/internal/agent/mcpconfig" agentsettingscontroller "github.com/kandev/kandev/internal/agent/settings/controller" @@ -1208,6 +1209,7 @@ func (h *Handlers) setSessionRunning(ctx context.Context, taskID, sessionID stri h.logger.Warn("failed to update session state to RUNNING", zap.String("session_id", sessionID), zap.Error(err)) + return } if taskID != "" { if err := h.taskRepo.UpdateTaskState(ctx, taskID, v1.TaskStateInProgress); err != nil { @@ -1224,6 +1226,13 @@ func (h *Handlers) setSessionRunning(ctx context.Context, taskID, sessionID stri "session_id": sessionID, "new_state": string(models.TaskSessionStateRunning), } + if updatedAt, ok := h.sessionUpdatedAtForStateEvent(ctx, sessionID); ok { + eventData["updated_at"] = updatedAt + } else { + h.logger.Warn("skipping session state_changed publish; could not load authoritative updated_at", + zap.String("session_id", sessionID)) + return + } _ = h.eventBus.Publish(ctx, events.TaskSessionStateChanged, bus.NewEvent( events.TaskSessionStateChanged, "mcp-handlers", @@ -1239,6 +1248,7 @@ func (h *Handlers) setSessionWaitingForInput(ctx context.Context, taskID, sessio h.logger.Warn("failed to update session state to WAITING_FOR_INPUT", zap.String("session_id", sessionID), zap.Error(err)) + return } // Update task state to REVIEW @@ -1257,6 +1267,13 @@ func (h *Handlers) setSessionWaitingForInput(ctx context.Context, taskID, sessio "session_id": sessionID, "new_state": string(models.TaskSessionStateWaitingForInput), } + if updatedAt, ok := h.sessionUpdatedAtForStateEvent(ctx, sessionID); ok { + eventData["updated_at"] = updatedAt + } else { + h.logger.Warn("skipping session state_changed publish; could not load authoritative updated_at", + zap.String("session_id", sessionID)) + return + } _ = h.eventBus.Publish(ctx, events.TaskSessionStateChanged, bus.NewEvent( events.TaskSessionStateChanged, "mcp-handlers", @@ -1265,6 +1282,13 @@ func (h *Handlers) setSessionWaitingForInput(ctx context.Context, taskID, sessio } } +func (h *Handlers) sessionUpdatedAtForStateEvent(ctx context.Context, sessionID string) (string, bool) { + if session, err := h.sessionRepo.GetTaskSession(ctx, sessionID); err == nil && session != nil && !session.UpdatedAt.IsZero() { + return session.UpdatedAt.UTC().Format(time.RFC3339Nano), true + } + return "", false +} + // handleCreateTaskPlan creates a new task plan. func (h *Handlers) handleCreateTaskPlan(ctx context.Context, msg *ws.Message) (*ws.Message, error) { var req struct { diff --git a/apps/backend/internal/mcp/handlers/handlers_test.go b/apps/backend/internal/mcp/handlers/handlers_test.go index a7c74d818b..033e916fea 100644 --- a/apps/backend/internal/mcp/handlers/handlers_test.go +++ b/apps/backend/internal/mcp/handlers/handlers_test.go @@ -64,6 +64,42 @@ func newTestTaskService(t *testing.T) (*service.Service, *sqliterepo.Repository) return svc, repo } +func seedMCPHandlerSession(t *testing.T, repo *sqliterepo.Repository, taskID, sessionID string, state models.TaskSessionState) { + t.Helper() + ctx := context.Background() + now := time.Now().UTC() + require.NoError(t, repo.CreateWorkspace(ctx, &models.Workspace{ + ID: "ws-state-event", + Name: "State Event", + CreatedAt: now, + UpdatedAt: now, + })) + require.NoError(t, repo.CreateTask(ctx, &models.Task{ + ID: taskID, + WorkspaceID: "ws-state-event", + Title: "State Event Task", + State: v1.TaskStateInProgress, + CreatedAt: now, + UpdatedAt: now, + })) + require.NoError(t, repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: sessionID, + TaskID: taskID, + State: state, + StartedAt: now, + UpdatedAt: now, + })) +} + +type mcpRecordingEventBus struct { + events []*bus.Event +} + +func (b *mcpRecordingEventBus) Publish(_ context.Context, _ string, event *bus.Event) error { + b.events = append(b.events, event) + return nil +} + func TestHandleCreateTask_MissingTitle(t *testing.T) { h := &Handlers{} msg := makeWSMessage(t, ws.ActionMCPCreateTask, map[string]interface{}{ @@ -76,6 +112,54 @@ func TestHandleCreateTask_MissingTitle(t *testing.T) { assertWSError(t, resp, ws.ErrorCodeValidation) } +func TestSessionStateEventsIncludeUpdatedAt(t *testing.T) { + tests := []struct { + name string + initial models.TaskSessionState + run func(*Handlers, context.Context, string, string) + wantState models.TaskSessionState + }{ + { + name: "running", + initial: models.TaskSessionStateWaitingForInput, + run: (*Handlers).setSessionRunning, + wantState: models.TaskSessionStateRunning, + }, + { + name: "waiting for input", + initial: models.TaskSessionStateRunning, + run: (*Handlers).setSessionWaitingForInput, + wantState: models.TaskSessionStateWaitingForInput, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, repo := newTestTaskService(t) + seedMCPHandlerSession(t, repo, "task-state-event", "session-state-event", tt.initial) + eventBus := &mcpRecordingEventBus{} + h := &Handlers{ + sessionRepo: repo, + taskRepo: repo, + eventBus: eventBus, + logger: testLogger(t).WithFields(), + } + + tt.run(h, context.Background(), "task-state-event", "session-state-event") + + require.Len(t, eventBus.events, 1) + data, ok := eventBus.events[0].Data.(map[string]interface{}) + require.True(t, ok) + assert.Equal(t, string(tt.wantState), data["new_state"]) + gotUpdatedAt, ok := data["updated_at"].(string) + require.True(t, ok, "expected updated_at for state ordering") + updatedSession, err := repo.GetTaskSession(context.Background(), "session-state-event") + require.NoError(t, err) + assert.Equal(t, updatedSession.UpdatedAt.UTC().Format(time.RFC3339Nano), gotUpdatedAt) + }) + } +} + func TestHandleCreateTask_SubtaskMissingDescription(t *testing.T) { h := &Handlers{} msg := makeWSMessage(t, ws.ActionMCPCreateTask, map[string]interface{}{ diff --git a/apps/backend/internal/office/testharness/routes.go b/apps/backend/internal/office/testharness/routes.go index bfbf090d3e..1f4b6b0e7b 100644 --- a/apps/backend/internal/office/testharness/routes.go +++ b/apps/backend/internal/office/testharness/routes.go @@ -387,6 +387,7 @@ func publishSessionStateChanged(ctx context.Context, eventBus bus.EventBus, sess "session_id": session.ID, "old_state": "", "new_state": string(session.State), + "updated_at": session.UpdatedAt.UTC().Format(time.RFC3339Nano), "agent_profile_id": session.AgentProfileID, } if session.Metadata != nil { diff --git a/apps/backend/internal/orchestrator/dto/dto.go b/apps/backend/internal/orchestrator/dto/dto.go index 8c97426d0c..d93f5465d9 100644 --- a/apps/backend/internal/orchestrator/dto/dto.go +++ b/apps/backend/internal/orchestrator/dto/dto.go @@ -35,6 +35,7 @@ type TaskSessionStatusResponse struct { SessionID string `json:"session_id"` TaskID string `json:"task_id"` State string `json:"state"` + UpdatedAt string `json:"updated_at,omitempty"` AgentProfileID string `json:"agent_profile_id,omitempty"` // Runtime status diff --git a/apps/backend/internal/orchestrator/event_handlers_streaming.go b/apps/backend/internal/orchestrator/event_handlers_streaming.go index 0ebec0f7d1..217c780156 100644 --- a/apps/backend/internal/orchestrator/event_handlers_streaming.go +++ b/apps/backend/internal/orchestrator/event_handlers_streaming.go @@ -376,7 +376,9 @@ func (s *Service) handleToolUpdateEvent(ctx context.Context, payload *lifecycle. // check, same-state check). This is an optimistic fast-path: between load and check another // goroutine may have changed the state in the DB. The guards are best-effort to avoid // unnecessary writes; the DB update via UpdateTaskSessionState is the atomic source of truth. -func (s *Service) updateTaskSessionState(ctx context.Context, taskID, sessionID string, nextState models.TaskSessionState, errorMessage string, allowWakeFromWaiting bool, preloadedSession ...*models.TaskSession) { +// Returns the session row after a successful write (refreshed from DB when possible); callers +// that need authoritative UpdatedAt should use the return value, not the preloaded input. +func (s *Service) updateTaskSessionState(ctx context.Context, taskID, sessionID string, nextState models.TaskSessionState, errorMessage string, allowWakeFromWaiting bool, preloadedSession ...*models.TaskSession) *models.TaskSession { var session *models.TaskSession if len(preloadedSession) > 0 && preloadedSession[0] != nil { session = preloadedSession[0] @@ -384,75 +386,97 @@ func (s *Service) updateTaskSessionState(ctx context.Context, taskID, sessionID var err error session, err = s.repo.GetTaskSession(ctx, sessionID) if err != nil { - return + return nil } } if session.State == models.TaskSessionStateWaitingForInput && nextState == models.TaskSessionStateRunning && !allowWakeFromWaiting { - return + return session } oldState := session.State switch session.State { case models.TaskSessionStateCompleted, models.TaskSessionStateFailed, models.TaskSessionStateCancelled: - return + return session } if session.State == nextState { - return + return session } if err := s.repo.UpdateTaskSessionState(ctx, sessionID, nextState, errorMessage); err != nil { s.logger.Error("failed to update task session state", zap.String("session_id", sessionID), zap.String("state", string(nextState)), zap.Error(err)) + return session } - s.logger.Debug("task session state updated", - zap.String("task_id", taskID), - zap.String("session_id", sessionID), - zap.String("old_state", string(oldState)), - zap.String("new_state", string(nextState))) - if s.eventBus != nil { - // Resolve agent_profile_id for the office per-agent live indicators on - // the frontend. Prefer the value stored on the session row (post - // office-task-session-lifecycle, populated for office sessions); fall - // back to the task's assignee for legacy rows that pre-date that spec. - // Best-effort: a missed lookup just omits the field, it doesn't drop - // the rest of the event. - agentProfileID := session.AgentProfileID - if agentProfileID == "" { - if task, terr := s.repo.GetTask(ctx, taskID); terr == nil && task != nil { - agentProfileID = task.AssigneeAgentProfileID - } + var authoritativeUpdatedAt *time.Time + if refreshed, err := s.repo.GetTaskSession(ctx, sessionID); err == nil && refreshed != nil { + session = refreshed + if !refreshed.UpdatedAt.IsZero() { + t := refreshed.UpdatedAt.UTC() + authoritativeUpdatedAt = &t } - eventData := map[string]interface{}{ - "task_id": taskID, - "session_id": sessionID, - "old_state": string(oldState), - "new_state": string(nextState), - "error_message": errorMessage, - "agent_profile_id": agentProfileID, - "agent_profile_snapshot": session.AgentProfileSnapshot, - "is_passthrough": session.IsPassthrough, - } - // Include review_status if present to ensure frontend state consistency - if session.ReviewStatus != models.ReviewStatusNone { - eventData["review_status"] = string(session.ReviewStatus) - } - // Include session metadata (e.g. plan_mode set by workflow events). - // Key is "session_metadata" to avoid conflict with message-level "metadata". - if len(session.Metadata) > 0 { - eventData["session_metadata"] = session.Metadata - } - // Propagate toast suppression flag set by failure handlers (e.g. missing branch). - if suppressed, ok := s.suppressToast.LoadAndDelete(sessionID); ok && suppressed.(bool) { - eventData["suppress_toast"] = true - } - if session.TaskEnvironmentID != "" { - eventData["task_environment_id"] = session.TaskEnvironmentID - } - _ = s.eventBus.Publish(ctx, events.TaskSessionStateChanged, bus.NewEvent(events.TaskSessionStateChanged, "task-session", eventData)) + } + if authoritativeUpdatedAt == nil { + s.logger.Warn("skipping session state_changed publish; could not read authoritative updated_at", + zap.String("task_id", taskID), + zap.String("session_id", sessionID), + zap.String("new_state", string(nextState))) + } else { + s.logger.Debug("task session state updated", + zap.String("task_id", taskID), + zap.String("session_id", sessionID), + zap.String("old_state", string(oldState)), + zap.String("new_state", string(nextState))) + s.publishTaskSessionStateChanged(ctx, taskID, sessionID, oldState, nextState, errorMessage, authoritativeUpdatedAt, session) } // Auto-promote another session to primary when the current primary enters a terminal state s.maybePromotePrimary(ctx, taskID, sessionID, nextState) + return session +} + +func (s *Service) publishTaskSessionStateChanged( + ctx context.Context, + taskID, sessionID string, + oldState, nextState models.TaskSessionState, + errorMessage string, + stateUpdatedAt *time.Time, + session *models.TaskSession, +) { + if s.eventBus == nil || session == nil { + return + } + agentProfileID := session.AgentProfileID + if agentProfileID == "" { + if task, terr := s.repo.GetTask(ctx, taskID); terr == nil && task != nil { + agentProfileID = task.AssigneeAgentProfileID + } + } + eventData := map[string]interface{}{ + metaKeyTaskID: taskID, + metaKeySessionID: sessionID, + "old_state": string(oldState), + metaKeyNewState: string(nextState), + "error_message": errorMessage, + metaKeyAgentProfileID: agentProfileID, + "agent_profile_snapshot": session.AgentProfileSnapshot, + "is_passthrough": session.IsPassthrough, + } + if stateUpdatedAt != nil && !stateUpdatedAt.IsZero() { + eventData[metaKeyUpdatedAt] = stateUpdatedAt.Format(time.RFC3339Nano) + } + if session.ReviewStatus != models.ReviewStatusNone { + eventData["review_status"] = string(session.ReviewStatus) + } + if len(session.Metadata) > 0 { + eventData["session_metadata"] = session.Metadata + } + if suppressed, ok := s.suppressToast.LoadAndDelete(sessionID); ok && suppressed.(bool) { + eventData["suppress_toast"] = true + } + if session.TaskEnvironmentID != "" { + eventData["task_environment_id"] = session.TaskEnvironmentID + } + _ = s.eventBus.Publish(ctx, events.TaskSessionStateChanged, bus.NewEvent(events.TaskSessionStateChanged, "task-session", eventData)) } // maybePromotePrimary promotes the next best active session to primary when the @@ -537,7 +561,11 @@ func (s *Service) setSessionWaitingForInput(ctx context.Context, taskID, session } wasAlreadyWaiting := session.State == models.TaskSessionStateWaitingForInput - s.updateTaskSessionState(ctx, taskID, sessionID, models.TaskSessionStateWaitingForInput, "", false, session) + if updatedSession := s.updateTaskSessionState(ctx, taskID, sessionID, models.TaskSessionStateWaitingForInput, "", false, session); updatedSession != nil { + if len(preloadedSession) > 0 && preloadedSession[0] != nil && preloadedSession[0] != updatedSession { + *preloadedSession[0] = *updatedSession + } + } if wasAlreadyWaiting { return @@ -582,7 +610,11 @@ func (s *Service) setSessionRunning(ctx context.Context, taskID, sessionID strin // (2,000+ redundant writes observed on long-running turns). wasAlreadyRunning := session.State == models.TaskSessionStateRunning - s.updateTaskSessionState(ctx, taskID, sessionID, models.TaskSessionStateRunning, "", true, session) + if updatedSession := s.updateTaskSessionState(ctx, taskID, sessionID, models.TaskSessionStateRunning, "", true, session); updatedSession != nil { + if len(preloadedSession) > 0 && preloadedSession[0] != nil && preloadedSession[0] != updatedSession { + *preloadedSession[0] = *updatedSession + } + } if wasAlreadyRunning { return diff --git a/apps/backend/internal/orchestrator/event_handlers_streaming_test.go b/apps/backend/internal/orchestrator/event_handlers_streaming_test.go index 862f4cddc6..297fb92bfb 100644 --- a/apps/backend/internal/orchestrator/event_handlers_streaming_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_streaming_test.go @@ -9,6 +9,7 @@ import ( "github.com/kandev/kandev/internal/agent/runtime/lifecycle" "github.com/kandev/kandev/internal/agentctl/types/streams" + "github.com/kandev/kandev/internal/events" "github.com/kandev/kandev/internal/events/bus" "github.com/kandev/kandev/internal/task/models" v1 "github.com/kandev/kandev/pkg/api/v1" @@ -40,6 +41,25 @@ func (b *recordingEventBus) Request(context.Context, string, *bus.Event, time.Du func (b *recordingEventBus) Close() {} func (b *recordingEventBus) IsConnected() bool { return true } +func TestUpdateTaskSessionStatePublishesPersistedUpdatedAt(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + eb := &recordingEventBus{} + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + svc.eventBus = eb + + svc.updateTaskSessionState(ctx, "t1", "s1", models.TaskSessionStateWaitingForInput, "", false) + + require.Len(t, eb.events, 1) + require.Equal(t, events.TaskSessionStateChanged, eb.events[0].subject) + data, ok := eb.events[0].event.Data.(map[string]interface{}) + require.True(t, ok) + session, err := repo.GetTaskSession(ctx, "s1") + require.NoError(t, err) + require.Equal(t, session.UpdatedAt.UTC().Format(time.RFC3339Nano), data["updated_at"]) +} + func TestHandleSessionModeEvent(t *testing.T) { t.Run("publishes plan mode", func(t *testing.T) { eb := &recordingEventBus{} diff --git a/apps/backend/internal/orchestrator/event_handlers_transient.go b/apps/backend/internal/orchestrator/event_handlers_transient.go index 64ea1e0509..3a96a1a75d 100644 --- a/apps/backend/internal/orchestrator/event_handlers_transient.go +++ b/apps/backend/internal/orchestrator/event_handlers_transient.go @@ -29,9 +29,12 @@ const recoveryCancelRetryButtonTestID = "recovery-cancel-retry-button" // keys are built in more than one place in this package (recovery + retry // status messages), which otherwise trips goconst on new code. const ( - metaKeyVariant = "variant" - metaKeySessionID = "session_id" - metaKeyTaskID = "task_id" + metaKeyVariant = "variant" + metaKeySessionID = "session_id" + metaKeyTaskID = "task_id" + metaKeyNewState = "new_state" + metaKeyAgentProfileID = "agent_profile_id" + metaKeyUpdatedAt = "updated_at" ) // metaVariantWarning is the status-message variant that drives the frontend's diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow.go b/apps/backend/internal/orchestrator/event_handlers_workflow.go index 0c6760addd..716d27bd74 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow.go +++ b/apps/backend/internal/orchestrator/event_handlers_workflow.go @@ -1788,10 +1788,10 @@ func (s *Service) publishSessionWaitingEvent(ctx context.Context, taskID, sessio return } eventData := map[string]interface{}{ - "task_id": taskID, - "session_id": sessionID, + metaKeyTaskID: taskID, + metaKeySessionID: sessionID, "workflow_step_id": stepID, - "new_state": string(models.TaskSessionStateWaitingForInput), + metaKeyNewState: string(models.TaskSessionStateWaitingForInput), } // Include agent_profile_id and session metadata so the frontend can // identify the agent (e.g. MCP support) without waiting for SSR hydration. @@ -1802,6 +1802,9 @@ func (s *Service) publishSessionWaitingEvent(ctx context.Context, taskID, sessio session = s } if session != nil { + if !session.UpdatedAt.IsZero() { + eventData[metaKeyUpdatedAt] = session.UpdatedAt.UTC().Format(time.RFC3339Nano) + } if session.AgentProfileID != "" { eventData["agent_profile_id"] = session.AgentProfileID } @@ -1831,14 +1834,17 @@ func (s *Service) publishSessionCreatedEvent(ctx context.Context, taskID, sessio return } eventData := map[string]interface{}{ - "task_id": taskID, - "session_id": sessionID, - "new_state": string(models.TaskSessionStateCreated), + metaKeyTaskID: taskID, + metaKeySessionID: sessionID, + metaKeyNewState: string(models.TaskSessionStateCreated), } if stepID != "" { eventData["workflow_step_id"] = stepID } if session, err := s.repo.GetTaskSession(ctx, sessionID); err == nil && session != nil { + if !session.UpdatedAt.IsZero() { + eventData[metaKeyUpdatedAt] = session.UpdatedAt.UTC().Format(time.RFC3339Nano) + } if session.AgentProfileID != "" { eventData["agent_profile_id"] = session.AgentProfileID } diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow_test.go b/apps/backend/internal/orchestrator/event_handlers_workflow_test.go index 8d5bd0e5d7..b7bdb3c30a 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_workflow_test.go @@ -95,6 +95,13 @@ func TestPublishSessionWaitingEvent(t *testing.T) { if data["new_state"] != string(models.TaskSessionStateWaitingForInput) { t.Errorf("expected new_state %q, got %q", models.TaskSessionStateWaitingForInput, data["new_state"]) } + session, err = repo.GetTaskSession(ctx, "s1") + if err != nil { + t.Fatalf("GetTaskSession: %v", err) + } + if data["updated_at"] != session.UpdatedAt.UTC().Format(time.RFC3339Nano) { + t.Errorf("expected updated_at %q, got %q", session.UpdatedAt.UTC().Format(time.RFC3339Nano), data["updated_at"]) + } if data["agent_profile_id"] != "profile-auggie" { t.Errorf("expected agent_profile_id %q, got %v", "profile-auggie", data["agent_profile_id"]) } @@ -138,3 +145,34 @@ func TestPublishSessionWaitingEvent(t *testing.T) { svc.publishSessionWaitingEvent(ctx, "t1", "s1", "step1") }) } + +func TestPublishSessionCreatedEventIncludesUpdatedAt(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + + eb := &mockEventBus{} + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + svc.eventBus = eb + + svc.publishSessionCreatedEvent(ctx, "t1", "s1", "step1") + + published := eb.published() + if len(published) != 1 { + t.Fatalf("expected 1 published event, got %d", len(published)) + } + data, ok := published[0].Event.Data.(map[string]any) + if !ok { + t.Fatalf("expected event data to be map[string]any, got %T", published[0].Event.Data) + } + if data["new_state"] != string(models.TaskSessionStateCreated) { + t.Errorf("expected new_state %q, got %q", models.TaskSessionStateCreated, data["new_state"]) + } + session, err := repo.GetTaskSession(ctx, "s1") + if err != nil { + t.Fatalf("GetTaskSession: %v", err) + } + if data["updated_at"] != session.UpdatedAt.UTC().Format(time.RFC3339Nano) { + t.Errorf("expected updated_at %q, got %q", session.UpdatedAt.UTC().Format(time.RFC3339Nano), data["updated_at"]) + } +} diff --git a/apps/backend/internal/orchestrator/task_operations.go b/apps/backend/internal/orchestrator/task_operations.go index 9f1eee3745..d2bfa359f7 100644 --- a/apps/backend/internal/orchestrator/task_operations.go +++ b/apps/backend/internal/orchestrator/task_operations.go @@ -1272,6 +1272,7 @@ func (s *Service) GetTaskSessionStatus(ctx context.Context, taskID, sessionID st } resp.State = string(session.State) + resp.UpdatedAt = session.UpdatedAt.UTC().Format(time.RFC3339Nano) resp.AgentProfileID = session.AgentProfileID s.populateExecutorStatusInfo(ctx, session, &resp) @@ -1297,6 +1298,9 @@ func (s *Service) GetTaskSessionStatus(ctx context.Context, taskID, sessionID st if refreshErr == nil && refreshedSession != nil { session = refreshedSession resp.State = string(session.State) + if !session.UpdatedAt.IsZero() { + resp.UpdatedAt = session.UpdatedAt.UTC().Format(time.RFC3339Nano) + } } } diff --git a/apps/backend/internal/orchestrator/task_operations_test.go b/apps/backend/internal/orchestrator/task_operations_test.go index 58f61f5f5c..97cdd80a26 100644 --- a/apps/backend/internal/orchestrator/task_operations_test.go +++ b/apps/backend/internal/orchestrator/task_operations_test.go @@ -1295,6 +1295,9 @@ func TestGetTaskSessionStatus_HealsStuckStartingSession(t *testing.T) { if updated.State != models.TaskSessionStateWaitingForInput { t.Fatalf("expected persisted session state %q, got %q", models.TaskSessionStateWaitingForInput, updated.State) } + if resp.UpdatedAt != updated.UpdatedAt.UTC().Format(time.RFC3339Nano) { + t.Fatalf("expected response updated_at %q, got %q", updated.UpdatedAt.UTC().Format(time.RFC3339Nano), resp.UpdatedAt) + } if state, ok := taskRepo.updatedStates["task1"]; !ok || state != v1.TaskStateReview { t.Fatalf("expected task state %q, got %q (ok=%v)", v1.TaskStateReview, state, ok) } diff --git a/apps/web/e2e/helpers/session.ts b/apps/web/e2e/helpers/session.ts index b49cc99f78..8953976864 100644 --- a/apps/web/e2e/helpers/session.ts +++ b/apps/web/e2e/helpers/session.ts @@ -93,6 +93,6 @@ export async function seedIdleSession( await testPage.goto(`/t/${task.id}`); const session = new SessionPage(testPage); await session.waitForLoad(); - await expect(session.idleInput()).toBeVisible({ timeout: 30_000 }); + await session.waitForChatIdle({ timeout: 30_000 }); return session; } diff --git a/apps/web/e2e/pages/session-page.ts b/apps/web/e2e/pages/session-page.ts index 5e4df28b56..6b60318362 100644 --- a/apps/web/e2e/pages/session-page.ts +++ b/apps/web/e2e/pages/session-page.ts @@ -107,33 +107,49 @@ export class SessionPage { * never renders. SSR picks up the right state on the next page load, so * one targeted reload-and-retry is enough to recover. * + * After a backend restart, auto-resume can briefly surface the recovery + * prompt ("Environment setup failed"); click through it when visible. + * * This is the same race the office agent-run-live spec rides out with * `expect.poll`-based re-seeding. */ - // opts.timeout is a soft cap on the *post-attempt* wait, not a hard cap - // on total elapsed time. Worst-case wall-clock is roughly - // attemptTimeout + reload + activeChat-wait + remaining - // where `remaining` is at least `attemptTimeout` so the second wait isn't - // starved when `timeout - attemptTimeout` is too small. The race we're - // covering resolves on the reload's SSR, so the second wait normally - // returns in well under a second; the generous floor exists for the - // pathological "page didn't fully load yet" cases. async waitForChatIdle(opts: { timeout?: number; attemptTimeout?: number } = {}) { const softTotalTimeout = opts.timeout ?? 45_000; - const attemptTimeout = opts.attemptTimeout ?? 15_000; + const attemptTimeout = + opts.attemptTimeout ?? Math.min(15_000, Math.max(5_000, Math.floor(softTotalTimeout / 3))); + const pollSlice = 1_500; const idle = this.idleInput(); - try { - await idle.waitFor({ state: "visible", timeout: attemptTimeout }); - return; - } catch { - // Reload once to re-SSR with the latest session state, then wait the - // remaining budget. If the placeholder still doesn't appear after a - // reload, fall through to the original error semantics. + const start = Date.now(); + let reloaded = false; + + while (Date.now() - start < softTotalTimeout) { + if (await idle.isVisible()) return; + + if (await this.recoveryResumeButton().isVisible()) { + await this.recoveryResumeButton().click(); + await this.recoveryResumeButton() + .waitFor({ state: "hidden", timeout: pollSlice }) + .catch(() => undefined); + continue; + } + + const elapsed = Date.now() - start; + if (!reloaded && elapsed >= attemptTimeout) { + reloaded = true; + await this.page.reload(); + await this.activeChat() + .waitFor({ state: "visible", timeout: attemptTimeout }) + .catch(() => undefined); + continue; + } + + const remaining = softTotalTimeout - elapsed; + await idle + .waitFor({ state: "visible", timeout: Math.min(pollSlice, remaining) }) + .catch(() => undefined); } - await this.page.reload(); - await this.activeChat().waitFor({ state: "visible", timeout: attemptTimeout }); - const remaining = Math.max(attemptTimeout, softTotalTimeout - attemptTimeout); - await idle.waitFor({ state: "visible", timeout: remaining }); + + await idle.waitFor({ state: "visible", timeout: 1_000 }); } /** Wait for the passthrough terminal to be visible (for TUI/passthrough sessions). */ diff --git a/apps/web/e2e/tests/search/shared.ts b/apps/web/e2e/tests/search/shared.ts index d9c716a8ce..43e2aa845c 100644 --- a/apps/web/e2e/tests/search/shared.ts +++ b/apps/web/e2e/tests/search/shared.ts @@ -1,4 +1,4 @@ -import { type Page, expect } from "@playwright/test"; +import { type Page } from "@playwright/test"; import type { SeedData } from "../../fixtures/test-base"; import type { ApiClient } from "../../helpers/api-client"; import { SessionPage } from "../../pages/session-page"; @@ -30,7 +30,7 @@ export async function seedTask( await page.goto(`/t/${task.id}${query}`); const session = new SessionPage(page); await session.waitForLoad(); - await expect(session.idleInput()).toBeVisible({ timeout: 30_000 }); + await session.waitForChatIdle({ timeout: 30_000 }); return { session, taskId: task.id, sessionId: task.session_id! }; } diff --git a/apps/web/e2e/tests/session/executor-not-found.spec.ts b/apps/web/e2e/tests/session/executor-not-found.spec.ts index e71dc43e2a..8cb429ed31 100644 --- a/apps/web/e2e/tests/session/executor-not-found.spec.ts +++ b/apps/web/e2e/tests/session/executor-not-found.spec.ts @@ -48,7 +48,7 @@ test.describe("Executor not found after backend restart", () => { await expect(session.chat.getByText("simple mock response", { exact: false })).toBeVisible({ timeout: 30_000, }); - await expect(session.idleInput()).toBeVisible({ timeout: 15_000 }); + await session.waitForChatIdle({ timeout: 15_000 }); // 3. Restart the backend — clears in-memory execution store, // but DB still has the old AgentExecutionID @@ -59,7 +59,7 @@ test.describe("Executor not found after backend restart", () => { await session.waitForLoad(); // 5. Wait for auto-resume to complete (workspace restoration) - await expect(session.idleInput()).toBeVisible({ timeout: 60_000 }); + await session.waitForChatIdle({ timeout: 60_000 }); // 6. Send a NEW prompt — this triggers LaunchPreparedSession which // reads the stale AgentExecutionID from DB and calls diff --git a/apps/web/e2e/tests/session/session-resume-commands.spec.ts b/apps/web/e2e/tests/session/session-resume-commands.spec.ts index 5a0b91b8be..21067e0687 100644 --- a/apps/web/e2e/tests/session/session-resume-commands.spec.ts +++ b/apps/web/e2e/tests/session/session-resume-commands.spec.ts @@ -66,7 +66,7 @@ test.describe("Slash commands after session resume", () => { await expect(session.chat.getByText("simple mock response", { exact: false })).toBeVisible({ timeout: 30_000, }); - await expect(session.idleInput()).toBeVisible({ timeout: 15_000 }); + await session.waitForChatIdle({ timeout: 15_000 }); // 3. Verify slash commands work before restart await expectSlashCommandsVisible(testPage); @@ -79,7 +79,7 @@ test.describe("Slash commands after session resume", () => { await session.waitForLoad(); // 6. Wait for auto-resume to complete - await expect(session.idleInput()).toBeVisible({ timeout: 60_000 }); + await session.waitForChatIdle({ timeout: 60_000 }); await expect(session.chat.getByText("Resumed agent Mock", { exact: false })).toBeVisible({ timeout: 15_000, }); @@ -90,7 +90,7 @@ test.describe("Slash commands after session resume", () => { await expect( session.chat.getByText("simple mock response", { exact: false }).nth(1), ).toBeVisible({ timeout: 30_000 }); - await expect(session.idleInput()).toBeVisible({ timeout: 15_000 }); + await session.waitForChatIdle({ timeout: 15_000 }); // 8. Verify slash commands work after resume await expectSlashCommandsVisible(testPage); diff --git a/apps/web/e2e/tests/session/session-resume-dedup.spec.ts b/apps/web/e2e/tests/session/session-resume-dedup.spec.ts index 527be2bccd..3cae874a30 100644 --- a/apps/web/e2e/tests/session/session-resume-dedup.spec.ts +++ b/apps/web/e2e/tests/session/session-resume-dedup.spec.ts @@ -17,33 +17,6 @@ async function openTaskSession(page: Page, title: string): Promise return session; } -/** - * Wait for the chat input to become idle, recovering from the transient - * "Environment setup failed" state that can appear after a backend restart - * (documented race in `executor_execute.go:356`). Mirrors what real users do: - * click "Resume" when the recovery prompt appears, then wait for idle. - */ -async function waitForIdleWithRecovery(session: SessionPage, timeoutMs: number): Promise { - const start = Date.now(); - for (;;) { - const remaining = timeoutMs - (Date.now() - start); - if (remaining <= 0) { - // Fall through to a final assertion so failure messages are useful. - await expect(session.idleInput()).toBeVisible({ timeout: 1_000 }); - return; - } - if (await session.idleInput().isVisible()) return; - if (await session.recoveryResumeButton().isVisible()) { - await session.recoveryResumeButton().click(); - continue; - } - await session - .idleInput() - .waitFor({ state: "visible", timeout: Math.min(remaining, 1_500) }) - .catch(() => undefined); - } -} - test.describe("Session resume boot-message dedup", () => { // Test restarts the backend multiple times — can be flaky under CI load. test.describe.configure({ retries: 1 }); @@ -73,7 +46,7 @@ test.describe("Session resume boot-message dedup", () => { await expect(session.chat.getByText("simple mock response", { exact: false })).toBeVisible({ timeout: 30_000, }); - await waitForIdleWithRecovery(session, 30_000); + await session.waitForChatIdle({ timeout: 15_000 }); // 2. Initial "Started agent Mock" row should be visible exactly once. await expect(session.chat.getByText("Started agent Mock", { exact: false })).toHaveCount(1, { @@ -89,7 +62,7 @@ test.describe("Session resume boot-message dedup", () => { await session.waitForLoad(); // Auto-resume + agent turn completion can be slow under CI load, and // can transit through the env-setup-failed recovery state. - await waitForIdleWithRecovery(session, 60_000); + await session.waitForChatIdle({ timeout: 60_000 }); await expect(session.chat.getByText("Resumed agent Mock", { exact: false })).toBeVisible({ timeout: 30_000, }); diff --git a/apps/web/e2e/tests/session/session-resume.spec.ts b/apps/web/e2e/tests/session/session-resume.spec.ts index 63400ea41e..92c0ba07e4 100644 --- a/apps/web/e2e/tests/session/session-resume.spec.ts +++ b/apps/web/e2e/tests/session/session-resume.spec.ts @@ -71,7 +71,7 @@ test.describe("Session resume (ACP mode)", () => { await expect(session.chat.getByText("simple mock response", { exact: false })).toBeVisible({ timeout: 30_000, }); - await expect(session.idleInput()).toBeVisible({ timeout: 15_000 }); + await session.waitForChatIdle({ timeout: 15_000 }); // 3. Verify the "Started agent Mock" boot message appeared on initial launch await expect(session.chat.getByText("Started agent Mock", { exact: false })).toBeVisible({ @@ -95,7 +95,7 @@ test.describe("Session resume (ACP mode)", () => { // needs_resume=true and relaunches the agent via session.launch. // The full cycle (backend restart → health check → page reload → SSR → // WS reconnect → auto-resume → agent turn) can be slow under CI load. - await expect(session.idleInput()).toBeVisible({ timeout: 60_000 }); + await session.waitForChatIdle({ timeout: 60_000 }); // 8. Verify the "Resumed agent Mock" boot message appeared after resume await expect(session.chat.getByText("Resumed agent Mock", { exact: false })).toBeVisible({ @@ -146,7 +146,7 @@ test.describe("Task status during resume", () => { await expect(session.chat.getByText("simple mock response", { exact: false })).toBeVisible({ timeout: 30_000, }); - await expect(session.idleInput()).toBeVisible({ timeout: 15_000 }); + await session.waitForChatIdle({ timeout: 15_000 }); // 3. Confirm the task moved to the "Turn Finished" section after the turn completed await expect(session.taskInSection("Status Stable Task", "Turn Finished")).toBeVisible({ @@ -173,7 +173,7 @@ test.describe("Task status during resume", () => { }); // 7. Wait for auto-resume to complete (agent relaunches and becomes idle) - await expect(session.idleInput()).toBeVisible({ timeout: 60_000 }); + await session.waitForChatIdle({ timeout: 60_000 }); // 8. After resume completes, the task must still be in "Turn Finished" await expect(session.taskInSection("Status Stable Task", "Turn Finished")).toBeVisible({ @@ -351,7 +351,7 @@ test.describe("Session resume (multi-session)", () => { await expect(session.chat.getByText("simple mock response", { exact: false })).toBeVisible({ timeout: 30_000, }); - await expect(session.idleInput()).toBeVisible({ timeout: 15_000 }); + await session.waitForChatIdle({ timeout: 15_000 }); // 2. Open second session via the new-session dialog await session.openNewSessionDialog(); @@ -462,7 +462,7 @@ test.describe("Session resume (active turn interrupted)", () => { // the safety-net path: post-restart, no live agent process exists, so // UpdateTaskSessionState moves RUNNING → WAITING_FOR_INPUT and // CompletePendingToolCallsForTurn closes any dangling tool calls. - await expect(session.idleInput()).toBeVisible({ timeout: 90_000 }); + await session.waitForChatIdle({ timeout: 90_000 }); // 6. Send a fresh prompt — proves the resume path produced a usable // execution (not stuck "Agent is not running") even though the previous diff --git a/apps/web/hooks/domains/session/use-session-resumption.test.ts b/apps/web/hooks/domains/session/use-session-resumption.test.ts index 1977f97638..b73acea7cc 100644 --- a/apps/web/hooks/domains/session/use-session-resumption.test.ts +++ b/apps/web/hooks/domains/session/use-session-resumption.test.ts @@ -1,13 +1,29 @@ +import { renderHook, waitFor } from "@testing-library/react"; import { describe, it, expect, vi, beforeEach } from "vitest"; const mockRequest = vi.fn(); +const mockSetTaskSession = vi.fn(); +const mockSetSessionAgentctlStatus = vi.fn(); +let mockConnectionStatus = "connected"; +let mockSessionItems: Record = {}; vi.mock("@/lib/ws/connection", () => ({ getWebSocketClient: () => ({ request: mockRequest }), })); +vi.mock("@/components/state-provider", () => ({ + useAppStore: (selector: (state: Record) => unknown) => + selector({ + connection: { status: mockConnectionStatus }, + taskSessions: { items: mockSessionItems }, + setTaskSession: mockSetTaskSession, + setSessionAgentctlStatus: mockSetSessionAgentctlStatus, + }), +})); + import { resumeWithSilentFallback, + useSessionResumption, type ResumeStateSetter, type ResumptionState, } from "./use-session-resumption"; @@ -52,6 +68,8 @@ function createSetters(): { setters: ResumeStateSetter; calls: SetterCalls } { describe("resumeWithSilentFallback", () => { beforeEach(() => { vi.clearAllMocks(); + mockConnectionStatus = "connected"; + mockSessionItems = {}; // tryLaunch logs caught errors via console.error; silence in tests so the // expected error paths don't pollute the test output. vi.spyOn(console, "error").mockImplementation(() => {}); @@ -192,3 +210,31 @@ describe("resumeWithSilentFallback", () => { expect(setAgentctlReady).not.toHaveBeenCalled(); }); }); + +describe("useSessionResumption", () => { + beforeEach(() => { + vi.clearAllMocks(); + mockConnectionStatus = "connected"; + mockSessionItems = { + s1: { + started_at: "2026-01-01T00:00:00.000Z", + }, + }; + }); + + it("does not mint client timestamps when status has no updated_at", async () => { + mockRequest.mockResolvedValueOnce({ + session_id: "s1", + task_id: "t1", + state: "WAITING_FOR_INPUT", + is_agent_running: false, + is_resumable: false, + needs_resume: false, + }); + + renderHook(() => useSessionResumption("t1", "s1")); + + await waitFor(() => expect(mockSetTaskSession).toHaveBeenCalled()); + expect(mockSetTaskSession).toHaveBeenCalledWith(expect.objectContaining({ updated_at: "" })); + }); +}); diff --git a/apps/web/hooks/domains/session/use-session-resumption.ts b/apps/web/hooks/domains/session/use-session-resumption.ts index 0f2a9aa54d..360380c0fe 100644 --- a/apps/web/hooks/domains/session/use-session-resumption.ts +++ b/apps/web/hooks/domains/session/use-session-resumption.ts @@ -18,6 +18,7 @@ export type SessionStatus = { session_id: string; task_id: string; state: string; + updated_at?: string; agent_profile_id?: string; is_agent_running: boolean; is_resumable: boolean; @@ -240,7 +241,7 @@ function applyStatusToState( task_id: toTaskId(taskId), state: status.state as TaskSessionState, started_at: session?.started_at ?? "", - updated_at: session?.updated_at ?? "", + updated_at: status.updated_at ?? session?.updated_at ?? "", }); } } diff --git a/apps/web/lib/types/backend.ts b/apps/web/lib/types/backend.ts index 7fa3d38259..67e5c1cbe1 100644 --- a/apps/web/lib/types/backend.ts +++ b/apps/web/lib/types/backend.ts @@ -263,6 +263,8 @@ export type TaskSessionStateChangedPayload = { session_id: string; old_state?: string; new_state?: string; + /** Authoritative row timestamp — used to drop out-of-order subscribe snapshots. */ + updated_at?: string; /** * Agent profile id — drives the per-agent live-session selectors on the * sidebar. Empty for sessions launched without a profile. diff --git a/apps/web/lib/ws/handlers/agent-session-pure.test.ts b/apps/web/lib/ws/handlers/agent-session-pure.test.ts index f8dde95bff..c327683aed 100644 --- a/apps/web/lib/ws/handlers/agent-session-pure.test.ts +++ b/apps/web/lib/ws/handlers/agent-session-pure.test.ts @@ -1,6 +1,7 @@ import { describe, it, expect } from "vitest"; import { isTerminalSessionState, + isStaleSessionStateEvent, pickReplacementSessionId, shouldAdoptNewSession, } from "./agent-session"; @@ -36,6 +37,36 @@ describe("isTerminalSessionState", () => { }); }); +describe("isStaleSessionStateEvent", () => { + it("returns false when payload has no updated_at", () => { + expect(isStaleSessionStateEvent({ updated_at: "2026-01-02T00:00:00.000Z" }, undefined)).toBe( + false, + ); + }); + + it("returns false when existing session has no updated_at", () => { + expect(isStaleSessionStateEvent({}, "2026-01-01T00:00:00.000Z")).toBe(false); + }); + + it("returns true when payload updated_at is older than store", () => { + expect( + isStaleSessionStateEvent( + { updated_at: "2026-01-02T00:00:00.000Z" }, + "2026-01-01T00:00:00.000Z", + ), + ).toBe(true); + }); + + it("returns false when payload updated_at is newer than store", () => { + expect( + isStaleSessionStateEvent( + { updated_at: "2026-01-01T00:00:00.000Z" }, + "2026-01-02T00:00:00.000Z", + ), + ).toBe(false); + }); +}); + describe("shouldAdoptNewSession", () => { it("adopts when there is no active session for the task", () => { const state = makeAppState({ diff --git a/apps/web/lib/ws/handlers/agent-session.test.ts b/apps/web/lib/ws/handlers/agent-session.test.ts index ab4b09dfae..77f034588b 100644 --- a/apps/web/lib/ws/handlers/agent-session.test.ts +++ b/apps/web/lib/ws/handlers/agent-session.test.ts @@ -131,6 +131,41 @@ describe("session.state_changed handler", () => { }); }); +describe("session.state_changed stale guard", () => { + let store: ReturnType; + // eslint-disable-next-line @typescript-eslint/no-explicit-any + let handler: (msg: any) => void; + + it("ignores older state events before upserting the session", () => { + const upsertTaskSessionFromEvent = vi.fn(); + store = makeStore({ + taskSessions: { + items: { + "s-1": { + id: "s-1", + task_id: "t-1", + state: "WAITING_FOR_INPUT", + updated_at: "2026-01-02T00:00:00.000Z", + }, + }, + }, + upsertTaskSessionFromEvent, + }); + handler = registerTaskSessionHandlers(store)[STATE_CHANGED_EVENT]!; + + handler( + makeMessage({ + task_id: "t-1", + session_id: "s-1", + new_state: "RUNNING", + updated_at: "2026-01-01T00:00:00.000Z", + }), + ); + + expect(upsertTaskSessionFromEvent).not.toHaveBeenCalled(); + }); +}); + describe("session.state_changed → active session switching", () => { beforeEach(() => { vi.clearAllMocks(); diff --git a/apps/web/lib/ws/handlers/agent-session.ts b/apps/web/lib/ws/handlers/agent-session.ts index b8be8f2aaf..c907d302ab 100644 --- a/apps/web/lib/ws/handlers/agent-session.ts +++ b/apps/web/lib/ws/handlers/agent-session.ts @@ -91,13 +91,26 @@ export function pickReplacementSessionId(state: AppState, taskId: string): strin return null; } +/** Ignore subscribe snapshots that were read before a newer state landed. */ +export function isStaleSessionStateEvent( + existing: { updated_at?: string } | null | undefined, + payloadUpdatedAt: string | undefined, +): boolean { + if (!payloadUpdatedAt || !existing?.updated_at) return false; + const payloadTime = Date.parse(payloadUpdatedAt); + const existingTime = Date.parse(existing.updated_at); + if (Number.isNaN(payloadTime) || Number.isNaN(existingTime)) return false; + // Strict less-than: equal timestamps are treated as not-stale so identical + // events upsert idempotently rather than being silently dropped. + return payloadTime < existingTime; +} + /** Build a session update object from the state_changed payload. */ // eslint-disable-next-line @typescript-eslint/no-explicit-any function buildSessionUpdate(payload: any): Record { const update: Record = {}; if (payload.new_state) update.state = payload.new_state; if (payload.agent_profile_id) update.agent_profile_id = payload.agent_profile_id; - if (payload.agent_profile_id) update.agent_profile_id = payload.agent_profile_id; if (payload.review_status !== undefined) update.review_status = payload.review_status; if (payload.error_message !== undefined) update.error_message = payload.error_message; if (payload.agent_profile_snapshot) @@ -105,6 +118,7 @@ function buildSessionUpdate(payload: any): Record { if (payload.is_passthrough !== undefined) update.is_passthrough = payload.is_passthrough; if (payload.session_metadata !== undefined) update.metadata = payload.session_metadata; if (payload.task_environment_id) update.task_environment_id = payload.task_environment_id; + if (payload.updated_at) update.updated_at = payload.updated_at; return update; } @@ -129,7 +143,7 @@ function upsertTaskSessionList( task_id: taskId, state: (newState ?? existing?.state) as TaskSessionState, started_at: existing?.started_at ?? "", - updated_at: existing?.updated_at ?? "", + updated_at: (sessionUpdate.updated_at as string | undefined) ?? existing?.updated_at ?? "", ...(payload.agent_profile_id ? { agent_profile_id: payload.agent_profile_id } : {}), ...sessionUpdate, }); @@ -341,6 +355,17 @@ export function registerTaskSessionHandlers(store: StoreApi): WsHandle const sessionUpdate = buildSessionUpdate(payload); const existingSession = store.getState().taskSessions.items[sessionId]; + if (isStaleSessionStateEvent(existingSession, payload.updated_at)) { + debug("state_changed ignored stale snapshot", { + sessionId, + task_id: taskId, + existingUpdatedAt: existingSession?.updated_at, + payloadUpdatedAt: payload.updated_at, + newState: newState ?? "-", + }); + return; + } + debug("state_changed", { sessionId, // Logged before upsertTaskSessionList below, so on the first event for a