From b041702866bc7e1aa0e95ad75e9bcc4acf2fa658 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20Flor=C3=AAncio?= Date: Mon, 27 Jul 2026 18:10:36 +0100 Subject: [PATCH] fix(orchestrator): clear sidebar spinner when session settles (#1981) --- .../background_work_accounting_test.go | 32 ++++-- .../orchestrator/event_handlers_streaming.go | 24 +++++ .../foreground_activity_signal_test.go | 63 +++++++++++ apps/backend/internal/review/runner.go | 31 ++++-- apps/backend/internal/review/runner_test.go | 76 +++++++++++++ .../task/mobile-add-workspace-sources.spec.ts | 8 ++ .../task/sidebar-settled-spinner.spec.ts | 101 ++++++++++++++++++ 7 files changed, 321 insertions(+), 14 deletions(-) create mode 100644 apps/web/e2e/tests/task/sidebar-settled-spinner.spec.ts diff --git a/apps/backend/internal/orchestrator/background_work_accounting_test.go b/apps/backend/internal/orchestrator/background_work_accounting_test.go index fee47b55ad..b81a85053e 100644 --- a/apps/backend/internal/orchestrator/background_work_accounting_test.go +++ b/apps/backend/internal/orchestrator/background_work_accounting_test.go @@ -1046,8 +1046,14 @@ func TestStopSessionPathPublishesStateAndTeardownActivityClears(t *testing.T) { if stateClears != 1 || activityClears != 1 { t.Fatalf("session.stop clear cardinality: state=%d activity=%d, want 1/1", stateClears, activityClears) } - if len(taskEvents.activityTaskIDs) != 1 || taskEvents.activityTaskIDs[0] != taskID { - t.Fatalf("task aggregate cleanup publications = %v, want [%s]", taskEvents.activityTaskIDs, taskID) + // Two task-aggregate recomputes: the teardown activity clear, plus the + // republish when the session leaves RUNNING (republishTaskActivityOnSettle), + // which corrects a task aggregate that a detached record would otherwise leave + // stuck at generating. + if len(taskEvents.activityTaskIDs) != 2 || + taskEvents.activityTaskIDs[0] != taskID || + taskEvents.activityTaskIDs[1] != taskID { + t.Fatalf("task aggregate cleanup publications = %v, want [%s %s]", taskEvents.activityTaskIDs, taskID, taskID) } if svc.hasBackgroundTask(sessionID, "tool-stop-path") { t.Fatal("session.stop teardown retained owned background work") @@ -1104,10 +1110,14 @@ func TestExecutionTerminalEvents_ReconcileMissingBackgroundCompletion(t *testing if got := countActivityClears(recorded); got != 1 { t.Fatalf("terminal session activity clears = %d, want exactly one", got) } - if len(taskEvents.activityTaskIDs) != 2 || + // Three task-aggregate recomputes: the retirement activity clear, the + // reconciled background completion, plus the settle republish when the + // session leaves RUNNING (republishTaskActivityOnSettle). + if len(taskEvents.activityTaskIDs) != 3 || taskEvents.activityTaskIDs[0] != taskID || - taskEvents.activityTaskIDs[1] != taskID { - t.Fatalf("terminal task recomputes = %v, want [%s %s]", taskEvents.activityTaskIDs, taskID, taskID) + taskEvents.activityTaskIDs[1] != taskID || + taskEvents.activityTaskIDs[2] != taskID { + t.Fatalf("terminal task recomputes = %v, want [%s %s %s]", taskEvents.activityTaskIDs, taskID, taskID, taskID) } waitForStopCall(t, agentManager) }) @@ -1145,8 +1155,16 @@ func TestTransientFailurePreservesBackgroundRegistration(t *testing.T) { if !svc.hasBackgroundTask("s1", "tool-transient") { t.Fatal("transient failure cleaned background registration before retry teardown") } - if got := countActivityClears(recorded); got != 0 || len(taskEvents.activityTaskIDs) != 0 { - t.Fatalf("transient cleanup publications: session=%d task=%v", got, taskEvents.activityTaskIDs) + // The transient path keeps the background registration and never clears the + // session-level activity (no session activity_changed clear). It does park the + // session in WAITING_FOR_INPUT, so the task aggregate is republished once + // (republishTaskActivityOnSettle) to reflect the calm, non-generating retry + // state on the sidebar/board. + if got := countActivityClears(recorded); got != 0 { + t.Fatalf("transient cleanup must not clear session activity, got %d", got) + } + if len(taskEvents.activityTaskIDs) != 1 || taskEvents.activityTaskIDs[0] != "t1" { + t.Fatalf("transient park must republish task aggregate once, got %v", taskEvents.activityTaskIDs) } } diff --git a/apps/backend/internal/orchestrator/event_handlers_streaming.go b/apps/backend/internal/orchestrator/event_handlers_streaming.go index e92e299f76..b503512a07 100644 --- a/apps/backend/internal/orchestrator/event_handlers_streaming.go +++ b/apps/backend/internal/orchestrator/event_handlers_streaming.go @@ -818,11 +818,34 @@ func (s *Service) updateTaskSessionStateWithHook( s.publishTaskSessionStateChanged(ctx, taskID, sessionID, oldState, nextState, errorMessage, authoritativeUpdatedAt, session) } + s.republishTaskActivityOnSettle(ctx, taskID, oldState, nextState) + // Auto-promote another session to primary when the current primary enters a terminal state s.maybePromotePrimary(ctx, taskID, sessionID, nextState) return session, true } +// republishTaskActivityOnSettle recomputes the task-level MOST-ACTIVE-WINS +// activity aggregate when a session leaves the generating-capable RUNNING state. +// On agent completion the turn-activity record is retired (detached) while the +// session is still RUNNING, then the session settles to WAITING_FOR_INPUT (or a +// terminal state) without a task-level republish. A detached record safely +// defaults to "generating" (turn_activity.go isForegroundTurnGenerating), so the +// last cached task aggregate stays "generating" and the sidebar/board spinner +// never clears. Republishing off the settled session list corrects the aggregate +// (the completed session is no longer RUNNING, so it drops out); the call is +// deduplicated by the task service and is a no-op when unchanged. +func (s *Service) republishTaskActivityOnSettle( + ctx context.Context, + taskID string, + oldState, nextState models.TaskSessionState, +) { + if oldState != models.TaskSessionStateRunning || nextState == models.TaskSessionStateRunning { + return + } + s.publishTaskActivityIfChanged(ctx, taskID) +} + func (s *Service) persistTaskSessionState( ctx context.Context, sessionID string, @@ -926,6 +949,7 @@ func (s *Service) transitionTaskSessionState( authoritativeUpdatedAt, refreshed, ) + s.republishTaskActivityOnSettle(ctx, taskID, oldState, nextState) s.maybePromotePrimary(ctx, taskID, sessionID, nextState) return true, nextState, nil } diff --git a/apps/backend/internal/orchestrator/foreground_activity_signal_test.go b/apps/backend/internal/orchestrator/foreground_activity_signal_test.go index 416ee34190..170bea6581 100644 --- a/apps/backend/internal/orchestrator/foreground_activity_signal_test.go +++ b/apps/backend/internal/orchestrator/foreground_activity_signal_test.go @@ -911,6 +911,69 @@ func TestForegroundActivitySignal_TurnCompletionIsSessionIsolated(t *testing.T) } } +// TestForegroundActivitySignal_SettleRepublishesTaskAggregate reproduces the +// stuck-sidebar-spinner bug: on agent completion the turn-activity record is +// retired (detached) while the session is still RUNNING, so a task-aggregate +// recompute at that instant reads the detached record's "generating" safe +// default and caches the task-level activity as generating. The session then +// settles to WAITING_FOR_INPUT via the state funnel, which must republish the +// task aggregate so the at-a-glance surfaces (sidebar row, board card) clear. +func TestForegroundActivitySignal_SettleRepublishesTaskAggregate(t *testing.T) { + repo := setupTestRepo(t) + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + eb := &recordingEventBus{} + svc.eventBus = eb + taskEvents := &recordingTaskEvents{} + svc.SetTaskEventPublisher(taskEvents) + + const ( + taskID = "task-settle-republish" + sessionID = "session-settle-republish" + ) + // RUNNING session with no tracked turn-activity record: the detached state + // left behind after retirement, whose foreground activity safely defaults to + // generating. + seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateRunning) + if got := svc.ForegroundActivity(sessionID); got != v1.ForegroundActivityGenerating { + t.Fatalf("untracked RUNNING session must default generating, got %q", got) + } + taskEvents.activityTaskIDs = nil + + // The session settles out of RUNNING; the funnel must republish the task + // aggregate so a stuck generating value is recomputed off the settled list. + svc.updateTaskSessionState(t.Context(), taskID, sessionID, models.TaskSessionStateWaitingForInput, "", false) + + if len(taskEvents.activityTaskIDs) != 1 || taskEvents.activityTaskIDs[0] != taskID { + t.Fatalf("settle must republish task aggregate once for %q, got %v", taskID, taskEvents.activityTaskIDs) + } +} + +// TestForegroundActivitySignal_NonSettleTransitionDoesNotRepublish guards the +// settle predicate: transitions that do not leave the generating-capable RUNNING +// state (e.g. STARTING->RUNNING) must not trigger a task-aggregate republish, +// so the fix adds no per-frame or startup noise. +func TestForegroundActivitySignal_NonSettleTransitionDoesNotRepublish(t *testing.T) { + repo := setupTestRepo(t) + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + eb := &recordingEventBus{} + svc.eventBus = eb + taskEvents := &recordingTaskEvents{} + svc.SetTaskEventPublisher(taskEvents) + + const ( + taskID = "task-nonsettle" + sessionID = "session-nonsettle" + ) + seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateStarting) + taskEvents.activityTaskIDs = nil + + svc.updateTaskSessionState(t.Context(), taskID, sessionID, models.TaskSessionStateRunning, "", false) + + if len(taskEvents.activityTaskIDs) != 0 { + t.Fatalf("entering RUNNING must not republish task aggregate, got %v", taskEvents.activityTaskIDs) + } +} + func TestForegroundActivitySignal_DelayedOldProviderIdleCannotYieldSuccessor(t *testing.T) { repo := setupTestRepo(t) agentMgr := &mockAgentManager{currentPromptExecutionID: "execution-provider-idle"} diff --git a/apps/backend/internal/review/runner.go b/apps/backend/internal/review/runner.go index be9dceb75f..7300ccb84f 100644 --- a/apps/backend/internal/review/runner.go +++ b/apps/backend/internal/review/runner.go @@ -463,15 +463,20 @@ func (r *Runner) reviewBatches(ctx context.Context, plan BatchPlan, identity Rev acc := accumulator{} for i, batch := range plan.Batches { if err := ctx.Err(); err != nil { - return acc, fmt.Errorf("%w: review cancelled after batch %d", ErrExecutionFailed, i) + // Keep context.Canceled / DeadlineExceeded in the chain so fail() can + // leave a user cancel as cancelled instead of overwriting it failed. + return acc, fmt.Errorf("%w: review cancelled after batch %d: %w", ErrExecutionFailed, i, err) } prompt, err := r.prompts.Build(ctx, batch, promptCtx) if err != nil { - return acc, fmt.Errorf("%w: build prompt: %v", ErrExecutionFailed, err) + return acc, fmt.Errorf("%w: build prompt: %w", ErrExecutionFailed, err) } result, err := r.inference.Run(ctx, identity, sessionID, prompt) if err != nil { - return acc, fmt.Errorf("%w: %v", ErrExecutionFailed, err) + // Preserve the inference error chain (esp. context cancellation). Using + // %v here used to drop context.Canceled, so fail() marked cancelled runs + // as failed when CancelRun lost the race with FailRun. + return acc, fmt.Errorf("%w: %w", ErrExecutionFailed, err) } if result == nil { return acc, fmt.Errorf("%w: reviewer returned no result", ErrExecutionFailed) @@ -483,7 +488,7 @@ func (r *Runner) reviewBatches(ctx context.Context, plan BatchPlan, identity Rev if err != nil { // One unreadable batch fails the run: reporting a partial review as // complete would read as an all-clear for the files we could not parse. - return acc, fmt.Errorf("%w: batch %d: %v", ErrUnparseableResponse, i+1, err) + return acc, fmt.Errorf("%w: batch %d: %w", ErrUnparseableResponse, i+1, err) } acc.findings = append(acc.findings, parsed.Findings...) acc.rejected += parsed.Rejected @@ -498,12 +503,17 @@ func (r *Runner) fail(ctx context.Context, runID string, cause error, started ti // An explicit cancel already recorded the terminal state, so re-reporting it // as a failure would only churn the row (and would overwrite it on any store // without a terminal guard). A deadline is different: nobody recorded that. - if errors.Is(cause, context.Canceled) && !errors.Is(cause, context.DeadlineExceeded) { + // Check both the wrapped cause and the run context: wrappers historically used + // %v and dropped context.Canceled from the chain. + if isReviewCanceled(cause) || isReviewCanceled(ctx.Err()) { r.logger.Debug("review run unwound after cancellation", zap.String("run_id", runID)) - return cause + if cause != nil { + return cause + } + return ctx.Err() } code := CodeFor(cause) - if errors.Is(cause, context.DeadlineExceeded) { + if errors.Is(cause, context.DeadlineExceeded) || errors.Is(ctx.Err(), context.DeadlineExceeded) { code = CodeCancelled } // Persist the failure on a context detached from the (possibly cancelled) @@ -516,6 +526,13 @@ func (r *Runner) fail(ctx context.Context, runID string, cause error, started ti return cause } +// isReviewCanceled reports a user/cancel signal (context.Canceled), not a +// deadline. DeadlineExceeded also satisfies errors.Is(..., Canceled) in some +// chains, so it is excluded explicitly. +func isReviewCanceled(err error) bool { + return err != nil && errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) +} + // anchorFindings attaches the authoritative diff hash and anchor text from the // change set to each reported finding, and drops findings naming a file the // reviewer was never shown. diff --git a/apps/backend/internal/review/runner_test.go b/apps/backend/internal/review/runner_test.go index 9fa129ee95..92024f28cd 100644 --- a/apps/backend/internal/review/runner_test.go +++ b/apps/backend/internal/review/runner_test.go @@ -789,6 +789,82 @@ func TestRunner_CancelStopsInferenceAndKeepsRunCancelled(t *testing.T) { } } +// TestRunner_failSkipsCanceledCause reproduces CI flake where cancel raced +// CancelRun: inference errors wrapped with %v lost context.Canceled, so fail() +// called FailRun before CancelRun and left the row terminal-failed. fail must +// treat a cancelled run context as cancel even when the cause chain is broken. +func TestRunner_failSkipsCanceledCause(t *testing.T) { + runner, store, _ := newBlockingHarness(t, okResponse("a.go", 2)) + runner.Stop() // no live pass; we call fail directly + + run, err := store.CreateRun(context.Background(), taskservice.CreateRunRequest{ + TaskID: "task-fail-cancel", + }) + if err != nil { + t.Fatalf("CreateRun: %v", err) + } + if _, err := store.MarkRunRunning(context.Background(), run.ID); err != nil { + t.Fatalf("MarkRunRunning: %v", err) + } + + // Broken wrapper: historical reviewBatches path used %v and dropped Canceled. + broken := fmt.Errorf("%w: %v", ErrExecutionFailed, context.Canceled) + if errors.Is(broken, context.Canceled) { + t.Fatal("precondition: broken wrapper must not preserve context.Canceled") + } + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + _ = runner.fail(ctx, run.ID, broken, time.Now()) + + if got := store.statusOf(run.ID); got == models.ReviewRunFailed { + t.Fatalf("fail must not mark a cancelled run failed, got %q", got) + } + if failure, ok := store.lastFailure(); ok { + t.Fatalf("fail must not record FailRun on cancel, got %#v", failure) + } +} + +func singleBatchPlan() BatchPlan { + return BatchPlan{Batches: [][]ChangedFile{{{Path: "a.go", Diff: "@@ -1 +1,2 @@\n x\n+y\n"}}}} +} + +// reviewBatches must keep context.Canceled reachable through its %w wrapper so +// fail() can leave a user cancel as cancelled. Reverting the inference wrapper +// to %v would drop it and break this contract. +func TestRunner_reviewBatchesPreservesCanceled(t *testing.T) { + runner := NewRunner(RunnerDeps{ + Store: newFakeStore(), + Resolver: NewResolver(nil, &fakeUtility{found: true, enabled: true, agentID: "a", model: "m"}, nil), + Changes: &fakeChangeSource{}, + Inference: &fakeInference{err: fmt.Errorf("inference aborted: %w", context.Canceled)}, + Prompts: &fakePrompts{}, + Sessions: &fakeSessions{sessionID: "sess-1"}, + Logger: testLogger(t), + }) + + _, err := runner.reviewBatches(context.Background(), singleBatchPlan(), + ReviewerIdentity{Model: "m"}, "sess-1", PromptContext{}) + if !errors.Is(err, ErrExecutionFailed) { + t.Fatalf("expected ErrExecutionFailed in the chain, got %v", err) + } + if !errors.Is(err, context.Canceled) { + t.Fatalf("reviewBatches must keep context.Canceled reachable, got %v", err) + } +} + +// reviewBatches must keep the underlying parse error reachable through its %w +// wrapper on the unparseable-response path. +func TestRunner_reviewBatchesPreservesParseError(t *testing.T) { + h := newRunnerHarness(t, map[string]any{}, []string{"no findings here at all"}) + + _, err := h.runner.reviewBatches(context.Background(), singleBatchPlan(), + ReviewerIdentity{Model: "m"}, "sess-1", PromptContext{}) + if !errors.Is(err, ErrUnparseableResponse) { + t.Fatalf("expected ErrUnparseableResponse in the chain, got %v", err) + } +} + // TestRunner_ConcurrentLaunchesCreateOneRun covers the race where the DB check // and the in-memory claim were not atomic, leaving the loser's pending run row // orphaned with no goroutine behind it. diff --git a/apps/web/e2e/tests/task/mobile-add-workspace-sources.spec.ts b/apps/web/e2e/tests/task/mobile-add-workspace-sources.spec.ts index fd4a6e2caf..b4b85c27ef 100644 --- a/apps/web/e2e/tests/task/mobile-add-workspace-sources.spec.ts +++ b/apps/web/e2e/tests/task/mobile-add-workspace-sources.spec.ts @@ -147,6 +147,14 @@ test("mobile Files drawer attaches sources with fixed controls and persisted wor const addRepository = drawer.getByRole("button", { name: "Add repository" }); const addFolder = drawer.getByRole("button", { name: "Add folder" }); const submit = drawer.getByTestId("add-workspace-sources-submit"); + // The "Add folder" control is gated on activeTask.primaryExecutorType, which + // arrives from a follow-up office.task.updated WS event once the primary + // session's executor is bound and can lag the drawer mount on a loaded CI + // shard. The drawer re-renders reactively when that field hydrates, so wait + // for the button with a generous timeout instead of measuring a not-yet- + // rendered button (a null box that otherwise stalls until the suite timeout). + await expect(addRepository).toBeVisible(); + await expect(addFolder).toBeVisible({ timeout: 30_000 }); const [addRepositoryBox, addFolderBox] = await Promise.all([ addRepository.boundingBox(), addFolder.boundingBox(), diff --git a/apps/web/e2e/tests/task/sidebar-settled-spinner.spec.ts b/apps/web/e2e/tests/task/sidebar-settled-spinner.spec.ts new file mode 100644 index 0000000000..bdfeecb27b --- /dev/null +++ b/apps/web/e2e/tests/task/sidebar-settled-spinner.spec.ts @@ -0,0 +1,101 @@ +/** + * Regression test for the sidebar "stuck generating" spinner: when a task's + * session settles out of RUNNING (agent turn completes -> WAITING_FOR_INPUT, + * task -> COMPLETED), the sidebar row must stop showing the yellow generating + * spinner (`task-state-running`) from a live task.updated stream — without ever + * opening the completed task. + * + * The bug (backend): the turn-activity record was detached while the session was + * still RUNNING, so the recomputed task-level foreground_activity aggregate cached + * "generating"; the session then settled with only a session-level state_changed + * event, so the task aggregate was never republished and the spinner kept spinning. + * Fixed by republishing the task aggregate when a session leaves RUNNING. + * + * The browser must be mounted before the agent starts so the client observes the + * generating task.updated and retains that cached aggregate through settle. A + * post-settle fresh load only hydrates the durable settled state and would pass + * against an unfixed parent revision. + */ +import { expect, test } from "../../fixtures/test-base"; +import type { ApiClient } from "../../helpers/api-client"; +import { SessionPage } from "../../pages/session-page"; + +/** Hold RUNNING long enough for the sidebar spinner assertion, then finish. */ +const SETTLE_SCRIPT = [ + 'e2e:message("starting settle turn...")', + "e2e:delay(4000)", + 'e2e:message("settle turn done")', +].join("\n"); + +async function waitForSessionWaitingForInput( + apiClient: ApiClient, + taskId: string, + message: string, +): Promise { + await expect + .poll( + async () => { + const { sessions } = await apiClient.listTaskSessions(taskId); + return sessions[0]?.state ?? ""; + }, + { message, timeout: 60_000 }, + ) + .toBe("WAITING_FOR_INPUT"); +} + +test.describe("Sidebar spinner clears when a session settles", () => { + test("completed task drops the generating spinner without opening it", async ({ + apiClient, + seedData, + testPage, + }) => { + test.setTimeout(120_000); + + // Mount the sidebar on a different task first so the client is live before + // the agent runs and can receive the generating → settled task.updated stream. + const navTask = await apiClient.createTask(seedData.workspaceId, "Nav Task", { + workflow_id: seedData.workflowId, + workflow_step_id: seedData.startStepId, + }); + await testPage.goto(`/t/${navTask.id}`); + + const session = new SessionPage(testPage); + await session.waitForLoad(); + await expect(session.sidebar).toBeVisible({ timeout: 10_000 }); + + const settledTask = await apiClient.createTaskWithAgent( + seedData.workspaceId, + "Settled Turn", + seedData.agentProfileId, + { + description: SETTLE_SCRIPT, + repository_ids: [seedData.repositoryId], + workflow_id: seedData.workflowId, + workflow_step_id: seedData.startStepId, + }, + ); + + const settledRow = session.sidebarTaskItem("Settled Turn"); + await expect(settledRow).toBeVisible({ timeout: 15_000 }); + + // Client must observe generating while the session is still RUNNING. + await expect(settledRow.getByTestId("task-state-running")).toBeVisible({ + timeout: 30_000, + }); + + await waitForSessionWaitingForInput( + apiClient, + settledTask.id, + "settled task session should finish its turn", + ); + + // Core regression: the settle task.updated must clear the spinner the + // client already painted — never reopen the completed task. + await expect(settledRow.getByTestId("task-state-running")).toHaveCount(0, { + timeout: 15_000, + }); + + // Settled turn with no pending clarification buckets into "review". + await expect(settledRow.getByTestId("task-state-review")).toBeVisible({ timeout: 10_000 }); + }); +});