dreampiler/kandev
1
0
Fork 0
Code Issues 43 Pull Requests Actions 6 Packages Projects Releases Wiki Activity

fix(e2e): ignore stale session state snapshots blocking idle input (#1208)

Browse Source
This commit is contained in:
Carlos Florêncio
2026-06-01 10:35:07 +01:00
committed by GitHub
parent d8e0507d17
commit 6734b64b54
26 changed files with 523 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/cmd/kandev/helpers.go
+7 -4
View File
@@ -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,
},
apps/backend/cmd/kandev/helpers_test.go
+45
View File
@@ -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.
apps/backend/internal/mcp/handlers/handlers.go
+24
View File
@@ -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 {
apps/backend/internal/mcp/handlers/handlers_test.go
+84
View File
@@ -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{}{
apps/backend/internal/office/testharness/routes.go
+1
View File
@@ -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 {
apps/backend/internal/orchestrator/dto/dto.go
+1
View File
@@ -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
apps/backend/internal/orchestrator/event_handlers_streaming.go
+83 -51
View File
@@ -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
apps/backend/internal/orchestrator/event_handlers_streaming_test.go
+20
View File
@@ -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{}
apps/backend/internal/orchestrator/event_handlers_transient.go
+6 -3
View File
@@ -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
apps/backend/internal/orchestrator/event_handlers_workflow.go
+12 -6
View File
@@ -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
}
apps/backend/internal/orchestrator/event_handlers_workflow_test.go
+38
View File
@@ -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"])
}
}
apps/backend/internal/orchestrator/task_operations.go
+4
View File
@@ -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)
}
}
}
apps/backend/internal/orchestrator/task_operations_test.go
+3
View File
@@ -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)
}
apps/web/e2e/helpers/session.ts
+1 -1
View File
@@ -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;
}
apps/web/e2e/pages/session-page.ts
+36 -20
View File
@@ -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). */
apps/web/e2e/tests/search/shared.ts
+2 -2
View File
@@ -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! };
}
apps/web/e2e/tests/session/executor-not-found.spec.ts
+2 -2
View File
@@ -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
apps/web/e2e/tests/session/session-resume-commands.spec.ts
+3 -3
View File
@@ -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);
apps/web/e2e/tests/session/session-resume-dedup.spec.ts
+2 -29
View File
@@ -17,33 +17,6 @@ async function openTaskSession(page: Page, title: string): Promise<SessionPage>
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<void> {
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,
});
apps/web/e2e/tests/session/session-resume.spec.ts
+6 -6
View File
@@ -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
apps/web/hooks/domains/session/use-session-resumption.test.ts
+46
View File
@@ -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<string, { started_at?: string; updated_at?: string }> = {};
vi.mock("@/lib/ws/connection", () => ({
getWebSocketClient: () => ({ request: mockRequest }),
}));
vi.mock("@/components/state-provider", () => ({
useAppStore: (selector: (state: Record<string, unknown>) => 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: "" }));
});
});
apps/web/hooks/domains/session/use-session-resumption.ts
+2 -1
View File
@@ -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 ?? "",
});
}
}
apps/web/lib/types/backend.ts
+2
View File
@@ -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.
apps/web/lib/ws/handlers/agent-session-pure.test.ts
+31
View File
@@ -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({
apps/web/lib/ws/handlers/agent-session.test.ts
+35
View File
@@ -131,6 +131,41 @@ describe("session.state_changed handler", () => {
});
});
describe("session.state_changed stale guard", () => {
let store: ReturnType<typeof makeStore>;
// 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();
apps/web/lib/ws/handlers/agent-session.ts
+27 -2
View File
@@ -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<string, unknown> {
const update: Record<string, unknown> = {};
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<string, unknown> {
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<AppState>): 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