diff --git a/apps/backend/internal/agent/runtime/lifecycle/env_preparer_local.go b/apps/backend/internal/agent/runtime/lifecycle/env_preparer_local.go index 033ad95e60..3e26279d9b 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/env_preparer_local.go +++ b/apps/backend/internal/agent/runtime/lifecycle/env_preparer_local.go @@ -6,6 +6,8 @@ import ( "fmt" "os" "os/exec" + "regexp" + "sort" "strings" "sync" "time" @@ -106,7 +108,7 @@ func (p *LocalPreparer) Prepare(ctx context.Context, req *EnvPrepareRequest, onP step = beginStep("Checkout branch") step.Command = fmt.Sprintf("git fetch origin %s && git checkout %s", effectiveBranch, effectiveBranch) reportProgress(onProgress, step, stepIdx, totalSteps) - output, err := checkoutBranch(ctx, workspacePath, effectiveBranch) + output, err := checkoutBranch(ctx, workspacePath, effectiveBranch, gitCredentialValues(req.Env)) if err != nil { errMsg := fmt.Sprintf("failed to checkout branch %q: %s", effectiveBranch, output) completeStepError(&step, errMsg) @@ -171,21 +173,62 @@ func readCurrentBranchForLocal(workDir string) string { // Best-effort fetch first so newly-created remote branches are visible, then // the checkout. If the local branch doesn't exist but the remote tracking // branch does (from the fetch), git creates a local branch tracking it. -func checkoutBranch(ctx context.Context, workDir, branch string) (string, error) { +func checkoutBranch(ctx context.Context, workDir, branch string, sensitiveValues []string) (string, error) { fetchCmd := exec.CommandContext(ctx, "git", "fetch", "origin", branch) fetchCmd.Dir = workDir - _ = subproc.RunGit(ctx, fetchCmd) // fetch failure is non-fatal, fall back to local + fetchOut, fetchErr := subproc.RunGitCombinedOutput(ctx, fetchCmd) cmd := exec.CommandContext(ctx, "git", "checkout", branch) cmd.Dir = workDir out, err := subproc.RunGitCombinedOutput(ctx, cmd) - outStr := strings.TrimSpace(string(out)) + outStr := redactCheckoutOutput(strings.TrimSpace(string(out)), sensitiveValues) if err != nil { + if fetchErr != nil { + fetchDetail := redactCheckoutOutput(strings.TrimSpace(string(fetchOut)), sensitiveValues) + outStr = fmt.Sprintf("fetch branch failed: %v: %s\ncheckout branch failed: %s", fetchErr, fetchDetail, outStr) + } return outStr, worktree.ClassifyGitError(outStr, err) } return outStr, nil } +var credentialURLPattern = regexp.MustCompile(`(?i)(https?://)[^\s/@]+@`) + +// redactCheckoutOutput removes credentials from git diagnostics before they +// are persisted or shown to a user. Git commonly echoes the configured remote +// URL on fetch failures, and local executor remotes may contain URL userinfo. +func redactCheckoutOutput(output string, sensitiveValues []string) string { + redacted := credentialURLPattern.ReplaceAllString(output, "${1}[REDACTED]@") + for _, value := range sensitiveValues { + if value != "" { + redacted = strings.ReplaceAll(redacted, value, "[REDACTED]") + } + } + return redacted +} + +// gitCredentialValues returns request environment values that may authenticate +// git operations. Longest values go first so overlapping values are fully +// replaced rather than leaving a suffix exposed. +func gitCredentialValues(env map[string]string) []string { + values := make([]string, 0) + for key, value := range env { + if value != "" && isSensitiveGitEnvKey(key) { + values = append(values, value) + } + } + sort.Slice(values, func(i, j int) bool { return len(values[i]) > len(values[j]) }) + return values +} + +func isSensitiveGitEnvKey(key string) bool { + key = strings.ToUpper(key) + return strings.Contains(key, "TOKEN") || + strings.Contains(key, "SECRET") || + strings.Contains(key, "PASSWORD") || + strings.HasSuffix(key, "_KEY") +} + // setupScriptStreamInterval is the minimum gap between streaming-output callbacks // while a setup script runs. Chatty scripts (e.g. npm install with progress bars) // can emit hundreds of writes per second; throttling keeps WS event volume sane diff --git a/apps/backend/internal/agent/runtime/lifecycle/env_preparer_local_test.go b/apps/backend/internal/agent/runtime/lifecycle/env_preparer_local_test.go index 9d6ac91048..55ea40c71c 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/env_preparer_local_test.go +++ b/apps/backend/internal/agent/runtime/lifecycle/env_preparer_local_test.go @@ -327,6 +327,49 @@ func TestLocalPreparer_CheckoutFailureSurfaces(t *testing.T) { } } +func TestLocalPreparer_PreservesFetchFailureWhenCheckoutAlsoFails(t *testing.T) { + const token = "github-token-that-must-not-leak" + fakeBin := t.TempDir() + fakeGit := filepath.Join(fakeBin, "git") + script := `#!/bin/sh +case "$1" in +fetch) + echo "fatal: unable to access 'https://oauth2:github-token-that-must-not-leak@github.com/kdlbs/kandev.git/': Could not resolve host: github.com" >&2 + exit 128 + ;; +checkout) + echo "error: pathspec '$2' did not match any file(s) known to git; token=github-token-that-must-not-leak" >&2 + exit 1 + ;; +esac +` + if err := os.WriteFile(fakeGit, []byte(script), 0o755); err != nil { + t.Fatalf("write fake git: %v", err) + } + t.Setenv("PATH", fakeBin+string(os.PathListSeparator)+os.Getenv("PATH")) + + preparer := NewLocalPreparer(newTestLocalLogger()) + result, err := preparer.Prepare(context.Background(), &EnvPrepareRequest{ + TaskID: "task-1", + RepositoryPath: t.TempDir(), + CheckoutBranch: "feature/missing", + Env: map[string]string{"GITHUB_TOKEN": token}, + }, nil) + if err == nil { + t.Fatal("expected Prepare() to fail") + } + for _, want := range []string{"fetch branch failed", "Could not resolve host: github.com", "pathspec 'feature/missing' did not match", "[REDACTED]"} { + if !strings.Contains(result.ErrorMessage, want) || !strings.Contains(err.Error(), want) { + t.Fatalf("expected result and error to contain %q\nresult: %s\nerror: %v", want, result.ErrorMessage, err) + } + } + for _, secret := range []string{"oauth2", token} { + if strings.Contains(result.ErrorMessage, secret) || strings.Contains(err.Error(), secret) { + t.Fatalf("checkout failure leaked %q\nresult: %s\nerror: %v", secret, result.ErrorMessage, err) + } + } +} + func TestLocalPreparer_RunsSetupScriptWithoutCheckout(t *testing.T) { isolateGitEnv(t) log := newTestLocalLogger() diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_startup_test.go b/apps/backend/internal/agent/runtime/lifecycle/manager_startup_test.go index fd9faae16c..98720175f0 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_startup_test.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_startup_test.go @@ -22,8 +22,9 @@ type blockingAgentProfileResolver struct { } type cancellableAgentProfileResolver struct { - entered chan struct{} - release chan struct{} + entered chan struct{} + cancelled chan struct{} + release chan struct{} } func (r *cancellableAgentProfileResolver) ResolveProfile( @@ -33,6 +34,8 @@ func (r *cancellableAgentProfileResolver) ResolveProfile( close(r.entered) select { case <-ctx.Done(): + close(r.cancelled) + <-r.release return nil, ctx.Err() case <-r.release: return &AgentProfileInfo{AgentID: "mock-agent"}, nil @@ -140,7 +143,7 @@ func TestTerminalInvalidationCancelsAndDrainsAdmittedStart(t *testing.T) { coordinator := activity.NewCoordinator(activity.Options{}) mgr.SetActivityCoordinator(coordinator) resolver := &cancellableAgentProfileResolver{ - entered: make(chan struct{}), release: make(chan struct{}), + entered: make(chan struct{}), cancelled: make(chan struct{}), release: make(chan struct{}), } mgr.profileResolver = resolver if err := mgr.executionStore.Add(&AgentExecution{ @@ -153,6 +156,7 @@ func TestTerminalInvalidationCancelsAndDrainsAdmittedStart(t *testing.T) { go func() { result <- mgr.StartAgentProcess(context.Background(), "exec-invalidated") }() <-resolver.entered mgr.releaseActivity(executionActivityKey("exec-invalidated")) + <-resolver.cancelled maintenance, _, maintenanceErr := coordinator.TryAcquireMaintenance(context.Background(), 0) if maintenance != nil { maintenance.Release() @@ -162,6 +166,7 @@ func TestTerminalInvalidationCancelsAndDrainsAdmittedStart(t *testing.T) { <-result t.Fatalf("maintenance error = %v, want ErrBusy until invalidated start drains", maintenanceErr) } + close(resolver.release) if err := <-result; !errors.Is(err, errExecutionActivityInvalidated) { t.Fatalf("StartAgentProcess error = %v, want claim invalidation", err) } diff --git a/apps/backend/internal/orchestrator/event_handlers_streaming.go b/apps/backend/internal/orchestrator/event_handlers_streaming.go index 81bc98f765..8d1acaf470 100644 --- a/apps/backend/internal/orchestrator/event_handlers_streaming.go +++ b/apps/backend/internal/orchestrator/event_handlers_streaming.go @@ -477,6 +477,25 @@ func (s *Service) shouldDropCompletedExecutionStreamEvent(payload *lifecycle.Age // 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 { + updated, _ := s.updateTaskSessionStateWithHook( + ctx, taskID, sessionID, nextState, errorMessage, allowWakeFromWaiting, nil, preloadedSession..., + ) + return updated +} + +// updateTaskSessionStateWithHook is updateTaskSessionState with an optional +// callback that runs only after the state CAS succeeds and before the state +// change is published. It lets a caller attach state-specific UI metadata +// without creating it when a concurrent terminal transition won the race. +func (s *Service) updateTaskSessionStateWithHook( + ctx context.Context, + taskID, sessionID string, + nextState models.TaskSessionState, + errorMessage string, + allowWakeFromWaiting bool, + onChanged func(), + preloadedSession ...*models.TaskSession, +) (*models.TaskSession, bool) { var session *models.TaskSession if len(preloadedSession) > 0 && preloadedSession[0] != nil { session = preloadedSession[0] @@ -484,25 +503,28 @@ func (s *Service) updateTaskSessionState(ctx context.Context, taskID, sessionID var err error session, err = s.repo.GetTaskSession(ctx, sessionID) if err != nil { - return nil + return nil, false } } if session.State == models.TaskSessionStateWaitingForInput && nextState == models.TaskSessionStateRunning && !allowWakeFromWaiting { - return session + return session, false } oldState := session.State switch session.State { case models.TaskSessionStateCompleted, models.TaskSessionStateFailed, models.TaskSessionStateCancelled: - return session + return session, false } if session.State == nextState { - return session + return session, false } session, authoritativeUpdatedAt, changed := s.persistTaskSessionState( ctx, sessionID, session, nextState, errorMessage, ) if !changed { - return session + return session, false + } + if onChanged != nil { + onChanged() } if authoritativeUpdatedAt == nil { s.logger.Warn("skipping session state_changed publish; could not read authoritative updated_at", @@ -520,7 +542,7 @@ func (s *Service) updateTaskSessionState(ctx context.Context, taskID, sessionID // Auto-promote another session to primary when the current primary enters a terminal state s.maybePromotePrimary(ctx, taskID, sessionID, nextState) - return session + return session, true } func (s *Service) persistTaskSessionState( @@ -590,6 +612,7 @@ func (s *Service) transitionTaskSessionState( taskID, sessionID string, nextState models.TaskSessionState, errorMessage string, + onChanged func(), ) (bool, models.TaskSessionState, error) { session, err := s.repo.GetTaskSession(ctx, sessionID) if err != nil { @@ -612,6 +635,9 @@ func (s *Service) transitionTaskSessionState( if !changed { return false, refreshed.State, nil } + if onChanged != nil { + onChanged() + } s.publishTaskSessionStateChanged( ctx, taskID, diff --git a/apps/backend/internal/orchestrator/event_handlers_streaming_test.go b/apps/backend/internal/orchestrator/event_handlers_streaming_test.go index 7d58dff853..9f75dbf254 100644 --- a/apps/backend/internal/orchestrator/event_handlers_streaming_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_streaming_test.go @@ -266,6 +266,7 @@ func TestTransitionTaskSessionStateReportsAcceptedWrite(t *testing.T) { "s1", models.TaskSessionStateCancelled, "coordinator stop", + nil, ) require.NoError(t, err) @@ -291,6 +292,7 @@ func TestTransitionTaskSessionStateReportsPersistenceFailure(t *testing.T) { "s1", models.TaskSessionStateCancelled, "coordinator stop", + nil, ) require.ErrorIs(t, err, writeFailure) diff --git a/apps/backend/internal/orchestrator/executor/executor.go b/apps/backend/internal/orchestrator/executor/executor.go index 73523cb956..b9b6f413c0 100644 --- a/apps/backend/internal/orchestrator/executor/executor.go +++ b/apps/backend/internal/orchestrator/executor/executor.go @@ -551,11 +551,16 @@ type TaskRuntimeStateReconcileFunc func(ctx context.Context, taskID, sessionID s // publish events (e.g. WebSocket notifications) alongside the DB update. type SessionStateChangeFunc func(ctx context.Context, taskID, sessionID string, state models.TaskSessionState, errorMessage string) error -// SessionStateTransitionFunc performs a guarded session-state transition and -// reports whether the requested state was accepted plus the final observed -// state. Coordinator stop uses this stricter callback so terminal races and -// persistence failures cannot be mistaken for accepted cancellation. -type SessionStateTransitionFunc func(ctx context.Context, taskID, sessionID string, state models.TaskSessionState, errorMessage string) (changed bool, finalState models.TaskSessionState, err error) +// SessionStateTransitionFunc performs a guarded session-state transition, +// invokes onChanged after persistence but before publication, and reports the +// final observed state. +type SessionStateTransitionFunc func( + ctx context.Context, + taskID, sessionID string, + state models.TaskSessionState, + errorMessage string, + onChanged func(), +) (changed bool, finalState models.TaskSessionState, err error) // SessionStartingFunc is called when the executor has prepared/resumed an // execution and needs to mark the session STARTING while preserving other diff --git a/apps/backend/internal/orchestrator/executor/executor_execute.go b/apps/backend/internal/orchestrator/executor/executor_execute.go index 377f5a9bcb..207ae5ae5e 100644 --- a/apps/backend/internal/orchestrator/executor/executor_execute.go +++ b/apps/backend/internal/orchestrator/executor/executor_execute.go @@ -391,9 +391,19 @@ func (e *Executor) transitionSessionState( taskID, sessionID string, state models.TaskSessionState, errorMessage string, +) (bool, models.TaskSessionState, error) { + return e.transitionSessionStateWithHook(ctx, taskID, sessionID, state, errorMessage, nil) +} + +func (e *Executor) transitionSessionStateWithHook( + ctx context.Context, + taskID, sessionID string, + state models.TaskSessionState, + errorMessage string, + onChanged func(), ) (bool, models.TaskSessionState, error) { if e.onSessionStateTransition != nil { - return e.onSessionStateTransition(ctx, taskID, sessionID, state, errorMessage) + return e.onSessionStateTransition(ctx, taskID, sessionID, state, errorMessage, onChanged) } current, err := e.repo.GetTaskSession(ctx, sessionID) @@ -419,6 +429,9 @@ func (e *Executor) transitionSessionState( if refreshed.State != state { return false, refreshed.State, nil } + if onChanged != nil { + onChanged() + } return true, refreshed.State, nil } @@ -976,20 +989,26 @@ func (e *Executor) handleLaunchFailure(ctx context.Context, taskID, sessionID, r e.logger.Error("failed to launch agent", zap.String("task_id", taskID), zap.Error(launchErr)) - // Call onLaunchFailed before state updates so it can set the suppressToast - // flag that updateSessionState will propagate to the frontend. + var onChanged func() if e.onLaunchFailed != nil { - e.onLaunchFailed(failCtx, taskID, sessionID, repositoryID, launchErr) + onChanged = func() { + e.onLaunchFailed(failCtx, taskID, sessionID, repositoryID, launchErr) + } } - if updateErr := e.updateSessionState(failCtx, taskID, sessionID, models.TaskSessionStateFailed, launchErr.Error()); updateErr != nil { + changed, _, updateErr := e.transitionSessionStateWithHook( + failCtx, taskID, sessionID, models.TaskSessionStateFailed, launchErr.Error(), onChanged, + ) + if updateErr != nil { e.logger.Warn("failed to mark session as failed after launch error", zap.String("session_id", sessionID), zap.Error(updateErr)) } - if updateErr := e.updateTaskState(failCtx, taskID, v1.TaskStateFailed); updateErr != nil { - e.logger.Warn("failed to mark task as failed after launch error", - zap.String("task_id", taskID), - zap.Error(updateErr)) + if changed { + if updateErr := e.updateTaskState(failCtx, taskID, v1.TaskStateFailed); updateErr != nil { + e.logger.Warn("failed to mark task as failed after launch error", + zap.String("task_id", taskID), + zap.Error(updateErr)) + } } return launchErr } diff --git a/apps/backend/internal/orchestrator/executor/executor_test.go b/apps/backend/internal/orchestrator/executor/executor_test.go index f318eb3d02..32a6a18454 100644 --- a/apps/backend/internal/orchestrator/executor/executor_test.go +++ b/apps/backend/internal/orchestrator/executor/executor_test.go @@ -1510,6 +1510,7 @@ func TestHandleAgentProcessStartFailure_CancellationDuringCallbackStopsUnclaimed string, models.TaskSessionState, string, + func(), ) (bool, models.TaskSessionState, error) { transitionCalls.Add(1) return false, models.TaskSessionStateCancelled, nil diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index 0399a98f6e..1bea8038cf 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -1543,9 +1543,24 @@ func isMissingBranchError(err error) bool { return false } msg := strings.ToLower(err.Error()) - return strings.Contains(msg, "couldn't find remote ref") || - (strings.Contains(msg, "not found locally or on remote") && strings.Contains(msg, "branch")) || - (strings.Contains(msg, "pathspec") && strings.Contains(msg, "did not match")) + if strings.Contains(msg, "couldn't find remote ref") { + return true + } + if strings.Contains(msg, "pathspec") && strings.Contains(msg, "did not match") && + !strings.Contains(msg, "fetch branch failed") { + return true + } + + const missingBranchMarker = "not found locally or on remote" + markerIndex := strings.Index(msg, missingBranchMarker) + if markerIndex < 0 || !strings.Contains(msg[:markerIndex], "branch") { + return false + } + + // The worktree layer appends the underlying fetch failure after this marker. + // Only the marker by itself is evidence that the remote branch is missing. + detail := strings.TrimSpace(msg[markerIndex+len(missingBranchMarker):]) + return detail == "" } var ( @@ -1573,6 +1588,25 @@ func extractMissingBranchName(err error) string { return "" } +const missingPRBranchRecoveryClaimKey = "missing_pr_branch_recovery_claimed" + +type failedSessionMetadataClaimer interface { + SetSessionMetadataKeyIfAbsentIfState( + ctx context.Context, + sessionID, key string, + value interface{}, + expectedState models.TaskSessionState, + ) (bool, error) +} + +type failedSessionMetadataClaimReleaser interface { + RemoveSessionMetadataKeyIfState( + ctx context.Context, + sessionID, key string, + expectedState models.TaskSessionState, + ) (bool, error) +} + func (s *Service) matchingTaskPRState(ctx context.Context, taskID, repositoryID, branch string) string { repositoryID = strings.TrimSpace(repositoryID) if s.githubService == nil || repositoryID == "" || branch == "" { @@ -1635,16 +1669,56 @@ func (s *Service) hasOnlyTaskRepository(ctx context.Context, taskID, repositoryI strings.TrimSpace(repositories[0].RepositoryID) == repositoryID } +// handleSessionLaunchFailed creates branch guidance only after the caller has +// persisted FAILED. The claim is stored on the session because prepare, start, +// and resume may race. func (s *Service) handleSessionLaunchFailed(ctx context.Context, taskID, sessionID, repositoryID string, launchErr error) { if s.messageCreator == nil || !isMissingBranchError(launchErr) { return } - + recoveryCtx := context.WithoutCancel(ctx) branch := extractMissingBranchName(launchErr) - prState := s.matchingTaskPRState(ctx, taskID, repositoryID, branch) + prState := s.matchingTaskPRState(recoveryCtx, taskID, repositoryID, branch) if prState == githubPRStateOpen { return } + claimer, ok := s.repo.(failedSessionMetadataClaimer) + if !ok { + s.logger.Warn("session repository cannot claim missing-branch recovery", + zap.String("task_id", taskID), zap.String("session_id", sessionID)) + return + } + claimed, err := claimer.SetSessionMetadataKeyIfAbsentIfState( + recoveryCtx, sessionID, missingPRBranchRecoveryClaimKey, true, models.TaskSessionStateFailed, + ) + if err != nil { + s.logger.Warn("failed to claim missing-branch recovery", + zap.String("task_id", taskID), zap.String("session_id", sessionID), zap.Error(err)) + return + } + if !claimed { + return + } + if err := s.createMissingPRBranchRecoveryMessage(recoveryCtx, taskID, sessionID, branch, prState); err != nil { + releaser, ok := s.repo.(failedSessionMetadataClaimReleaser) + if !ok { + s.logger.Warn("session repository cannot release missing-branch recovery claim after message failure", + zap.String("task_id", taskID), zap.String("session_id", sessionID)) + return + } + if _, releaseErr := releaser.RemoveSessionMetadataKeyIfState( + recoveryCtx, sessionID, missingPRBranchRecoveryClaimKey, models.TaskSessionStateFailed, + ); releaseErr != nil { + s.logger.Warn("failed to release missing-branch recovery claim after message failure", + zap.String("task_id", taskID), zap.String("session_id", sessionID), zap.Error(releaseErr)) + } + } +} + +func (s *Service) createMissingPRBranchRecoveryMessage( + ctx context.Context, + taskID, sessionID, branch, prState string, +) error { content := "Kandev couldn't fetch the requested branch from the configured repository. Verify the task's repository branch or PR link, then retry." if branch != "" { content = "Kandev couldn't fetch branch \"" + branch + "\" from the configured repository. Verify the task's repository branch or PR link, then retry." @@ -1681,7 +1755,6 @@ func (s *Service) handleSessionLaunchFailed(ctx context.Context, taskID, session }, } } - s.suppressToast.Store(sessionID, true) msgCtx, cancel := context.WithTimeout(ctx, 5*time.Second) defer cancel() if err := s.messageCreator.CreateSessionMessage( @@ -1698,7 +1771,10 @@ func (s *Service) handleSessionLaunchFailed(ctx context.Context, taskID, session zap.String("task_id", taskID), zap.String("session_id", sessionID), zap.Error(err)) + return err } + s.suppressToast.Store(sessionID, true) + return nil } // IsRunning returns true if the service is running diff --git a/apps/backend/internal/orchestrator/task_launch_failure_test.go b/apps/backend/internal/orchestrator/task_launch_failure_test.go index 9197bcefd2..d44dc8a8bc 100644 --- a/apps/backend/internal/orchestrator/task_launch_failure_test.go +++ b/apps/backend/internal/orchestrator/task_launch_failure_test.go @@ -102,6 +102,14 @@ func TestHandleSessionLaunchFailed_UnavailablePRStateCreatesNeutralFetchGuidance err := errors.New("environment preparation failed: branch \"feature/foo\" not found locally or on remote: fatal: couldn't find remote ref feature/foo") svc.handleSessionLaunchFailed(ctx, "task1", "session1", "repo-a", err) + if len(mc.sessionMessages) != 0 { + t.Fatalf("launch callback created guidance before the FAILED transition: %d messages", len(mc.sessionMessages)) + } + if err := repo.UpdateTaskSessionState(ctx, "session1", models.TaskSessionStateFailed, err.Error()); err != nil { + t.Fatalf("mark session failed: %v", err) + } + svc.handleSessionLaunchFailed(ctx, "task1", "session1", "repo-a", err) + svc.handleSessionLaunchFailed(ctx, "task1", "session1", "repo-a", err) if len(mc.sessionMessages) != 1 { t.Fatalf("expected 1 session message, got %d", len(mc.sessionMessages)) @@ -134,7 +142,7 @@ func TestHandleSessionLaunchFailed_UnavailablePRStateCreatesNeutralFetchGuidance func TestHandleSessionLaunchFailed_OpenMatchingPRDoesNotCreateMissingBranchGuidance(t *testing.T) { ctx := context.Background() repo := setupTestRepo(t) - seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateStarting) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateFailed) mc := &mockMessageCreator{} svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) @@ -157,7 +165,7 @@ func TestHandleSessionLaunchFailed_OpenMatchingPRDoesNotCreateMissingBranchGuida func TestHandleSessionLaunchFailed_SecondaryOpenPRDoesNotUsePrimaryTerminalPR(t *testing.T) { ctx := context.Background() repo := setupTestRepo(t) - seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateStarting) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateFailed) mc := &mockMessageCreator{} svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) @@ -177,7 +185,7 @@ func TestHandleSessionLaunchFailed_SecondaryOpenPRDoesNotUsePrimaryTerminalPR(t func TestHandleSessionLaunchFailed_BoundsBlockingPRStateLookup(t *testing.T) { repo := setupTestRepo(t) - seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateStarting) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateFailed) mc := &mockMessageCreator{} svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) @@ -213,7 +221,7 @@ func TestHandleSessionLaunchFailed_BoundsBlockingPRStateLookup(t *testing.T) { func TestHandleSessionLaunchFailed_ConflictingSameBranchPRStatesCreateNeutralGuidance(t *testing.T) { ctx := context.Background() repo := setupTestRepo(t) - seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateStarting) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateFailed) mc := &mockMessageCreator{} svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) @@ -244,7 +252,7 @@ func TestHandleSessionLaunchFailed_ConflictingSameBranchPRStatesCreateNeutralGui func TestHandleSessionLaunchFailed_MultipleTerminalMatchingPRsCreateNeutralGuidance(t *testing.T) { ctx := context.Background() repo := setupTestRepo(t) - seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateStarting) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateFailed) mc := &mockMessageCreator{} svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) @@ -277,7 +285,7 @@ func TestHandleSessionLaunchFailed_ClosedOrMergedMatchingPRCreatesMissingBranchG t.Run(prState, func(t *testing.T) { ctx := context.Background() repo := setupTestRepo(t) - seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateStarting) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateFailed) mc := &mockMessageCreator{} svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) @@ -310,7 +318,7 @@ func TestHandleSessionLaunchFailed_ClosedOrMergedMatchingPRCreatesMissingBranchG func TestHandleSessionLaunchFailed_UnrelatedOpenPRDoesNotSuppressNeutralGuidance(t *testing.T) { ctx := context.Background() repo := setupTestRepo(t) - seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateStarting) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateFailed) mc := &mockMessageCreator{} svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) @@ -337,7 +345,7 @@ func TestHandleSessionLaunchFailed_UnrelatedOpenPRDoesNotSuppressNeutralGuidance func TestHandleSessionLaunchFailed_ClosedPRInAnotherRepositoryCreatesNeutralGuidance(t *testing.T) { ctx := context.Background() repo := setupTestRepo(t) - seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateStarting) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateFailed) mc := &mockMessageCreator{} svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) @@ -367,7 +375,7 @@ func TestHandleSessionLaunchFailed_ClosedPRInAnotherRepositoryCreatesNeutralGuid func TestHandleSessionLaunchFailed_SingleRepositoryLegacyPRCreatesMissingBranchGuidance(t *testing.T) { ctx := context.Background() repo := setupTestRepo(t) - seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateStarting) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateFailed) if err := repo.CreateRepository(ctx, &models.Repository{ ID: "repo-a", WorkspaceID: "ws1", Name: "repo-a", SourceType: "local", }); err != nil { @@ -404,7 +412,7 @@ func TestHandleSessionLaunchFailed_SingleRepositoryLegacyPRCreatesMissingBranchG func TestHandleSessionLaunchFailed_LegacyPRForDifferentRepositoryCreatesNeutralGuidance(t *testing.T) { ctx := context.Background() repo := setupTestRepo(t) - seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateStarting) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateFailed) if err := repo.CreateRepository(ctx, &models.Repository{ ID: "repo-a", WorkspaceID: "ws1", Name: "repo-a", SourceType: "local", }); err != nil { @@ -441,7 +449,7 @@ func TestHandleSessionLaunchFailed_LegacyPRForDifferentRepositoryCreatesNeutralG func TestHandleSessionLaunchFailed_MultipleRepositoriesLegacyPRCreatesNeutralGuidance(t *testing.T) { ctx := context.Background() repo := setupTestRepo(t) - seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateStarting) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateFailed) for position, repositoryID := range []string{"repo-a", "repo-b"} { if err := repo.CreateRepository(ctx, &models.Repository{ ID: repositoryID, WorkspaceID: "ws1", Name: repositoryID, SourceType: "local", @@ -480,7 +488,7 @@ func TestHandleSessionLaunchFailed_MultipleRepositoriesLegacyPRCreatesNeutralGui func TestHandleSessionLaunchFailed_EmptyRepositoryIDCreatesNeutralGuidance(t *testing.T) { ctx := context.Background() repo := setupTestRepo(t) - seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateStarting) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateFailed) mc := &mockMessageCreator{} svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) @@ -507,10 +515,46 @@ func TestHandleSessionLaunchFailed_EmptyRepositoryIDCreatesNeutralGuidance(t *te } } +func TestHandleSessionLaunchFailed_RetriesAfterGuidanceMessageWriteFailure(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateFailed) + + writeErr := errors.New("message store unavailable") + mc := &mockMessageCreator{sessionMessageErr: writeErr} + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + svc.messageCreator = mc + err := errors.New("environment preparation failed: fatal: couldn't find remote ref feature/foo") + + svc.handleSessionLaunchFailed(ctx, "task1", "session1", "", err) + if len(mc.sessionMessages) != 0 { + t.Fatalf("failed message write persisted %d messages", len(mc.sessionMessages)) + } + if _, suppressed := svc.suppressToast.Load("session1"); suppressed { + t.Fatal("failed message write suppressed the fallback toast") + } + failed, getErr := repo.GetTaskSession(ctx, "session1") + if getErr != nil { + t.Fatalf("get failed session: %v", getErr) + } + if _, claimed := failed.Metadata[missingPRBranchRecoveryClaimKey]; claimed { + t.Fatal("failed message write left the recovery claim persisted") + } + + mc.sessionMessageErr = nil + svc.handleSessionLaunchFailed(ctx, "task1", "session1", "", err) + if len(mc.sessionMessages) != 1 { + t.Fatalf("retry created %d messages, want 1", len(mc.sessionMessages)) + } + if _, suppressed := svc.suppressToast.Load("session1"); !suppressed { + t.Fatal("successful recovery message did not suppress the fallback toast") + } +} + func TestHandleSessionLaunchFailed_IgnoresUnrelatedErrors(t *testing.T) { ctx := context.Background() repo := setupTestRepo(t) - seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateStarting) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateFailed) mc := &mockMessageCreator{} svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) @@ -522,3 +566,45 @@ func TestHandleSessionLaunchFailed_IgnoresUnrelatedErrors(t *testing.T) { t.Fatalf("expected no session messages for unrelated error, got %d", len(mc.sessionMessages)) } } + +func TestHandleSessionLaunchFailed_IgnoresMissingBranchWrapperAroundTransportError(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateFailed) + + mc := &mockMessageCreator{} + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + svc.messageCreator = mc + + err := errors.New("environment preparation failed: branch \"codex/enhance-prompt-result-delivery\" not found locally or on remote: fatal: unable to access 'https://github.com/kdlbs/kandev.git/': Could not resolve host: github.com") + if isMissingBranchError(err) { + t.Fatal("expected transport failure not to be classified as a missing PR branch") + } + + svc.handleSessionLaunchFailed(ctx, "task1", "session1", "", err) + + if len(mc.sessionMessages) != 0 { + t.Fatalf("expected no guidance message for transport failure, got %d", len(mc.sessionMessages)) + } +} + +func TestHandleSessionLaunchFailed_IgnoresPathspecAfterFetchFailure(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateFailed) + + mc := &mockMessageCreator{} + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + svc.messageCreator = mc + + err := errors.New("environment preparation failed: checkout branch: git command failed: fetch branch failed: exit status 128: fatal: unable to access 'https://github.com/kdlbs/kandev.git/': Could not resolve host: github.com\ncheckout branch failed: error: pathspec 'codex/enhance-prompt-result-delivery' did not match any file(s) known to git") + if isMissingBranchError(err) { + t.Fatal("expected pathspec after fetch failure not to be classified as a missing PR branch") + } + + svc.handleSessionLaunchFailed(ctx, "task1", "session1", "", err) + + if len(mc.sessionMessages) != 0 { + t.Fatalf("expected no guidance message after fetch failure, got %d", len(mc.sessionMessages)) + } +} diff --git a/apps/backend/internal/orchestrator/task_operations.go b/apps/backend/internal/orchestrator/task_operations.go index efb2a18d25..0e85f99279 100644 --- a/apps/backend/internal/orchestrator/task_operations.go +++ b/apps/backend/internal/orchestrator/task_operations.go @@ -257,6 +257,10 @@ func (s *Service) PrepareTaskSession(ctx context.Context, taskID string, agentPr bgCtx := context.Background() prepExec, launchErr := s.executor.LaunchPreparedSession(bgCtx, task, sessionID, executor.LaunchOptions{AgentProfileID: agentProfileID, ExecutorID: executorID, WorkflowStepID: workflowStepID}) if launchErr != nil { + // LaunchAgent failures persist FAILED in the executor. Earlier + // workspace failures return here and are recorded through the same + // session-level recovery claim. + s.handleEarlyMissingPRBranchLaunchFailure(bgCtx, taskID, sessionID, launchErr) s.logger.Warn("failed to launch workspace for prepared session (file browsing may be unavailable)", zap.String("task_id", taskID), zap.String("session_id", sessionID), @@ -476,6 +480,9 @@ func (s *Service) StartCreatedSession( } execution, err := s.executor.LaunchPreparedSession(ctx, task, sessionID, executor.LaunchOptions{AgentProfileID: effectiveProfileID, ExecutorID: executorID, Prompt: effectivePrompt, StartAgent: true, McpMode: mcpMode, Attachments: attachments}) if err != nil { + // The executor persists LaunchAgent failures. Cover earlier prepared-session + // failures here; the session-level claim makes either completion order safe. + s.handleEarlyMissingPRBranchLaunchFailure(ctx, taskID, sessionID, err) return nil, err } @@ -492,6 +499,56 @@ func (s *Service) StartCreatedSession( return execution, nil } +// handleEarlyMissingPRBranchLaunchFailure covers failures that occur before +// Executor reaches AgentManager.LaunchAgent. Those failures do not run the +// executor's launch-failure bookkeeping, so this fallback owns the terminal +// transition and lets the shared persisted claim create guidance once. +func (s *Service) handleEarlyMissingPRBranchLaunchFailure( + ctx context.Context, + taskID, sessionID string, + launchErr error, +) { + if !isMissingBranchError(launchErr) { + return + } + s.recordSessionLaunchFailure(ctx, taskID, sessionID, launchErr) +} + +// recordSessionLaunchFailure transitions a still-active session to FAILED, +// attaches missing-branch guidance only after that CAS succeeds, and updates +// the task only while the same failed session still owns it. +func (s *Service) recordSessionLaunchFailure(ctx context.Context, taskID, sessionID string, launchErr error, preloadedSession ...*models.TaskSession) { + _, changed := s.updateTaskSessionStateWithHook( + ctx, taskID, sessionID, models.TaskSessionStateFailed, launchErr.Error(), false, + func() { s.handleSessionLaunchFailed(ctx, taskID, sessionID, "", launchErr) }, preloadedSession..., + ) + if !changed { + // A resume may begin from an already FAILED session. Its state CAS is + // intentionally a no-op, but the persisted claim still lets this newly + // classified failure publish guidance exactly once. The claim itself is + // state-guarded, so a concurrently cancelled/completed session cannot + // receive stale recovery UI. + if len(preloadedSession) > 0 && preloadedSession[0] != nil && preloadedSession[0].State == models.TaskSessionStateFailed { + s.handleSessionLaunchFailed(ctx, taskID, sessionID, "", launchErr) + } + return + } + updated, err := s.taskRepo.UpdateTaskStateIfSessionState( + ctx, taskID, sessionID, models.TaskSessionStateFailed, v1.TaskStateFailed, + ) + if err != nil { + s.logger.Warn("failed to update task state to FAILED after early launch error", + zap.String("task_id", taskID), + zap.String("session_id", sessionID), + zap.Error(err)) + return + } + if !updated { + return + } + s.processParentChildrenCompletedForTaskState(ctx, taskID, v1.TaskStateFailed) +} + func (s *Service) scheduleTaskForSession(ctx context.Context, taskID, sessionID string) error { s.taskRuntimeStateMu.Lock() defer s.taskRuntimeStateMu.Unlock() @@ -1205,15 +1262,10 @@ func (s *Service) ResumeTaskSession(ctx context.Context, taskID, sessionID strin // away), the SessionStateFailed and TaskStateFailed updates would // themselves fail with "context canceled" and leave the task stuck // looking "running" forever. - s.updateTaskSessionState(resumeCtx, taskID, sessionID, models.TaskSessionStateFailed, err.Error(), false, session) - if stateErr := s.taskRepo.UpdateTaskState(resumeCtx, taskID, v1.TaskStateFailed); stateErr != nil { - s.logger.Warn("failed to update task state to FAILED after resume error", - zap.String("task_id", taskID), - zap.String("session_id", sessionID), - zap.Error(stateErr)) - } else { - s.processParentChildrenCompletedForTaskState(resumeCtx, taskID, v1.TaskStateFailed) - } + // ResumeSession launches the workspace directly rather than through + // LaunchPreparedSession. Record this failure with the same state CAS, + // persisted recovery claim, and archive-safe task CAS as early launch. + s.recordSessionLaunchFailure(resumeCtx, taskID, sessionID, err, session) return nil, err } } diff --git a/apps/backend/internal/orchestrator/task_operations_test.go b/apps/backend/internal/orchestrator/task_operations_test.go index c014c037fa..a1da9d7925 100644 --- a/apps/backend/internal/orchestrator/task_operations_test.go +++ b/apps/backend/internal/orchestrator/task_operations_test.go @@ -15,6 +15,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/orchestrator/dto" "github.com/kandev/kandev/internal/orchestrator/executor" "github.com/kandev/kandev/internal/orchestrator/messagequeue" @@ -67,6 +68,21 @@ func seedTaskAndSession(t *testing.T, repo *sqliterepo.Repository, taskID, sessi } } +// taskEnvironmentFailureRepo injects an error after session creation but before +// Executor reaches AgentManager.LaunchAgent. This models workspace-preparation +// failures, which do not trigger the executor's launch-failure callback. +type taskEnvironmentFailureRepo struct { + *sqliterepo.Repository + err error + called chan struct{} + once sync.Once +} + +func (r *taskEnvironmentFailureRepo) GetTaskEnvironmentByTaskID(_ context.Context, _ string) (*models.TaskEnvironment, error) { + r.once.Do(func() { close(r.called) }) + return nil, r.err +} + func TestCreateStartSession_KanbanRunnerCreatesDistinctSession(t *testing.T) { ctx := context.Background() repo := setupTestRepo(t) @@ -2251,6 +2267,421 @@ func TestStartCreatedSession_NotInCreatedState(t *testing.T) { } } +func TestStartCreatedSession_MissingRemoteRefCreatesNeutralRecoveryMessage(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateCreated) + + launchErr := errors.New("environment preparation failed: fatal: couldn't find remote ref feature/deleted-pr") + agentMgr := &mockAgentManager{ + launchAgentFunc: func(_ context.Context, _ *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + return nil, launchErr + }, + } + taskRepo := newMockTaskRepo() + taskRepo.tasks["task1"] = &v1.Task{ + ID: "task1", + Title: "Test Task", + Description: "start the task", + State: v1.TaskStateInProgress, + } + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentMgr) + messages := &mockMessageCreator{} + svc.messageCreator = messages + svc.executor.SetOnLaunchFailed(svc.handleSessionLaunchFailed) + svc.executor.SetOnSessionStateChange(func(callbackCtx context.Context, callbackTaskID, callbackSessionID string, state models.TaskSessionState, errorMessage string) error { + svc.updateTaskSessionState(callbackCtx, callbackTaskID, callbackSessionID, state, errorMessage, true) + return nil + }) + + _, err := svc.StartCreatedSession(ctx, "task1", "session1", "profile1", "start the task", true, false, true, nil, nil) + if !errors.Is(err, launchErr) { + t.Fatalf("StartCreatedSession error = %v, want %v", err, launchErr) + } + + if len(messages.sessionMessages) != 1 { + t.Fatalf("expected one recovery message, got %d", len(messages.sessionMessages)) + } + message := messages.sessionMessages[0] + if message.metadata["failure_kind"] != "branch_fetch_failed" { + t.Fatalf("failure_kind = %#v, want branch_fetch_failed", message.metadata["failure_kind"]) + } + if _, ok := message.metadata["actions"]; ok { + t.Fatalf("expected neutral guidance without archive/delete actions, got %#v", message.metadata["actions"]) + } +} + +func TestStartCreatedSession_MissingRemoteRefDoesNotDuplicateExecutorRecoveryMessage(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateCreated) + + launchErr := errors.New("environment preparation failed: fatal: couldn't find remote ref feature/deleted-pr") + agentMgr := &mockAgentManager{ + launchAgentFunc: func(_ context.Context, _ *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + return nil, launchErr + }, + } + taskRepo := newMockTaskRepo() + taskRepo.tasks["task1"] = &v1.Task{ + ID: "task1", + Title: "Test Task", + Description: "start the task", + State: v1.TaskStateInProgress, + } + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentMgr) + messages := &mockMessageCreator{} + svc.messageCreator = messages + svc.eventBus = bus.NewMemoryEventBus(testLogger()) + svc.executor.SetOnLaunchFailed(svc.handleSessionLaunchFailed) + // Production routes terminal failures through the strict transition callback + // so recovery guidance can set suppressToast before state_changed publishes it. + svc.executor.SetOnSessionStateTransition(svc.transitionTaskSessionState) + + _, err := svc.StartCreatedSession(ctx, "task1", "session1", "profile1", "start the task", true, false, true, nil, nil) + if !errors.Is(err, launchErr) { + t.Fatalf("StartCreatedSession error = %v, want %v", err, launchErr) + } + if len(messages.sessionMessages) != 1 { + t.Fatalf("expected exactly one recovery message, got %d", len(messages.sessionMessages)) + } + if _, suppressed := svc.suppressToast.Load("session1"); suppressed { + t.Fatal("state-change publishing did not consume the toast-suppression marker") + } +} + +func TestPrepareTaskSession_WorkspaceLaunchFailureRecovery(t *testing.T) { + const ( + taskID = "task1" + ) + + newTask := func() *v1.Task { + return &v1.Task{ + ID: taskID, + WorkspaceID: "ws1", + WorkflowID: "wf1", + Title: "Test Task", + Description: "prepare the workspace", + State: v1.TaskStateInProgress, + } + } + + t.Run("early missing remote ref creates one neutral recovery message", func(t *testing.T) { + baseRepo := setupTestRepo(t) + seedTaskAndSession(t, baseRepo, taskID, "existing-session", models.TaskSessionStateCreated) + failureRepo := &taskEnvironmentFailureRepo{ + Repository: baseRepo, + err: errors.New("environment preparation failed: fatal: couldn't find remote ref feature/deleted-pr"), + called: make(chan struct{}), + } + taskRepo := newMockTaskRepo() + taskRepo.tasks[taskID] = newTask() + agentMgr := &mockAgentManager{} + exec := executor.NewExecutor(agentMgr, failureRepo, testLogger(), executor.ExecutorConfig{}) + svc := &Service{ + logger: testLogger(), + repo: failureRepo, + workflowStepGetter: newMockStepGetter(), + taskRepo: taskRepo, + agentManager: agentMgr, + executor: exec, + messageQueue: messagequeue.NewServiceMemory(testLogger()), + scheduler: scheduler.NewScheduler(queue.NewTaskQueue(1), exec, taskRepo, testLogger(), scheduler.SchedulerConfig{}), + } + messages := &mockMessageCreator{sessionMessageDone: make(chan struct{})} + svc.messageCreator = messages + svc.eventBus = bus.NewMemoryEventBus(testLogger()) + + sessionID, err := svc.PrepareTaskSession(context.Background(), taskID, "profile1", "", "", "", true) + if err != nil { + t.Fatalf("PrepareTaskSession: %v", err) + } + + select { + case <-messages.sessionMessageDone: + case <-time.After(time.Second): + t.Fatal("timed out waiting for missing-branch recovery message") + } + if len(messages.sessionMessages) != 1 { + t.Fatalf("expected exactly one recovery message, got %d", len(messages.sessionMessages)) + } + message := messages.sessionMessages[0] + if message.sessionID != sessionID { + t.Fatalf("recovery message session = %q, want %q", message.sessionID, sessionID) + } + if message.metadata["failure_kind"] != "branch_fetch_failed" { + t.Fatalf("failure_kind = %#v, want branch_fetch_failed", message.metadata["failure_kind"]) + } + if _, ok := message.metadata["actions"]; ok { + t.Fatalf("expected neutral guidance without archive/delete actions, got %#v", message.metadata["actions"]) + } + require.Eventually(t, func() bool { + failedSession, getErr := baseRepo.GetTaskSession(context.Background(), sessionID) + return getErr == nil && failedSession.State == models.TaskSessionStateFailed + }, time.Second, 10*time.Millisecond, "expected early launch failure to mark the session FAILED") + require.Eventually(t, func() bool { + taskRepo.mu.Lock() + defer taskRepo.mu.Unlock() + return taskRepo.updatedStates[taskID] == v1.TaskStateFailed + }, time.Second, 10*time.Millisecond, "expected early launch failure to mark the task FAILED") + if _, suppressed := svc.suppressToast.Load(sessionID); suppressed { + t.Fatal("missing-branch recovery left a stale toast-suppression marker") + } + }) + + t.Run("transport failure does not create missing-branch recovery", func(t *testing.T) { + baseRepo := setupTestRepo(t) + seedTaskAndSession(t, baseRepo, taskID, "existing-session", models.TaskSessionStateCreated) + failureRepo := &taskEnvironmentFailureRepo{ + Repository: baseRepo, + err: errors.New("environment preparation failed: branch \"feature/deleted-pr\" not found locally or on remote: fatal: unable to access 'https://github.com/kdlbs/kandev.git/': Could not resolve host: github.com"), + called: make(chan struct{}), + } + taskRepo := newMockTaskRepo() + taskRepo.tasks[taskID] = newTask() + agentMgr := &mockAgentManager{} + exec := executor.NewExecutor(agentMgr, failureRepo, testLogger(), executor.ExecutorConfig{}) + svc := &Service{ + logger: testLogger(), + repo: failureRepo, + workflowStepGetter: newMockStepGetter(), + taskRepo: taskRepo, + agentManager: agentMgr, + executor: exec, + messageQueue: messagequeue.NewServiceMemory(testLogger()), + scheduler: scheduler.NewScheduler(queue.NewTaskQueue(1), exec, taskRepo, testLogger(), scheduler.SchedulerConfig{}), + } + messages := &mockMessageCreator{sessionMessageDone: make(chan struct{})} + svc.messageCreator = messages + + if _, err := svc.PrepareTaskSession(context.Background(), taskID, "profile1", "", "", "", true); err != nil { + t.Fatalf("PrepareTaskSession: %v", err) + } + select { + case <-failureRepo.called: + case <-time.After(time.Second): + t.Fatal("timed out waiting for workspace launch") + } + select { + case <-messages.sessionMessageDone: + t.Fatal("transport failure created missing-branch recovery message") + case <-time.After(100 * time.Millisecond): + } + }) + + t.Run("executor callback recovery is not duplicated", func(t *testing.T) { + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, taskID, "existing-session", models.TaskSessionStateCreated) + taskRepo := newMockTaskRepo() + taskRepo.tasks[taskID] = newTask() + launchErr := errors.New("environment preparation failed: fatal: couldn't find remote ref feature/deleted-pr") + agentMgr := &mockAgentManager{ + launchAgentFunc: func(context.Context, *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + return nil, launchErr + }, + } + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentMgr) + svc.executor.SetOnLaunchFailed(svc.handleSessionLaunchFailed) + svc.eventBus = bus.NewMemoryEventBus(testLogger()) + svc.executor.SetOnSessionStateChange(func(callbackCtx context.Context, callbackTaskID, callbackSessionID string, state models.TaskSessionState, errorMessage string) error { + svc.updateTaskSessionState(callbackCtx, callbackTaskID, callbackSessionID, state, errorMessage, true) + return nil + }) + messages := &mockMessageCreator{sessionMessageDone: make(chan struct{})} + svc.messageCreator = messages + + sessionID, err := svc.PrepareTaskSession(context.Background(), taskID, "profile1", "", "", "", true) + if err != nil { + t.Fatalf("PrepareTaskSession: %v", err) + } + select { + case <-messages.sessionMessageDone: + case <-time.After(time.Second): + t.Fatal("timed out waiting for executor recovery message") + } + require.Eventually(t, func() bool { + session, getErr := repo.GetTaskSession(context.Background(), sessionID) + return getErr == nil && session.State == models.TaskSessionStateFailed + }, time.Second, 10*time.Millisecond, "expected failed state after workspace launch error") + if len(messages.sessionMessages) != 1 { + t.Fatalf("expected exactly one recovery message, got %d", len(messages.sessionMessages)) + } + }) +} + +func TestHandleEarlyMissingPRBranchLaunchFailure_SkipsTaskFailureWhenSessionTransitionLoses(t *testing.T) { + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateCancelled) + taskRepo := newMockTaskRepo() + taskRepo.tasks["task1"] = &v1.Task{ID: "task1", State: v1.TaskStateInProgress} + svc := createTestService(repo, newMockStepGetter(), taskRepo) + messages := &mockMessageCreator{} + svc.messageCreator = messages + svc.eventBus = bus.NewMemoryEventBus(testLogger()) + + svc.handleEarlyMissingPRBranchLaunchFailure( + context.Background(), "task1", "session1", + errors.New("fatal: couldn't find remote ref feature/deleted-pr"), + ) + + if len(messages.sessionMessages) != 0 { + t.Fatalf("terminal-race loser created %d recovery messages", len(messages.sessionMessages)) + } + taskRepo.mu.Lock() + defer taskRepo.mu.Unlock() + if taskRepo.stateWrites["task1"] != 0 { + t.Fatalf("terminal-race loser wrote task FAILED %d times", taskRepo.stateWrites["task1"]) + } + if got := taskRepo.tasks["task1"].State; got != v1.TaskStateInProgress { + t.Fatalf("task state = %s, want IN_PROGRESS", got) + } +} + +func TestHandleEarlyMissingPRBranchLaunchFailure_ArchiveSafeTaskWrite(t *testing.T) { + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateCreated) + taskRepo := newMockTaskRepo() + taskRepo.tasks["task1"] = &v1.Task{ID: "task1", State: v1.TaskStateInProgress} + // Model ArchiveTask winning after the session CAS but before the task-state + // CAS: UpdateTaskStateIfSessionState must decline the write. + taskRepo.updateIfSessionState = func(context.Context, string, string, models.TaskSessionState, v1.TaskState) (bool, error) { + return false, nil + } + svc := createTestService(repo, newMockStepGetter(), taskRepo) + messages := &mockMessageCreator{} + svc.messageCreator = messages + svc.eventBus = bus.NewMemoryEventBus(testLogger()) + + svc.handleEarlyMissingPRBranchLaunchFailure( + context.Background(), "task1", "session1", + errors.New("fatal: couldn't find remote ref feature/deleted-pr"), + ) + + failedSession, err := repo.GetTaskSession(context.Background(), "session1") + if err != nil { + t.Fatalf("GetTaskSession: %v", err) + } + if failedSession.State != models.TaskSessionStateFailed { + t.Fatalf("session state = %s, want FAILED", failedSession.State) + } + taskRepo.mu.Lock() + defer taskRepo.mu.Unlock() + if taskRepo.stateWrites["task1"] != 0 || taskRepo.unconditionalWrites["task1"] != 0 { + t.Fatalf("archive-safe task CAS was bypassed: state writes=%d unconditional=%d", taskRepo.stateWrites["task1"], taskRepo.unconditionalWrites["task1"]) + } + if got := taskRepo.tasks["task1"].State; got != v1.TaskStateInProgress { + t.Fatalf("archived task state = %s, want IN_PROGRESS", got) + } +} + +func TestHandleEarlyMissingPRBranchLaunchFailure_ConcurrentLaunchesCreateOneMessage(t *testing.T) { + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateCreated) + taskRepo := newMockTaskRepo() + taskRepo.tasks["task1"] = &v1.Task{ID: "task1", State: v1.TaskStateInProgress} + svc := createTestService(repo, newMockStepGetter(), taskRepo) + messages := &mockMessageCreator{} + svc.messageCreator = messages + svc.eventBus = bus.NewMemoryEventBus(testLogger()) + + launchErr := errors.New("fatal: couldn't find remote ref feature/deleted-pr") + start := make(chan struct{}) + var ready sync.WaitGroup + var done sync.WaitGroup + ready.Add(2) + done.Add(2) + for range 2 { + go func() { + defer done.Done() + ready.Done() + <-start + svc.handleEarlyMissingPRBranchLaunchFailure(context.Background(), "task1", "session1", launchErr) + }() + } + ready.Wait() + close(start) + done.Wait() + + messages.mu.Lock() + messageCount := len(messages.sessionMessages) + messages.mu.Unlock() + if messageCount != 1 { + t.Fatalf("recovery message count = %d, want 1", messageCount) + } + taskRepo.mu.Lock() + defer taskRepo.mu.Unlock() + if taskRepo.stateWrites["task1"] != 1 { + t.Fatalf("task FAILED writes = %d, want 1", taskRepo.stateWrites["task1"]) + } +} + +func TestPrepareAndStartCreatedSession_MissingRemoteRefClaimsRecoveryOnce(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, "task1", "existing-session", models.TaskSessionStateCreated) + taskRepo := newMockTaskRepo() + taskRepo.tasks["task1"] = &v1.Task{ + ID: "task1", WorkspaceID: "ws1", WorkflowID: "wf1", Title: "Test Task", Description: "start task", State: v1.TaskStateInProgress, + } + launchEntered := make(chan struct{}) + var launchEnteredOnce sync.Once + releaseLaunch := make(chan struct{}) + launchErr := errors.New("environment preparation failed: fatal: couldn't find remote ref feature/deleted-pr") + agentMgr := &mockAgentManager{ + launchAgentFunc: func(context.Context, *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + launchEnteredOnce.Do(func() { close(launchEntered) }) + <-releaseLaunch + return nil, launchErr + }, + } + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentMgr) + svc.eventBus = bus.NewMemoryEventBus(testLogger()) + svc.executor.SetOnSessionStateTransition(svc.transitionTaskSessionState) + svc.executor.SetOnLaunchFailed(svc.handleSessionLaunchFailed) + messages := &mockMessageCreator{sessionMessageDone: make(chan struct{})} + svc.messageCreator = messages + + sessionID, err := svc.PrepareTaskSession(ctx, "task1", "profile1", "", "", "", true) + if err != nil { + t.Fatalf("PrepareTaskSession: %v", err) + } + select { + case <-launchEntered: + case <-time.After(time.Second): + t.Fatal("prepared launch did not reach agent launch") + } + + startDone := make(chan error, 1) + go func() { + _, startErr := svc.StartCreatedSession(ctx, "task1", sessionID, "profile1", "start task", true, false, false, nil, nil) + startDone <- startErr + }() + close(releaseLaunch) + select { + case <-startDone: + case <-time.After(time.Second): + t.Fatal("StartCreatedSession did not settle after prepared launch failed") + } + select { + case <-messages.sessionMessageDone: + case <-time.After(time.Second): + t.Fatal("timed out waiting for missing-branch recovery message") + } + + failed, err := repo.GetTaskSession(ctx, sessionID) + if err != nil { + t.Fatalf("GetTaskSession: %v", err) + } + if failed.State != models.TaskSessionStateFailed { + t.Fatalf("session state = %s, want FAILED", failed.State) + } + messages.mu.Lock() + defer messages.mu.Unlock() + if len(messages.sessionMessages) != 1 { + t.Fatalf("recovery message count = %d, want 1", len(messages.sessionMessages)) + } +} + func TestStartCreatedSession_WorkflowOverridePromotesPreparedWhenTaskHasNoPrimary(t *testing.T) { ctx := context.Background() repo := setupTestRepo(t) @@ -2600,8 +3031,12 @@ func TestIssue1884_StepProfileSignalGateStaysInTaskMode(t *testing.T) { // mockMessageCreator implements MessageCreator for testing. // Only CreateUserMessage is tracked; all other methods are no-op stubs. type mockMessageCreator struct { + mu sync.Mutex userMessages []mockUserMessage sessionMessages []mockSessionMessage + sessionMessageDone chan struct{} + sessionMessageOnce sync.Once + sessionMessageErr error agentMessages []mockAgentMessage agentMessageWrites int agentStreamWrites int @@ -2651,6 +3086,11 @@ func (m *mockMessageCreator) UpdateToolCallMessage(context.Context, string, stri } func (m *mockMessageCreator) CreateSessionMessage(_ context.Context, taskID, content, sessionID, messageType, turnID string, metadata map[string]interface{}, requestsInput bool) error { + m.mu.Lock() + defer m.mu.Unlock() + if m.sessionMessageErr != nil { + return m.sessionMessageErr + } m.sessionMessages = append(m.sessionMessages, mockSessionMessage{ taskID: taskID, content: content, @@ -2660,6 +3100,9 @@ func (m *mockMessageCreator) CreateSessionMessage(_ context.Context, taskID, con metadata: metadata, requestsInput: requestsInput, }) + if m.sessionMessageDone != nil { + m.sessionMessageOnce.Do(func() { close(m.sessionMessageDone) }) + } return nil } @@ -2697,7 +3140,7 @@ func (m *mockMessageCreator) InvalidateModelCache(string) {} func TestBackfillInitialUserMessageIfMissing_RecordsWhenSessionEmpty(t *testing.T) { ctx := context.Background() repo := setupTestRepo(t) - seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateFailed) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateWaitingForInput) mc := &mockMessageCreator{} svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) @@ -3378,6 +3821,58 @@ func TestResumeTaskSession_ArchivedDuringLaunch(t *testing.T) { } } +func TestResumeTaskSession_FailureTaskWriteIsConditionalOnFailedSession(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + taskRepo := newMockTaskRepo() + taskRepo.tasks["task1"] = &v1.Task{ID: "task1", State: v1.TaskStateInProgress} + // Model ArchiveTask winning after the session FAILED CAS but before the + // task write. Resume must not use its legacy unconditional task update. + taskRepo.updateIfSessionState = func(context.Context, string, string, models.TaskSessionState, v1.TaskState) (bool, error) { + return false, nil + } + launchErr := errors.New("resume workspace failed") + agentMgr := &mockAgentManager{ + launchAgentFunc: func(context.Context, *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + return nil, launchErr + }, + } + svc := createTestServiceWithAgent(repo, newMockStepGetter(), taskRepo, agentMgr) + svc.executor = executor.NewExecutor(agentMgr, repo, testLogger(), executor.ExecutorConfig{}) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateWaitingForInput) + session, err := repo.GetTaskSession(ctx, "session1") + if err != nil { + t.Fatalf("GetTaskSession: %v", err) + } + session.AgentProfileID = "profile-1" + if err := repo.UpdateTaskSession(ctx, session); err != nil { + t.Fatalf("UpdateTaskSession: %v", err) + } + now := time.Now().UTC() + if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{ + ID: "resume-archive-race", SessionID: "session1", TaskID: "task1", CreatedAt: now, UpdatedAt: now, + }); err != nil { + t.Fatalf("UpsertExecutorRunning: %v", err) + } + + _, err = svc.ResumeTaskSession(ctx, "task1", "session1") + if !errors.Is(err, launchErr) { + t.Fatalf("ResumeTaskSession error = %v, want %v", err, launchErr) + } + failed, err := repo.GetTaskSession(ctx, "session1") + if err != nil { + t.Fatalf("GetTaskSession after failure: %v", err) + } + if failed.State != models.TaskSessionStateFailed { + t.Fatalf("session state = %s, want FAILED", failed.State) + } + taskRepo.mu.Lock() + defer taskRepo.mu.Unlock() + if taskRepo.stateWrites["task1"] != 0 || taskRepo.unconditionalWrites["task1"] != 0 { + t.Fatalf("resume bypassed conditional task write: conditional=%d unconditional=%d", taskRepo.stateWrites["task1"], taskRepo.unconditionalWrites["task1"]) + } +} + // TestResumeTaskSession_FailedKeepsResumeToken verifies that resuming a FAILED // session preserves the ACP resume token so the relaunched agent restores the // prior conversation via ACP session/load (for native-resume agents). @@ -3552,6 +4047,93 @@ func TestResumeTaskSession_FailedStateWriteSurvivesCancelledCallerCtx(t *testing } } +func TestResumeTaskSession_AlreadyFailedMissingRemoteRefCreatesNeutralRecoveryMessage(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + taskRepo := newMockTaskRepo() + launchErr := errors.New("environment preparation failed: fatal: couldn't find remote ref feature/foo") + agentMgr := &mockAgentManager{ + launchAgentFunc: func(_ context.Context, _ *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + return nil, launchErr + }, + } + svc := createTestServiceWithAgent(repo, newMockStepGetter(), taskRepo, agentMgr) + svc.executor = executor.NewExecutor(agentMgr, repo, testLogger(), executor.ExecutorConfig{}) + svc.messageCreator = &mockMessageCreator{} + + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateFailed) + session, err := repo.GetTaskSession(ctx, "session1") + if err != nil { + t.Fatalf("get seeded session: %v", err) + } + session.AgentProfileID = "profile-1" + if err := repo.UpdateTaskSession(ctx, session); err != nil { + t.Fatalf("update seeded session: %v", err) + } + now := time.Now().UTC() + if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{ + ID: "resume-missing-ref", SessionID: "session1", TaskID: "task1", CreatedAt: now, UpdatedAt: now, + }); err != nil { + t.Fatalf("seed executor record: %v", err) + } + + _, err = svc.ResumeTaskSession(ctx, "task1", "session1") + if !errors.Is(err, launchErr) { + t.Fatalf("ResumeTaskSession error = %v, want %v", err, launchErr) + } + + messages := svc.messageCreator.(*mockMessageCreator).sessionMessages + if len(messages) != 1 { + t.Fatalf("expected one recovery message, got %d", len(messages)) + } + message := messages[0] + if message.metadata["failure_kind"] != "branch_fetch_failed" { + t.Fatalf("failure_kind = %#v, want branch_fetch_failed", message.metadata["failure_kind"]) + } + if _, ok := message.metadata["actions"]; ok { + t.Fatalf("expected no destructive actions without a repository-scoped PR match, got %#v", message.metadata["actions"]) + } + taskRepo.mu.Lock() + defer taskRepo.mu.Unlock() + if taskRepo.stateWrites["task1"] != 0 { + t.Fatalf("already failed resume rewrote task state %d times", taskRepo.stateWrites["task1"]) + } +} + +func TestResumeTaskSession_UnrelatedLaunchFailureDoesNotCreateRecoveryMessage(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + taskRepo := newMockTaskRepo() + launchErr := errors.New("failed to launch container") + agentMgr := &mockAgentManager{ + launchAgentFunc: func(_ context.Context, _ *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + return nil, launchErr + }, + } + svc := createTestServiceWithAgent(repo, newMockStepGetter(), taskRepo, agentMgr) + svc.executor = executor.NewExecutor(agentMgr, repo, testLogger(), executor.ExecutorConfig{}) + messageCreator := &mockMessageCreator{} + svc.messageCreator = messageCreator + + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateFailed) + session, err := repo.GetTaskSession(ctx, "session1") + if err != nil { + t.Fatalf("get seeded session: %v", err) + } + session.AgentProfileID = "profile-1" + if err := repo.UpdateTaskSession(ctx, session); err != nil { + t.Fatalf("update seeded session: %v", err) + } + + _, err = svc.ResumeTaskSession(ctx, "task1", "session1") + if !errors.Is(err, launchErr) { + t.Fatalf("ResumeTaskSession error = %v, want %v", err, launchErr) + } + if len(messageCreator.sessionMessages) != 0 { + t.Fatalf("expected no recovery message, got %#v", messageCreator.sessionMessages) + } +} + // --- CompleteTask --- func TestCompleteTask_UpdatesTaskState(t *testing.T) { diff --git a/apps/backend/internal/task/repository/sqlite/session.go b/apps/backend/internal/task/repository/sqlite/session.go index 97955326ef..f49338d10d 100644 --- a/apps/backend/internal/task/repository/sqlite/session.go +++ b/apps/backend/internal/task/repository/sqlite/session.go @@ -952,6 +952,56 @@ func (r *Repository) SetSessionMetadataKeyIfAbsent( return rows > 0, nil } +// SetSessionMetadataKeyIfAbsentIfState atomically claims a metadata key only +// while the session remains in expectedState. It is used when a terminal +// transition owns a one-time side effect that must not be emitted by a stale +// launch callback. +func (r *Repository) SetSessionMetadataKeyIfAbsentIfState( + ctx context.Context, + sessionID, key string, + value interface{}, + expectedState models.TaskSessionState, +) (bool, error) { + valueJSON, err := json.Marshal(value) + if err != nil { + return false, fmt.Errorf("failed to serialize metadata value: %w", err) + } + now := time.Now().UTC() + driver := r.db.DriverName() + path := key + if !dialect.IsPostgres(driver) { + path = "$." + key + } + result, err := r.db.ExecContext(ctx, r.db.Rebind(setSessionMetadataKeyIfAbsentIfStateQuery(driver)), path, string(valueJSON), now, sessionID, path, expectedState) + if err != nil { + return false, err + } + rows, _ := result.RowsAffected() + return rows > 0, nil +} + +// RemoveSessionMetadataKeyIfState removes a claimed metadata key only while +// the session remains in expectedState. It releases one-time side-effect +// claims when their downstream write fails, allowing a later retry. +func (r *Repository) RemoveSessionMetadataKeyIfState( + ctx context.Context, + sessionID, key string, + expectedState models.TaskSessionState, +) (bool, error) { + now := time.Now().UTC() + driver := r.db.DriverName() + path := key + if !dialect.IsPostgres(driver) { + path = "$." + key + } + result, err := r.db.ExecContext(ctx, r.db.Rebind(removeSessionMetadataKeyIfStateQuery(driver)), path, now, sessionID, path, expectedState) + if err != nil { + return false, err + } + rows, _ := result.RowsAffected() + return rows > 0, nil +} + func setSessionMetadataKeyIfAbsentQuery(driver string) string { if dialect.IsPostgres(driver) { return ` @@ -979,6 +1029,37 @@ func setSessionMetadataKeyIfAbsentQuery(driver string) string { ` } +func setSessionMetadataKeyIfAbsentIfStateQuery(driver string) string { + return setSessionMetadataKeyIfAbsentQuery(driver) + " AND state = ?" +} + +func removeSessionMetadataKeyIfStateQuery(driver string) string { + if dialect.IsPostgres(driver) { + return ` + UPDATE task_sessions + SET metadata = ( + CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}'::jsonb ELSE metadata::jsonb END + #- ARRAY[?]::text[] + )::text, + updated_at = ? + WHERE id = ? + AND jsonb_extract_path( + CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}'::jsonb ELSE metadata::jsonb END, + ? + ) IS NOT NULL + AND state = ? + ` + } + return ` + UPDATE task_sessions + SET metadata = json_remove(CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}' ELSE metadata END, ?), + updated_at = ? + WHERE id = ? + AND json_type(CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}' ELSE metadata END, ?) IS NOT NULL + AND state = ? + ` +} + // SetSessionACPSessionID mirrors the agent's ACP session id into the session's // "acp" metadata map as a single atomic UPDATE. json_patch merges the sub-key, // so keys session_info already wrote (title, ...) survive without a diff --git a/apps/backend/internal/task/repository/sqlite/session_test.go b/apps/backend/internal/task/repository/sqlite/session_test.go index 0755fc7ab9..f2f434bdd0 100644 --- a/apps/backend/internal/task/repository/sqlite/session_test.go +++ b/apps/backend/internal/task/repository/sqlite/session_test.go @@ -223,6 +223,78 @@ func TestSetSessionMetadataKeyIfAbsentSQLiteIsWriteOnce(t *testing.T) { } } +func TestSetSessionMetadataKeyIfAbsentIfStateSQLiteRequiresFailedSession(t *testing.T) { + repo := newRepoForSessionTests(t) + seedForMsgTest(t, repo, "task-recovery", "session-recovery", "turn-recovery") + ctx := context.Background() + + stored, err := repo.SetSessionMetadataKeyIfAbsentIfState( + ctx, "session-recovery", "missing_pr_branch_recovery_claimed", true, models.TaskSessionStateFailed, + ) + if err != nil { + t.Fatalf("claim before failed state: %v", err) + } + if stored { + t.Fatal("claim should reject a session that is not FAILED") + } + if err := repo.UpdateTaskSessionState(ctx, "session-recovery", models.TaskSessionStateFailed, "missing branch"); err != nil { + t.Fatalf("mark failed: %v", err) + } + stored, err = repo.SetSessionMetadataKeyIfAbsentIfState( + ctx, "session-recovery", "missing_pr_branch_recovery_claimed", true, models.TaskSessionStateFailed, + ) + if err != nil { + t.Fatalf("claim after failed state: %v", err) + } + if !stored { + t.Fatal("claim should store after FAILED state") + } + stored, err = repo.SetSessionMetadataKeyIfAbsentIfState( + ctx, "session-recovery", "missing_pr_branch_recovery_claimed", true, models.TaskSessionStateFailed, + ) + if err != nil { + t.Fatalf("repeat claim: %v", err) + } + if stored { + t.Fatal("repeat claim should not overwrite") + } +} + +func TestRemoveSessionMetadataKeyIfStateSQLiteRequiresFailedSession(t *testing.T) { + repo := newRepoForSessionTests(t) + seedForMsgTest(t, repo, "task-recovery", "session-recovery", "turn-recovery") + ctx := context.Background() + const key = "missing_pr_branch_recovery_claimed" + + if err := repo.SetSessionMetadataKey(ctx, "session-recovery", key, true); err != nil { + t.Fatalf("seed claim: %v", err) + } + removed, err := repo.RemoveSessionMetadataKeyIfState(ctx, "session-recovery", key, models.TaskSessionStateFailed) + if err != nil { + t.Fatalf("remove before failed state: %v", err) + } + if removed { + t.Fatal("remove should reject a session that is not FAILED") + } + if err := repo.UpdateTaskSessionState(ctx, "session-recovery", models.TaskSessionStateFailed, "missing branch"); err != nil { + t.Fatalf("mark failed: %v", err) + } + removed, err = repo.RemoveSessionMetadataKeyIfState(ctx, "session-recovery", key, models.TaskSessionStateFailed) + if err != nil { + t.Fatalf("remove after failed state: %v", err) + } + if !removed { + t.Fatal("remove should delete the failed-session claim") + } + session, err := repo.GetTaskSession(ctx, "session-recovery") + if err != nil { + t.Fatalf("get session after remove: %v", err) + } + if _, exists := session.Metadata[key]; exists { + t.Fatal("removed claim remains in session metadata") + } +} + // TestSetSessionMetadataKeyIfAbsentQueryUsesPostgresJSONB verifies the // Postgres dialect's write-once query uses jsonb_set/jsonb_extract_path // rather than SQLite's json_set/json_type/json functions. diff --git a/apps/web/components/session/prepare-progress.test.tsx b/apps/web/components/session/prepare-progress.test.tsx index 9e2d4da24d..460d2014c3 100644 --- a/apps/web/components/session/prepare-progress.test.tsx +++ b/apps/web/components/session/prepare-progress.test.tsx @@ -1,4 +1,4 @@ -import { cleanup, render, screen } from "@testing-library/react"; +import { cleanup, fireEvent, render, screen } from "@testing-library/react"; import { afterEach, describe, expect, it, vi } from "vitest"; import type { Message } from "@/lib/types/http"; import type { PrepareStepInfo } from "@/lib/state/slices/session-runtime/types"; @@ -7,6 +7,8 @@ let mockSteps: PrepareStepInfo[] = []; let mockPrepareStatus: "preparing" | "completed" | "failed" = "preparing"; let mockSessionState: string = "STARTING"; let mockMessages: Message[] = []; +let mockPrepareError: string | undefined; +const CREATE_WORKTREE = "Create worktree"; vi.mock("@/components/state-provider", () => ({ useAppStore: (selector: (state: Record) => unknown) => @@ -17,6 +19,7 @@ vi.mock("@/components/state-provider", () => ({ sessionId: "session-1", status: mockPrepareStatus, steps: mockSteps, + errorMessage: mockPrepareError, }, }, }, @@ -45,6 +48,7 @@ describe("PrepareProgress", () => { afterEach(() => { cleanup(); mockMessages = []; + mockPrepareError = undefined; }); it("hides skipped steps that have no useful details", () => { @@ -127,6 +131,32 @@ describe("PrepareProgress", () => { expect(screen.getByText("Environment prepared with warnings")).toBeTruthy(); expect(screen.queryByText("Environment prepared on a fresh sandbox")).toBeNull(); }); + + it("keeps failed preparation diagnostics collapsed until requested", () => { + mockPrepareStatus = "failed"; + mockSessionState = "FAILED"; + mockPrepareError = + "branch feature/very-long-name not found locally or on remote: fatal: could not resolve host"; + mockSteps = [ + { + name: CREATE_WORKTREE, + status: "failed", + error: "fatal: could not resolve host github.com", + output: "git fetch origin feature/very-long-name", + }, + ]; + + render(); + + expect(screen.getByText("Environment setup failed")).toBeTruthy(); + expect(screen.queryByText(/branch feature\/very-long-name not found/)).toBeNull(); + expect(screen.queryByText(CREATE_WORKTREE)).toBeNull(); + + fireEvent.click(screen.getByRole("button", { name: "Show preparation details" })); + + expect(screen.getByText(/branch feature\/very-long-name not found/)).toBeTruthy(); + expect(screen.getByText(CREATE_WORKTREE)).toBeTruthy(); + }); }); function makeSetupScriptMessage(overrides: Partial = {}): Message { @@ -162,7 +192,7 @@ describe("PrepareProgress per-repo setup script", () => { mockSessionState = "STARTING"; mockSteps = [ { name: "Validate repository", status: "completed" }, - { name: "Create worktree", status: "completed" }, + { name: CREATE_WORKTREE, status: "completed" }, ]; mockMessages = [makeSetupScriptMessage()]; @@ -176,11 +206,12 @@ describe("PrepareProgress per-repo setup script", () => { // `failed` keeps the panel auto-expanded. mockPrepareStatus = "failed"; mockSessionState = "FAILED"; - mockSteps = [{ name: "Create worktree", status: "completed" }]; + mockSteps = [{ name: CREATE_WORKTREE, status: "completed" }]; mockMessages = [makeSetupScriptMessage({ exit_code: 2 })]; render(); + fireEvent.click(screen.getByRole("button", { name: "Show preparation details" })); expect(screen.getByText("Script exited with code 2")).toBeTruthy(); }); }); diff --git a/apps/web/components/session/prepare-progress.tsx b/apps/web/components/session/prepare-progress.tsx index 5f847b8c97..3e81a88177 100644 --- a/apps/web/components/session/prepare-progress.tsx +++ b/apps/web/components/session/prepare-progress.tsx @@ -78,7 +78,7 @@ function StepDetails({ step, blockCommand }: { step: PrepareStepInfo; blockComma )} {step.output && ( - 
+        
           {stripAnsi(step.output)}
         

       )}
@@ -101,7 +101,7 @@ function StepWarning({ warning, warningDetail }: { warning: string; warningDetai
             Details
           
           {detailExpanded && (
-            
+            
               {warningDetail}
             

           )}
@@ -116,7 +116,7 @@ function StepMessages({ step }: { step: PrepareStepInfo }) {
     <>
       {step.warning && }
       {step.error && (
-        
+        
           {step.error}
         

       )}
@@ -140,13 +140,15 @@ function StepRow({ step }: { step: PrepareStepInfo }) {
 
   return (
     

-      

+      

         

           
         

         {step.name || "Preparing..."}
         {inlineCommand && (
-          {inlineCommand}
+          
+            {inlineCommand}
+          
         )}
         {duration && {duration}}
         {hasExpandable && (
@@ -279,11 +281,9 @@ function HeaderIcon({ status }: { status: EffectiveStatus }) {
 
 function getHeaderLabel(status: EffectiveStatus, prepareState: SessionPrepareState): string {
   if (status === "preparing") return "Preparing environment...";
-  if (status === "failed") return prepareState.errorMessage || "Environment preparation failed";
+  if (status === "failed") return "Environment setup failed";
   if (status === "completed_with_error") {
-    const failed = prepareState.steps.find((s) => s.status === "failed");
-    // `||` (not `??`) so an empty-string error also falls through to the name fallback.
-    return failed?.error || `${failed?.name ?? "Environment prepared"} — step failed`;
+    return "Environment setup finished with errors";
   }
   if (status === "completed_with_warnings") {
     const warningSteps = prepareState.steps.filter((s) => s.warning);
@@ -452,13 +452,21 @@ function logPrepareSnapshotChange(
   });
 }
 
+function hasPrepareDetails(
+  visibleStepCount: number,
+  hasSessionInfo: boolean,
+  hasFailureDetails: boolean,
+): boolean {
+  return visibleStepCount > 0 || hasSessionInfo || hasFailureDetails;
+}
+
 export function PrepareProgress({ sessionId }: PrepareProgressProps) {
   const { status, prepareState } = usePrepareStatus(sessionId);
   const session = useAppStore((state) => state.taskSessions.items[sessionId]);
   const setupScriptSteps = useSetupScriptSteps(sessionId);
-  // Only the clean-success status auto-collapses. Anything with running/error/
-  // warning context stays open by default so the user sees what's happening.
-  const autoExpand = status !== "completed";
+  // Keep active progress and warnings visible. Failure diagnostics remain
+  // secondary until the user explicitly asks to inspect them.
+  const autoExpand = status === "preparing" || status === "completed_with_warnings";
   const [expanded, setExpanded] = useExpandedFlag(sessionId, autoExpand);
 
   // Track status/expand transitions over the panel's lifetime — the
@@ -496,20 +504,55 @@ export function PrepareProgress({ sessionId }: PrepareProgressProps) {
   const headerLabel = getHeaderLabel(status, prepareState);
   const isErrorStatus = status === "failed" || status === "completed_with_error";
   const headerClass = cn("text-xs", isErrorStatus ? "text-destructive" : "text-muted-foreground");
+  const hasFailureDetails = isErrorStatus && Boolean(prepareState.errorMessage);
+  const hasExpandableContent = hasPrepareDetails(
+    visibleSteps.length,
+    hasSessionInfo,
+    hasFailureDetails,
+  );
 
   return (
-    

+    

       }
         header={
-          
-            {headerLabel}
-          
+          

+            
+              {headerLabel}
+            
+            {hasExpandableContent && (
+              
+            )}
+          

         }
-        hasExpandableContent={visibleSteps.length > 0 || hasSessionInfo}
+        hasExpandableContent={hasExpandableContent}
         isExpanded={expanded}
         onToggle={() => setExpanded(!expanded)}
       >
+        {hasFailureDetails && (
+          
+            {prepareState.errorMessage}
+          

+        )}
         

           {visibleSteps.map((step, i) => (
             
diff --git a/apps/web/components/task/chat/message-list-footer.test.tsx b/apps/web/components/task/chat/message-list-footer.test.tsx
new file mode 100644
index 0000000000..a6d89f17a8
--- /dev/null
+++ b/apps/web/components/task/chat/message-list-footer.test.tsx
@@ -0,0 +1,113 @@
+import { cleanup, render, screen } from "@testing-library/react";
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { sessionId as toSessionId, taskId as toTaskId, type Message } from "@/lib/types/http";
+
+vi.mock("@/components/task/chat/messages/agent-status", () => ({
+  AgentStatus: () => 
Agent failed
,
+}));
+
+vi.mock("@/components/task/chat/message-renderer", () => ({
+  MessageRenderer: ({ comment }: { comment: Message }) => (
+    
{comment.content}

+  ),
+}));
+
+import { MessageListFooter } from "./message-list-footer";
+
+afterEach(cleanup);
+
+const AGENT_STATUS_TEST_ID = "agent-status";
+
+const actionableFailure = {
+  id: "failure-1",
+  session_id: toSessionId("session-1"),
+  task_id: toTaskId("task-1"),
+  author_type: "agent",
+  type: "status",
+  created_at: "2026-07-22T00:00:00Z",
+  content: "Branch recovery",
+  metadata: {
+    variant: "warning",
+    failure_kind: "missing_pr_branch",
+    actions: [{ type: "archive_task", label: "Archive task" }],
+  },
+} satisfies Message;
+
+const laterFailure = {
+  ...actionableFailure,
+  id: "failure-2",
+  content: "Agent encountered an authentication error",
+  metadata: {
+    variant: "error",
+    recovery_actions: true,
+    actions: [{ type: "ws_request", label: "Resume session" }],
+  },
+} satisfies Message;
+
+describe("MessageListFooter", () => {
+  it("lets an actionable footer failure own the failure presentation", () => {
+    render(
+      ,
+    );
+
+    expect(screen.getByTestId("action-message")).toBeTruthy();
+    expect(screen.queryByTestId(AGENT_STATUS_TEST_ID)).toBeNull();
+  });
+
+  it("keeps the generic status for a failed session without an actionable footer", () => {
+    render();
+
+    expect(screen.getByTestId(AGENT_STATUS_TEST_ID)).toBeTruthy();
+  });
+
+  it("retains the running status when an action message is hidden during startup", () => {
+    render(
+      ,
+    );
+
+    expect(screen.getByTestId(AGENT_STATUS_TEST_ID)).toBeTruthy();
+  });
+
+  it("switches ownership when missing-branch recovery arrives after failure", () => {
+    const { rerender } = render(
+      ,
+    );
+    expect(screen.getByTestId(AGENT_STATUS_TEST_ID)).toBeTruthy();
+
+    rerender(
+      ,
+    );
+
+    expect(screen.queryByTestId(AGENT_STATUS_TEST_ID)).toBeNull();
+    expect(screen.getByTestId("action-message")).toBeTruthy();
+  });
+
+  it("does not restore stale missing-branch recovery after a later failure", () => {
+    render(
+      ,
+    );
+
+    expect(screen.getByTestId(AGENT_STATUS_TEST_ID)).toBeTruthy();
+    expect(screen.queryByTestId("action-message")).toBeNull();
+  });
+});
diff --git a/apps/web/components/task/chat/message-list-footer.tsx b/apps/web/components/task/chat/message-list-footer.tsx
new file mode 100644
index 0000000000..68f05df8f6
--- /dev/null
+++ b/apps/web/components/task/chat/message-list-footer.tsx
@@ -0,0 +1,60 @@
+"use client";
+
+import type { Message, TaskSessionState } from "@/lib/types/http";
+import { AgentStatus } from "@/components/task/chat/messages/agent-status";
+import { MessageRenderer } from "@/components/task/chat/message-renderer";
+
+type MessageListFooterProps = {
+  sessionState?: TaskSessionState;
+  sessionId: string | null;
+  messages: Message[];
+  footerActionMessages?: Message[];
+};
+
+function isMissingBranchFailure(message: Message): boolean {
+  const metadata = message.metadata as Record | undefined;
+  const actions = metadata?.actions;
+  if (!Array.isArray(actions) || actions.length === 0) return false;
+  return metadata?.failure_kind === "missing_pr_branch";
+}
+
+function isActionableFailure(message: Message): boolean {
+  const metadata = message.metadata as Record | undefined;
+  return Array.isArray(metadata?.actions) && metadata.actions.length > 0;
+}
+
+function findCurrentActionableFailure(
+  messages: Message[],
+  footerActionMessages: Message[],
+): Message | undefined {
+  for (let index = messages.length - 1; index >= 0; index--) {
+    if (isActionableFailure(messages[index])) return messages[index];
+  }
+  return footerActionMessages.at(-1);
+}
+
+export function MessageListFooter({
+  sessionState,
+  sessionId,
+  messages,
+  footerActionMessages = [],
+}: MessageListFooterProps) {
+  const currentActionableFailure = findCurrentActionableFailure(messages, footerActionMessages);
+  const recoveryOwnsFailure =
+    sessionState === "FAILED" &&
+    currentActionableFailure !== undefined &&
+    isMissingBranchFailure(currentActionableFailure);
+  const visibleFooterActionMessages = footerActionMessages.filter(
+    (message) => !isMissingBranchFailure(message) || message.id === currentActionableFailure?.id,
+  );
+  return (
+    <>
+      {!recoveryOwnsFailure && (
+        
+      )}
+      {visibleFooterActionMessages.map((message) => (
+        
+      ))}
+    
+  );
+}
diff --git a/apps/web/components/task/chat/message-list-native.tsx b/apps/web/components/task/chat/message-list-native.tsx
index d328c33c28..7aa82e57d0 100644
--- a/apps/web/components/task/chat/message-list-native.tsx
+++ b/apps/web/components/task/chat/message-list-native.tsx
@@ -4,9 +4,8 @@ import { useCallback, useEffect, useLayoutEffect, useMemo, useRef, memo } from "
 import { SessionPanelContent } from "@kandev/ui/pannel-session";
 import { useDockviewStore } from "@/lib/state/dockview-store";
 import type { Message } from "@/lib/types/http";
-import { AgentStatus } from "@/components/task/chat/messages/agent-status";
-import { MessageRenderer } from "@/components/task/chat/message-renderer";
 import { useLazyLoadMessages } from "@/hooks/use-lazy-load-messages";
+import { MessageListFooter } from "./message-list-footer";
 import {
   type MessageListProps,
   MessageListStatus,
@@ -256,10 +255,12 @@ export const NativeMessageList = memo(function NativeMessageList({
         );
       })}
 
-      
-      {(footerActionMessages ?? []).map((msg: Message) => (
-        
-      ))}
+      
 
       {/* Bottom anchor — browser keeps scroll pinned here when new content appends */}
       

diff --git a/apps/web/components/task/chat/message-list-virtuoso.tsx b/apps/web/components/task/chat/message-list-virtuoso.tsx
index 5e4fcecfc1..20bd710984 100644
--- a/apps/web/components/task/chat/message-list-virtuoso.tsx
+++ b/apps/web/components/task/chat/message-list-virtuoso.tsx
@@ -6,9 +6,8 @@ import { Virtuoso, type VirtuosoHandle } from "react-virtuoso";
 import { SessionPanelContent } from "@kandev/ui/pannel-session";
 import type { RenderItem } from "@/hooks/use-processed-messages";
 import type { Message, TaskSessionState } from "@/lib/types/http";
-import { AgentStatus } from "@/components/task/chat/messages/agent-status";
-import { MessageRenderer } from "@/components/task/chat/message-renderer";
 import { useLazyLoadMessages } from "@/hooks/use-lazy-load-messages";
+import { MessageListFooter } from "./message-list-footer";
 import {
   type MessageListProps,
   MessageListStatus,
@@ -431,12 +430,12 @@ function useVirtuosoHeaderFooter(args: HeaderFooterArgs) {
 
   const Footer = useCallback(
     () => (
-      <>
-        
-        {footerActions.map((msg) => (
-          
-        ))}
-      
+      
     ),
     [sessionId, sessionState, messages, footerActions],
   );
@@ -503,10 +502,12 @@ export const VirtuosoMessageList = memo(function VirtuosoMessageList(props: Mess
           messagesCount={messages.length}
           onLoadMore={loadMore}
         />
-        
-        {footerActions.map((msg) => (
-          
-        ))}
+        
       
     );
   }
diff --git a/apps/web/components/task/chat/messages/action-message.test.tsx b/apps/web/components/task/chat/messages/action-message.test.tsx
index 1dcf3c9e04..8d3ae12aca 100644
--- a/apps/web/components/task/chat/messages/action-message.test.tsx
+++ b/apps/web/components/task/chat/messages/action-message.test.tsx
@@ -23,6 +23,7 @@ afterEach(() => {
 });
 
 const CANCEL_TEST_ID = "recovery-cancel-retry-button";
+const TECHNICAL_DETAILS = "Technical details";
 
 function retryMessage(overrides: Partial = {}): Message {
   return {
@@ -60,9 +61,15 @@ function retryMessage(overrides: Partial = {}): Message {
 
 /** ActionMessage reads session state from the store (keyed by comment.session_id),
  *  so seed it via the provider instead of passing a prop. */
-function renderAction(comment: Message, sessionState?: TaskSessionState) {
+function renderAction(comment: Message, sessionState?: TaskSessionState, sessionError?: string) {
   const initialState: Partial = sessionState
-    ? { taskSessions: { items: { "sess-1": { state: sessionState } as TaskSession } } }
+    ? {
+        taskSessions: {
+          items: {
+            "sess-1": { state: sessionState, error_message: sessionError } as TaskSession,
+          },
+        },
+      }
     : {};
   return render(, {
     wrapper: ({ children }) => (
@@ -95,6 +102,11 @@ describe("ActionMessage — transient retry (warning variant)", () => {
     expect(container.firstChild).toBeNull();
   });
 
+  it("hides while the session is STARTING so the startup status remains visible", () => {
+    const { container } = renderAction(retryMessage(), "STARTING");
+    expect(container.firstChild).toBeNull();
+  });
+
   it("renders the red variant for a non-warning recovery banner", () => {
     const errorMsg = retryMessage({
       content: "Agent encountered an error",
@@ -112,3 +124,77 @@ describe("ActionMessage — transient retry (warning variant)", () => {
     expect(text.className).not.toContain("text-amber-600");
   });
 });
+
+describe("ActionMessage — missing PR branch", () => {
+  it("renders a plain-language recovery panel with collapsed technical details", () => {
+    renderAction(
+      retryMessage({
+        content:
+          'The remote PR branch "codex/enhance-prompt-result-delivery" no longer exists (likely merged and deleted).',
+        metadata: {
+          variant: "warning",
+          failure_kind: "missing_pr_branch",
+          missing_branch: "codex/enhance-prompt-result-delivery",
+          error_output: "fatal: unable to access github.com: Could not resolve host",
+          actions: [
+            {
+              type: "archive_task",
+              label: "Archive task",
+              icon: "archive",
+              test_id: "missing-branch-archive-button",
+            },
+            {
+              type: "delete_task",
+              label: "Delete task",
+              icon: "trash",
+              variant: "destructive",
+              test_id: "missing-branch-delete-button",
+            },
+          ],
+        },
+      } as Partial),
+      "FAILED",
+    );
+
+    expect(screen.getByTestId("missing-branch-recovery")).toBeTruthy();
+    expect(screen.getByText("Branch is no longer available")).toBeTruthy();
+    expect(screen.getByText("codex/enhance-prompt-result-delivery")).toBeTruthy();
+    const technicalDetails = screen.getByText(TECHNICAL_DETAILS).closest("details");
+    expect(technicalDetails?.open).toBe(false);
+    expect(screen.getByTestId("missing-branch-archive-button").className).toContain("min-h-11");
+    expect(screen.getByTestId("missing-branch-delete-button").className).toContain("min-h-11");
+
+    fireEvent.click(screen.getByText(TECHNICAL_DETAILS));
+    expect(technicalDetails?.open).toBe(true);
+    expect(screen.getByText(/Could not resolve host/)).toBeTruthy();
+  });
+
+  it("uses the current session error as collapsed technical details", () => {
+    renderAction(
+      retryMessage({
+        content: 'The remote PR branch "feature/missing" no longer exists.',
+        metadata: {
+          variant: "warning",
+          failure_kind: "missing_pr_branch",
+          missing_branch: "feature/missing",
+          actions: [
+            {
+              type: "archive_task",
+              label: "Archive task",
+              test_id: "missing-branch-archive-button",
+            },
+          ],
+        },
+      } as Partial),
+      "FAILED",
+      "environment preparation failed: fatal: could not resolve host github.com",
+    );
+
+    const details = screen.getByText(TECHNICAL_DETAILS).closest("details");
+    expect(details?.open).toBe(false);
+    expect(screen.getByText(/could not resolve host github.com/)).toBeTruthy();
+
+    fireEvent.click(screen.getByText(TECHNICAL_DETAILS));
+    expect(details?.open).toBe(true);
+  });
+});
diff --git a/apps/web/components/task/chat/messages/action-message.tsx b/apps/web/components/task/chat/messages/action-message.tsx
index b0fe5fff34..05910257d4 100644
--- a/apps/web/components/task/chat/messages/action-message.tsx
+++ b/apps/web/components/task/chat/messages/action-message.tsx
@@ -10,6 +10,7 @@ import {
   IconSparkles,
   IconGitCommit,
   IconX,
+  IconChevronDown,
 } from "@tabler/icons-react";
 import { cn } from "@/lib/utils";
 import { Button } from "@kandev/ui/button";
@@ -41,6 +42,8 @@ type ActionMeta = {
   is_auth_error?: boolean;
   auth_methods?: RecoveryAuthMethod[];
   error_output?: string;
+  failure_kind?: string;
+  missing_branch?: string;
 };
 
 function isSessionActive(state?: TaskSessionState) {
@@ -56,6 +59,11 @@ export const ActionMessage = memo(function ActionMessage({ comment }: { comment:
       ? (state.taskSessions.items[comment.session_id]?.state ?? undefined)
       : undefined,
   );
+  const sessionError = useAppStore((state) =>
+    comment.session_id
+      ? (state.taskSessions.items[comment.session_id]?.error_message as string | undefined)
+      : undefined,
+  );
   const metadata = comment.metadata as ActionMeta | undefined;
   const isWarning = metadata?.variant === "warning";
   const message = comment.content || "An error occurred";
@@ -63,6 +71,17 @@ export const ActionMessage = memo(function ActionMessage({ comment }: { comment:
   // Hide once session is active again (recovery succeeded)
   if (isSessionActive(sessionState)) return null;
 
+  if (metadata?.failure_kind === "missing_pr_branch") {
+    return (
+      
+    );
+  }
+
   const iconClass = isWarning ? "text-amber-500" : "text-red-500";
   const textClass = isWarning
     ? "text-amber-600 dark:text-amber-400"
@@ -75,7 +94,7 @@ export const ActionMessage = memo(function ActionMessage({ comment }: { comment:
           
         

         

-          
{message}

+          
{message}

           
           {metadata?.actions && metadata.actions.length > 0 && (
             
@@ -86,7 +105,73 @@ export const ActionMessage = memo(function ActionMessage({ comment }: { comment:
   );
 });
 
-function ActionMessageDetails({ metadata }: { metadata: ActionMeta | undefined }) {
+function MissingBranchRecovery({
+  metadata,
+  taskId,
+  fallbackMessage,
+  technicalDetails,
+}: {
+  metadata: ActionMeta;
+  taskId?: string;
+  fallbackMessage: string;
+  technicalDetails?: string;
+}) {
+  const branch = metadata.missing_branch?.trim();
+  return (
+    

+      

+        

+    

+  );
+}
+
+function TechnicalDetails({ children }: { children: string }) {
+  return (
+    

+      
+        
+        Technical details
+      
+      
+        {children}
+      

+    

+  );
+}
+
+function ActionMessageDetails({
+  metadata,
+  technicalDetails,
+}: {
+  metadata: ActionMeta | undefined;
+  technicalDetails?: string;
+}) {
   const [hostShellOpen, setHostShellOpen] = useState(false);
   const [hostShellCommand, setHostShellCommand] = useState(undefined);
 
@@ -106,13 +191,10 @@ function ActionMessageDetails({ metadata }: { metadata: ActionMeta | undefined }
   }, []);
 
   if (!metadata) return null;
+  const errorOutput = metadata.error_output || technicalDetails;
   return (
     <>
-      {metadata.error_output && (
-        
-          {metadata.error_output}
-        

-      )}
+      {errorOutput && {errorOutput}}
       {metadata.is_auth_error && metadata.auth_methods && metadata.auth_methods.length > 0 && (
         
+    

       {actions.map((action, i) => (
         
       ))}
@@ -206,7 +288,7 @@ function ActionButton({
       variant="outline"
       size="sm"
       className={cn(
-        "h-7 text-xs cursor-pointer gap-1.5",
+        "h-auto min-h-11 w-full gap-1.5 text-xs cursor-pointer sm:min-h-8 sm:w-auto",
         isDestructive && "text-destructive hover:text-destructive",
       )}
       disabled={disabled}
diff --git a/apps/web/components/task/chat/messages/agent-status.tsx b/apps/web/components/task/chat/messages/agent-status.tsx
index 316fbe4058..b09423b2db 100644
--- a/apps/web/components/task/chat/messages/agent-status.tsx
+++ b/apps/web/components/task/chat/messages/agent-status.tsx
@@ -140,9 +140,8 @@ function AgentErrorStatus({
       ? (state.taskSessions.items[sessionId]?.error_message as string | undefined)
       : undefined,
   );
-  const [userCollapsed, setUserCollapsed] = useState(false);
-  const expanded = !!errorMessage && !userCollapsed;
-  const toggle = useCallback(() => setUserCollapsed((v) => !v), []);
+  const [expanded, setExpanded] = useState(false);
+  const toggle = useCallback(() => setExpanded((value) => !value), []);
 
   const displayLabel = resolveAgentErrorLabel(errorMessage, config.label);
   const hasDetails = !!errorMessage;
@@ -155,12 +154,16 @@ function AgentErrorStatus({
     >