From edee5a03cfb00bc63823f0481a7664f503183cb6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20Flor=C3=AAncio?= Date: Thu, 23 Jul 2026 14:45:01 +0100 Subject: [PATCH] fix: scope PR branch failure guidance (#1873) --- .../adapter/transport/acp/load_burst_test.go | 250 +++++++++-- .../internal/launcher/supervisor_test.go | 18 +- .../event_handlers_github_test.go | 6 +- .../orchestrator/executor/executor.go | 5 +- .../orchestrator/executor/executor_execute.go | 60 ++- .../executor/executor_multi_repo_test.go | 87 ++++ apps/backend/internal/orchestrator/service.go | 95 +++- .../orchestrator/task_launch_failure_test.go | 406 +++++++++++++++++- 8 files changed, 867 insertions(+), 60 deletions(-) diff --git a/apps/backend/internal/agentctl/server/adapter/transport/acp/load_burst_test.go b/apps/backend/internal/agentctl/server/adapter/transport/acp/load_burst_test.go index 438ce8d15a..cad7e778d5 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/acp/load_burst_test.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/acp/load_burst_test.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "io" + "sync" "testing" "time" @@ -12,6 +13,16 @@ import ( "github.com/kandev/kandev/internal/agentctl/types/streams" ) +const ( + replayBurstProducerWatchdogTimeout = 20 * time.Second + burstProducerCleanupTimeout = time.Second +) + +var ( + errReplayBurstProducerStalled = errors.New("replay burst producer stalled") + errReplayBurstProducerBeforeWriteStalled = errors.New("replay burst producer stalled before writer entry") +) + // burstAgent is a minimal acp.Agent stub used by the load-replay regression // test. None of these methods are exercised — we only need a peer that holds // the connection open while we push notifications agent → client. @@ -99,26 +110,202 @@ func (*burstClient) WaitForTerminalExit(context.Context, acp.WaitForTerminalExit return acp.WaitForTerminalExitResponse{}, errors.New("not implemented") } -// newBurstPair wires a real acp.ClientSideConnection / AgentSideConnection -// pair over in-memory pipes. The client's SessionUpdate forwards to the -// supplied handler. Cleanup is registered on the test. -func newBurstPair(t *testing.T, onUpdate func(acp.SessionNotification)) (*acp.ClientSideConnection, *acp.AgentSideConnection) { +type burstPair struct { + clientConn *acp.ClientSideConnection + agentConn *acp.AgentSideConnection + + c2aR *io.PipeReader + c2aW *io.PipeWriter + a2cR *io.PipeReader + a2cW *io.PipeWriter + + agentToClientWriteEntered <-chan struct{} + closed chan struct{} + closeOnce sync.Once +} + +func (p *burstPair) Close() { + p.closeOnce.Do(func() { + _ = p.c2aR.Close() + _ = p.c2aW.Close() + _ = p.a2cR.Close() + _ = p.a2cW.Close() + close(p.closed) + }) +} + +// writeEntrySignalWriter reports immediately before delegating to its writer. +// When it wraps an unread io.PipeWriter, receiving the signal proves the +// producer has entered the write that will block until the pipe is closed. +type writeEntrySignalWriter struct { + writer io.Writer + entered chan struct{} + once sync.Once +} + +func (w *writeEntrySignalWriter) Write(p []byte) (int, error) { + w.once.Do(func() { close(w.entered) }) + return w.writer.Write(p) +} + +// newBurstPair wires an AgentSideConnection over in-memory pipes. When +// onUpdate is supplied, it also creates a ClientSideConnection whose +// SessionUpdate forwards to that handler. Cleanup is registered on the test. +func newBurstPair(t *testing.T, onUpdate func(acp.SessionNotification)) *burstPair { t.Helper() c2aR, c2aW := io.Pipe() a2cR, a2cW := io.Pipe() + a2cWriteEntered := make(chan struct{}) + a2cWriteObserver := &writeEntrySignalWriter{writer: a2cW, entered: a2cWriteEntered} - clientConn := acp.NewClientSideConnection(&burstClient{onUpdate: onUpdate}, c2aW, a2cR) - agentConn := acp.NewAgentSideConnection(burstAgent{}, a2cW, c2aR) + pair := &burstPair{ + agentConn: acp.NewAgentSideConnection(burstAgent{}, a2cWriteObserver, c2aR), + c2aR: c2aR, + c2aW: c2aW, + a2cR: a2cR, + a2cW: a2cW, + agentToClientWriteEntered: a2cWriteEntered, + closed: make(chan struct{}), + } + if onUpdate != nil { + pair.clientConn = acp.NewClientSideConnection( + &burstClient{onUpdate: onUpdate}, c2aW, a2cR, + acp.WithMaxQueuedNotifications(acpNotifQueueCapacity()), + ) + } + t.Cleanup(pair.Close) - t.Cleanup(func() { - _ = c2aR.Close() - _ = c2aW.Close() - _ = a2cR.Close() - _ = a2cW.Close() + return pair +} + +// newStalledBurstPair leaves the agent-to-client pipe unread so writes from +// AgentSideConnection block until the pair is closed. +func newStalledBurstPair(t *testing.T) *burstPair { + return newBurstPair(t, nil) +} + +// runBurstProducer ensures a stalled synchronous io.Pipe write fails the +// test promptly. acp-go-sdk's write path cannot observe context cancellation +// after entering io.Pipe.Write, so the watchdog must close this pair's pipes. +func runBurstProducer(pair *burstPair, timeout time.Duration, produce func() error) error { + done := make(chan error, 1) + go func() { + done <- produce() + }() + + entryTimer := time.NewTimer(timeout) + defer entryTimer.Stop() + + // JSON encoding and SDK setup happen before the synchronous io.Pipe write. + // Bound that phase separately: it must not prevent the test from cleaning + // up a producer which never reaches the instrumented writer. + select { + case <-pair.agentToClientWriteEntered: + case err := <-done: + return err + case <-entryTimer.C: + return stalledBurstProducerError(pair, timeout, done, errReplayBurstProducerBeforeWriteStalled) + } + + timer := time.NewTimer(timeout) + defer timer.Stop() + + select { + case err := <-done: + return err + case <-timer.C: + return stalledBurstProducerError(pair, timeout, done, errReplayBurstProducerStalled) + } +} + +// stalledBurstProducerError owns the failure path for both producer phases. +// It always reports the timeout that triggered cleanup, even if closing the +// pipes lets the producer return nil. +func stalledBurstProducerError(pair *burstPair, timeout time.Duration, done <-chan error, stallErr error) error { + pair.Close() + + cleanupTimer := time.NewTimer(burstProducerCleanupTimeout) + defer cleanupTimer.Stop() + select { + case err := <-done: + return fmt.Errorf("%w after %s: %v", stallErr, timeout, err) + case <-cleanupTimer.C: + return fmt.Errorf("%w after %s: pipe cleanup did not release producer", stallErr, timeout) + } +} + +func TestBurstProducerWatchdog_ClosesStalledPipes(t *testing.T) { + const watchdogTimeout = 100 * time.Millisecond + + pair := newStalledBurstPair(t) + producerExited := make(chan struct{}) + err := runBurstProducer(pair, watchdogTimeout, func() error { + defer close(producerExited) + return pair.agentConn.SessionUpdate(context.Background(), makeReplayNotification("stalled-session", 1)) }) - return clientConn, agentConn + select { + case <-pair.agentToClientWriteEntered: + default: + t.Fatal("producer did not enter the agent-to-client pipe write") + } + select { + case <-producerExited: + default: + t.Fatal("watchdog returned before the stalled producer exited") + } + if !errors.Is(err, errReplayBurstProducerStalled) { + t.Fatalf("watchdog error = %v, want stalled producer error", err) + } +} + +func TestBurstProducerWatchdog_ClosesPipesBeforeAgentToClientWrite(t *testing.T) { + const watchdogTimeout = 100 * time.Millisecond + + pair := newStalledBurstPair(t) + producerBlocked := make(chan struct{}) + producerExited := make(chan struct{}) + runDone := make(chan error, 1) + go func() { + runDone <- runBurstProducer(pair, watchdogTimeout, func() error { + close(producerBlocked) + <-pair.closed + close(producerExited) + return nil + }) + }() + + select { + case <-producerBlocked: + case <-time.After(watchdogTimeout): + t.Fatal("producer did not block before SessionUpdate") + } + + select { + case err := <-runDone: + if !errors.Is(err, errReplayBurstProducerBeforeWriteStalled) { + t.Fatalf("watchdog error = %v, want pre-write stalled producer error", err) + } + case <-time.After(watchdogTimeout + burstProducerCleanupTimeout + time.Second): + t.Fatal("writer-entry watchdog did not close pipes and return") + } + + select { + case <-producerExited: + default: + t.Fatal("writer-entry watchdog returned before the pre-write producer exited") + } + select { + case <-pair.closed: + default: + t.Fatal("writer-entry watchdog did not close the owned pipes") + } + select { + case <-pair.agentToClientWriteEntered: + t.Fatal("producer entered agent-to-client write before watchdog cleanup") + default: + } } // TestLoadReplayBurst_HandlesLargeReplay is a regression test for the @@ -157,7 +344,7 @@ func TestLoadReplayBurst_HandlesLargeReplay(t *testing.T) { // just rely on Close to drain it. t.Cleanup(func() { _ = a.Close() }) - clientConn, agentConn := newBurstPair(t, a.enqueueACPUpdate) + pair := newBurstPair(t, a.enqueueACPUpdate) const sessionID = "burst-session" @@ -167,25 +354,32 @@ func TestLoadReplayBurst_HandlesLargeReplay(t *testing.T) { // Push the burst of replay notifications. These would all be suppressed // by the adapter (AgentMessageChunk + ToolCall + Plan are on the suppress // list) — the point is whether the SDK's 1024-deep queue stays drained. - for i := 0; i < notificationBurstSize; i++ { - if err := agentConn.SessionUpdate(ctx, makeReplayNotification(sessionID, i)); err != nil { - t.Fatalf("SessionUpdate at i=%d failed: %v", i, err) + // A stalled SDK pipe write does not honor ctx; bound the producer phase and + // close only this pair's pipes if that happens. + if err := runBurstProducer(pair, replayBurstProducerWatchdogTimeout, func() error { + for i := 0; i < notificationBurstSize; i++ { + if err := pair.agentConn.SessionUpdate(ctx, makeReplayNotification(sessionID, i)); err != nil { + return fmt.Errorf("SessionUpdate at i=%d: %w", i, err) + } } - } - // Sentinel: AvailableCommandsUpdate passes through during load, so it - // will land on updatesCh once the consumer drains the queue. - if err := agentConn.SessionUpdate(ctx, acp.SessionNotification{ - SessionId: acp.SessionId(sessionID), - Update: acp.SessionUpdate{ - AvailableCommandsUpdate: &acp.SessionAvailableCommandsUpdate{ - AvailableCommands: []acp.AvailableCommand{ - {Name: sentinelCmd, Description: "burst sentinel"}, + // Sentinel: AvailableCommandsUpdate passes through during load, so it + // will land on updatesCh once the consumer drains the queue. + if err := pair.agentConn.SessionUpdate(ctx, acp.SessionNotification{ + SessionId: acp.SessionId(sessionID), + Update: acp.SessionUpdate{ + AvailableCommandsUpdate: &acp.SessionAvailableCommandsUpdate{ + AvailableCommands: []acp.AvailableCommand{ + {Name: sentinelCmd, Description: "burst sentinel"}, + }, }, }, - }, + }); err != nil { + return fmt.Errorf("sentinel SessionUpdate: %w", err) + } + return nil }); err != nil { - t.Fatalf("sentinel SessionUpdate failed: %v", err) + t.Fatal(err) } // The sentinel-wait loop below is the authoritative overflow check: the @@ -202,7 +396,7 @@ func TestLoadReplayBurst_HandlesLargeReplay(t *testing.T) { if ev.Type == streams.EventTypeAvailableCommands { got = &ev } - case <-clientConn.Done(): + case <-pair.clientConn.Done(): t.Fatal("client connection closed while waiting for sentinel — overflow") case <-deadline: t.Fatal("timeout waiting for sentinel AvailableCommands event after replay burst") diff --git a/apps/backend/internal/launcher/supervisor_test.go b/apps/backend/internal/launcher/supervisor_test.go index 8777866b33..e849c53940 100644 --- a/apps/backend/internal/launcher/supervisor_test.go +++ b/apps/backend/internal/launcher/supervisor_test.go @@ -48,14 +48,30 @@ func TestPrepareSupervisorEnvWritesExpectedPaths(t *testing.T) { } func TestListenControlSocketUsesOwnerOnlyMode(t *testing.T) { + const unixSocketTempRoot = "/tmp" + if runtime.GOOS == "windows" { t.Skip("unix sockets are not supported on windows") } - dir, err := os.MkdirTemp("", "kandev-sock-*") + shortRoot, err := os.MkdirTemp(unixSocketTempRoot, "kandev-inherited-tmp-*") if err != nil { + t.Fatalf("create temporary directory under /tmp: %v", err) + } + t.Cleanup(func() { _ = os.RemoveAll(shortRoot) }) + longTMPDIR := filepath.Join(shortRoot, "this-inherited-tmpdir-is-deliberately-long-to-exceed-the-unix-socket-path-limit") + if err := os.Mkdir(longTMPDIR, 0o700); err != nil { t.Fatal(err) } + t.Setenv("TMPDIR", longTMPDIR) + + dir, err := os.MkdirTemp(unixSocketTempRoot, "kandev-sock-*") + if err != nil { + t.Fatalf("create socket directory under /tmp: %v", err) + } t.Cleanup(func() { _ = os.RemoveAll(dir) }) + if filepath.Dir(dir) != unixSocketTempRoot { + t.Fatalf("socket directory = %q, want it directly under %s", dir, unixSocketTempRoot) + } socket := filepath.Join(dir, "control.sock") ln, err := listenControlSocket(socket) if err != nil { diff --git a/apps/backend/internal/orchestrator/event_handlers_github_test.go b/apps/backend/internal/orchestrator/event_handlers_github_test.go index 6c0c24fd5b..8842571e94 100644 --- a/apps/backend/internal/orchestrator/event_handlers_github_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_github_test.go @@ -20,6 +20,7 @@ type mockGitHubService struct { taskPR *github.TaskPR taskPRs []*github.TaskPR taskPRErr error + taskPRListFunc func(context.Context, []string) (map[string][]*github.TaskPR, error) triggerPRSyncAllPRs []*github.TaskPR triggerPRSyncAllErr error triggerPRSyncAllCalls int @@ -91,7 +92,10 @@ func (m *mockGitHubService) GetTaskPR(_ context.Context, _ string) (*github.Task m.getTaskPRCalls++ return m.taskPR, m.taskPRErr } -func (m *mockGitHubService) ListTaskPRs(_ context.Context, taskIDs []string) (map[string][]*github.TaskPR, error) { +func (m *mockGitHubService) ListTaskPRs(ctx context.Context, taskIDs []string) (map[string][]*github.TaskPR, error) { + if m.taskPRListFunc != nil { + return m.taskPRListFunc(ctx, taskIDs) + } if m.taskPRErr != nil { return nil, m.taskPRErr } diff --git a/apps/backend/internal/orchestrator/executor/executor.go b/apps/backend/internal/orchestrator/executor/executor.go index 618d472d8e..73523cb956 100644 --- a/apps/backend/internal/orchestrator/executor/executor.go +++ b/apps/backend/internal/orchestrator/executor/executor.go @@ -587,8 +587,9 @@ type TaskReviewStateReconcileFunc func(ctx context.Context, taskID, completedSes type AgentStartFailedFunc func(ctx context.Context, taskID, sessionID, agentExecutionID string, err error, fromResume bool) (handled bool) // LaunchFailedFunc is called when session launch fails before the agent starts. -// Useful for creating user-facing status messages tied to launch errors. -type LaunchFailedFunc func(ctx context.Context, taskID, sessionID string, err error) +// repositoryID identifies the repository whose launch failed. Useful for +// creating repository-scoped user-facing status messages tied to launch errors. +type LaunchFailedFunc func(ctx context.Context, taskID, sessionID, repositoryID string, err error) // PrimarySessionSetFunc is called when the first session for a task is marked // primary. This lets the orchestrator publish a task.updated event so the diff --git a/apps/backend/internal/orchestrator/executor/executor_execute.go b/apps/backend/internal/orchestrator/executor/executor_execute.go index 81885bbb64..9099fcb340 100644 --- a/apps/backend/internal/orchestrator/executor/executor_execute.go +++ b/apps/backend/internal/orchestrator/executor/executor_execute.go @@ -7,6 +7,7 @@ import ( "hash/fnv" "maps" "os/exec" + "regexp" "sort" "strings" "sync" @@ -887,7 +888,7 @@ func (e *Executor) LaunchPreparedSession(ctx context.Context, task *v1.Task, ses // Call the AgentManager to launch the container resp, err := e.agentManager.LaunchAgent(ctx, req) if err != nil { - return nil, e.handleLaunchFailure(ctx, task.ID, sessionID, err) + return nil, e.handleLaunchFailure(ctx, task.ID, sessionID, failingLaunchRepositoryID(req, err), err) } // Create or update the task environment with launch results @@ -906,6 +907,59 @@ func (e *Executor) LaunchPreparedSession(ctx context.Context, task *v1.Task, ses return e.finalizeLaunch(ctx, task, session, agentProfileID, sessionID, primaryRepo, resp, startAgent, execCfg) } +// failingLaunchRepositoryID identifies the repository that caused a +// multi-repository branch-fetch failure. A lifecycle launch error only carries +// the failed branch name, so correlate it with the unique per-repository +// checkout branch in the request. Ambiguous or unrecognized branches fail +// closed: no repository-scoped destructive guidance can be offered. +func failingLaunchRepositoryID(req *LaunchAgentRequest, launchErr error) string { + if req == nil { + return "" + } + if len(req.Repositories) == 0 { + return req.RepositoryID + } + + branch := extractLaunchFailureBranch(launchErr) + if branch == "" { + return "" + } + var repositoryID string + for _, spec := range req.Repositories { + if strings.TrimSpace(spec.CheckoutBranch) != branch { + continue + } + if repositoryID != "" { + return "" + } + repositoryID = spec.RepositoryID + } + return repositoryID +} + +var ( + launchQuotedBranchPattern = regexp.MustCompile(`branch "([^"]+)"`) + launchRemoteRefPattern = regexp.MustCompile(`remote ref ([^\s]+)`) + launchPathspecBranchPattern = regexp.MustCompile(`pathspec '([^']+)'`) +) + +func extractLaunchFailureBranch(err error) string { + if err == nil { + return "" + } + message := err.Error() + for _, pattern := range []*regexp.Regexp{ + launchQuotedBranchPattern, + launchRemoteRefPattern, + launchPathspecBranchPattern, + } { + if match := pattern.FindStringSubmatch(message); len(match) == 2 { + return strings.TrimSpace(match[1]) + } + } + return "" +} + func resumeTokenForExecutionProfile(running *models.ExecutorRunning, profileID string) string { if running == nil || profileID == "" || (running.ExecutionProfileID != "" && running.ExecutionProfileID != profileID) { @@ -915,7 +969,7 @@ func resumeTokenForExecutionProfile(running *models.ExecutorRunning, profileID s } // handleLaunchFailure marks the session and task as FAILED and returns the original error. -func (e *Executor) handleLaunchFailure(ctx context.Context, taskID, sessionID string, launchErr error) error { +func (e *Executor) handleLaunchFailure(ctx context.Context, taskID, sessionID, repositoryID string, launchErr error) error { // Detach from caller context so failure bookkeeping completes even if the // original request context was cancelled. failCtx := context.WithoutCancel(ctx) @@ -925,7 +979,7 @@ func (e *Executor) handleLaunchFailure(ctx context.Context, taskID, sessionID st // Call onLaunchFailed before state updates so it can set the suppressToast // flag that updateSessionState will propagate to the frontend. if e.onLaunchFailed != nil { - e.onLaunchFailed(failCtx, taskID, sessionID, launchErr) + e.onLaunchFailed(failCtx, taskID, sessionID, repositoryID, launchErr) } if updateErr := e.updateSessionState(failCtx, taskID, sessionID, models.TaskSessionStateFailed, launchErr.Error()); updateErr != nil { e.logger.Warn("failed to mark session as failed after launch error", diff --git a/apps/backend/internal/orchestrator/executor/executor_multi_repo_test.go b/apps/backend/internal/orchestrator/executor/executor_multi_repo_test.go index b1aeb5d85f..d2ab062d81 100644 --- a/apps/backend/internal/orchestrator/executor/executor_multi_repo_test.go +++ b/apps/backend/internal/orchestrator/executor/executor_multi_repo_test.go @@ -2,6 +2,7 @@ package executor import ( "context" + "errors" "testing" "time" @@ -100,6 +101,92 @@ func TestLaunchPreparedSession_MultiRepo_PopulatesRequestRepositories(t *testing } } +func TestLaunchPreparedSession_MultiRepo_LaunchFailureReportsFailingSecondaryRepositoryID(t *testing.T) { + repo := newMockRepository() + const taskID = "task-multi-launch-failure" + const sessionID = "session-multi-launch-failure" + seedMultiRepoTask(t, repo, taskID) + repo.taskRepositories["tr-2"].CheckoutBranch = "feature/foo" + repo.sessions[sessionID] = &models.TaskSession{ + ID: sessionID, + TaskID: taskID, + AgentProfileID: "profile-123", + State: models.TaskSessionStateCreated, + StartedAt: time.Now(), + UpdatedAt: time.Now(), + } + + launchErr := errors.New("fatal: couldn't find remote ref feature/foo") + agentManager := &mockAgentManager{ + launchAgentFunc: func(_ context.Context, _ *LaunchAgentRequest) (*LaunchAgentResponse, error) { + return nil, launchErr + }, + } + exec := newTestExecutor(t, agentManager, repo) + var callbackRepositoryID string + exec.SetOnLaunchFailed(func(_ context.Context, callbackTaskID, callbackSessionID, repositoryID string, callbackErr error) { + if callbackTaskID != taskID || callbackSessionID != sessionID { + t.Errorf("unexpected callback target: task=%q session=%q", callbackTaskID, callbackSessionID) + } + if !errors.Is(callbackErr, launchErr) { + t.Errorf("unexpected callback error: %v", callbackErr) + } + callbackRepositoryID = repositoryID + }) + + _, err := exec.LaunchPreparedSession(context.Background(), &v1.Task{ + ID: taskID, WorkspaceID: "ws-1", Title: "Multi", + }, sessionID, LaunchOptions{AgentProfileID: "profile-123", StartAgent: false}) + if !errors.Is(err, launchErr) { + t.Fatalf("LaunchPreparedSession error = %v, want %v", err, launchErr) + } + if callbackRepositoryID != "repo-back" { + t.Fatalf("launch failure repository ID = %q, want failing secondary repository %q", callbackRepositoryID, "repo-back") + } +} + +func TestFailingLaunchRepositoryID_UsesExactBranchToken(t *testing.T) { + req := &LaunchAgentRequest{Repositories: []RepoSpec{{ + RepositoryID: "repo-secondary", + CheckoutBranch: "feature/foo", + }}} + + tests := []struct { + name string + err error + want string + }{ + { + name: "remote ref exact match", + err: errors.New("fatal: couldn't find remote ref feature/foo"), + want: "repo-secondary", + }, + { + name: "quoted branch exact match", + err: errors.New("branch \"feature/foo\" not found locally or on remote"), + want: "repo-secondary", + }, + { + name: "pathspec exact match", + err: errors.New("error: pathspec 'feature/foo' did not match any file(s) known to git"), + want: "repo-secondary", + }, + { + name: "remote ref prefix collision", + err: errors.New("fatal: couldn't find remote ref feature/foo-deleted"), + want: "", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := failingLaunchRepositoryID(req, tt.err); got != tt.want { + t.Fatalf("failingLaunchRepositoryID() = %q, want %q", got, tt.want) + } + }) + } +} + func TestLaunchPreparedSession_MultiRepo_PersistsPerRepoEnvironmentAndWorktreeRows(t *testing.T) { repo := newMockRepository() taskID := "task-multi-2" diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index a4efc5d05d..7e1c951a03 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -24,6 +24,7 @@ import ( "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/events" "github.com/kandev/kandev/internal/events/bus" + "github.com/kandev/kandev/internal/github" "github.com/kandev/kandev/internal/orchestrator/executor" "github.com/kandev/kandev/internal/orchestrator/messagequeue" "github.com/kandev/kandev/internal/orchestrator/queue" @@ -1533,7 +1534,7 @@ func canResumeRunning(running *models.ExecutorRunning) bool { return models.IsAlwaysResumableRuntime(running.Runtime) } -func isMissingMergedPRBranchError(err error) bool { +func isMissingBranchError(err error) bool { if err == nil { return false } @@ -1549,6 +1550,8 @@ var ( pathspecBranchPattern = regexp.MustCompile(`pathspec '([^']+)'`) ) +const launchFailurePRLookupTimeout = time.Second + func extractMissingBranchName(err error) string { if err == nil { return "" @@ -1566,21 +1569,97 @@ func extractMissingBranchName(err error) string { return "" } -func (s *Service) handleSessionLaunchFailed(ctx context.Context, taskID, sessionID string, launchErr error) { - if s.messageCreator == nil || !isMissingMergedPRBranchError(launchErr) { +func (s *Service) matchingTaskPRState(ctx context.Context, taskID, repositoryID, branch string) string { + repositoryID = strings.TrimSpace(repositoryID) + if s.githubService == nil || repositoryID == "" || branch == "" { + return "" + } + lookupCtx, cancel := context.WithTimeout(ctx, launchFailurePRLookupTimeout) + defer cancel() + prsByTask, err := s.githubService.ListTaskPRs(lookupCtx, []string{taskID}) + if err != nil { + s.logger.Debug("failed to load task PR state for missing branch guidance", + zap.String("task_id", taskID), + zap.Error(err)) + return "" + } + prs := prsByTask[taskID] + if state, found := taskPRState(prs, repositoryID, branch); found { + return state + } + state, found := taskPRState(prs, "", branch) + if !found || !s.hasOnlyTaskRepository(lookupCtx, taskID, repositoryID) { + return "" + } + return state +} + +func taskPRState(prs []*github.TaskPR, repositoryID, branch string) (string, bool) { + matched := false + state := "" + for _, pr := range prs { + if pr == nil || strings.TrimSpace(pr.RepositoryID) != repositoryID || strings.TrimSpace(pr.HeadBranch) != branch { + continue + } + if matched { + return "", true + } + matched = true + state = strings.ToLower(strings.TrimSpace(pr.State)) + if state != "open" && state != "closed" && state != "merged" { + return "", true + } + } + return state, matched +} + +func (s *Service) hasOnlyTaskRepository(ctx context.Context, taskID, repositoryID string) bool { + store, ok := s.repo.(interface { + ListTaskRepositories(context.Context, string) ([]*models.TaskRepository, error) + }) + if !ok { + return false + } + repositories, err := store.ListTaskRepositories(ctx, taskID) + if err != nil { + s.logger.Debug("failed to load task repositories for legacy PR guidance", + zap.String("task_id", taskID), + zap.Error(err)) + return false + } + return len(repositories) == 1 && repositories[0] != nil && + strings.TrimSpace(repositories[0].RepositoryID) == repositoryID +} + +func (s *Service) handleSessionLaunchFailed(ctx context.Context, taskID, sessionID, repositoryID string, launchErr error) { + if s.messageCreator == nil || !isMissingBranchError(launchErr) { return } branch := extractMissingBranchName(launchErr) - content := "This task references a PR branch that no longer exists on remote (likely merged and deleted)." + prState := s.matchingTaskPRState(ctx, taskID, repositoryID, branch) + if prState == "open" { + return + } + 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 = "The remote PR branch \"" + branch + "\" no longer exists (likely merged and deleted)." + content = "Kandev couldn't fetch branch \"" + branch + "\" from the configured repository. Verify the task's repository branch or PR link, then retry." + } + authoritativeMissingBranch := prState == "closed" || prState == "merged" + if authoritativeMissingBranch { + content = "This task references a PR branch that no longer exists on remote." + if branch != "" { + content = "The remote PR branch \"" + branch + "\" no longer exists." + } } metadata := map[string]interface{}{ "variant": "warning", - "failure_kind": "missing_pr_branch", + "failure_kind": "branch_fetch_failed", "missing_branch": branch, - "actions": []map[string]interface{}{ + } + if authoritativeMissingBranch { + metadata["failure_kind"] = "missing_pr_branch" + metadata["actions"] = []map[string]interface{}{ { actionMetaKeyType: "archive_task", actionMetaKeyLabel: "Archive task", @@ -1596,7 +1675,7 @@ func (s *Service) handleSessionLaunchFailed(ctx context.Context, taskID, session actionMetaKeyIcon: "trash", actionMetaKeyTestID: "missing-branch-delete-button", }, - }, + } } s.suppressToast.Store(sessionID, true) msgCtx, cancel := context.WithTimeout(ctx, 5*time.Second) diff --git a/apps/backend/internal/orchestrator/task_launch_failure_test.go b/apps/backend/internal/orchestrator/task_launch_failure_test.go index ae75adb2e8..9197bcefd2 100644 --- a/apps/backend/internal/orchestrator/task_launch_failure_test.go +++ b/apps/backend/internal/orchestrator/task_launch_failure_test.go @@ -5,12 +5,14 @@ import ( "errors" "strings" "testing" + "time" + "github.com/kandev/kandev/internal/github" "github.com/kandev/kandev/internal/task/models" v1 "github.com/kandev/kandev/pkg/api/v1" ) -func TestIsMissingMergedPRBranchError(t *testing.T) { +func TestIsMissingBranchError(t *testing.T) { tests := []struct { name string err error @@ -44,8 +46,8 @@ func TestIsMissingMergedPRBranchError(t *testing.T) { } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - if got := isMissingMergedPRBranchError(tt.err); got != tt.want { - t.Fatalf("isMissingMergedPRBranchError()=%v, want %v", got, tt.want) + if got := isMissingBranchError(tt.err); got != tt.want { + t.Fatalf("isMissingBranchError()=%v, want %v", got, tt.want) } }) } @@ -88,7 +90,7 @@ func TestExtractMissingBranchName(t *testing.T) { } } -func TestHandleSessionLaunchFailed_CreatesGuidanceMessageForMissingPRBranch(t *testing.T) { +func TestHandleSessionLaunchFailed_UnavailablePRStateCreatesNeutralFetchGuidance(t *testing.T) { ctx := context.Background() repo := setupTestRepo(t) seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateStarting) @@ -96,9 +98,10 @@ func TestHandleSessionLaunchFailed_CreatesGuidanceMessageForMissingPRBranch(t *t mc := &mockMessageCreator{} svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) svc.messageCreator = mc + svc.SetGitHubService(&mockGitHubService{taskPRErr: errors.New("persisted PR state unavailable")}) 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", 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)) @@ -113,25 +116,394 @@ func TestHandleSessionLaunchFailed_CreatesGuidanceMessageForMissingPRBranch(t *t if msg.turnID != "" { t.Fatalf("expected empty turn ID for pre-start failure, got %q", msg.turnID) } - if !strings.Contains(msg.content, "feature/foo") || !strings.Contains(strings.ToLower(msg.content), "merged") { - t.Fatalf("expected content with branch name and merged hint, got %q", msg.content) + lowerContent := strings.ToLower(msg.content) + if !strings.Contains(msg.content, "feature/foo") || !strings.Contains(lowerContent, "retry") || strings.Contains(lowerContent, "merged") { + t.Fatalf("expected actionable content with branch name and no unverified merge claim, got %q", msg.content) } - if kind, ok := msg.metadata["failure_kind"].(string); !ok || kind != "missing_pr_branch" { + if kind, ok := msg.metadata["failure_kind"].(string); !ok || kind != "branch_fetch_failed" { t.Fatalf("expected failure_kind metadata, got %#v", msg.metadata["failure_kind"]) } if branch, ok := msg.metadata["missing_branch"].(string); !ok || branch != "feature/foo" { t.Fatalf("expected missing_branch metadata, got %#v", msg.metadata["missing_branch"]) } - // Verify actions array is present with archive and delete actions - actions, ok := msg.metadata["actions"].([]map[string]interface{}) - if !ok || len(actions) != 2 { - t.Fatalf("expected 2 actions, got %#v", msg.metadata["actions"]) + if _, ok := msg.metadata["actions"]; ok { + t.Fatalf("expected neutral guidance without archive/delete actions, got %#v", msg.metadata["actions"]) } - if actions[0]["type"] != "archive_task" { - t.Fatalf("expected first action type archive_task, got %v", actions[0]["type"]) +} + +func TestHandleSessionLaunchFailed_OpenMatchingPRDoesNotCreateMissingBranchGuidance(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateStarting) + + mc := &mockMessageCreator{} + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + svc.messageCreator = mc + svc.SetGitHubService(&mockGitHubService{taskPRs: []*github.TaskPR{ + {TaskID: "task1", RepositoryID: "repo-a", HeadBranch: "feature/foo", State: "open"}, + }}) + + err := errors.New("environment preparation failed: branch \"feature/foo\" not found locally or on remote") + svc.handleSessionLaunchFailed(ctx, "task1", "session1", "repo-a", err) + + if len(mc.sessionMessages) != 0 { + t.Fatalf("expected no persistent missing-branch guidance for an open PR, got %d messages", len(mc.sessionMessages)) } - if actions[1]["type"] != "delete_task" { - t.Fatalf("expected second action type delete_task, got %v", actions[1]["type"]) + if _, ok := svc.suppressToast.Load("session1"); ok { + t.Fatal("expected the ordinary launch error path to remain available") + } +} + +func TestHandleSessionLaunchFailed_SecondaryOpenPRDoesNotUsePrimaryTerminalPR(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateStarting) + + mc := &mockMessageCreator{} + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + svc.messageCreator = mc + svc.SetGitHubService(&mockGitHubService{taskPRs: []*github.TaskPR{ + {TaskID: "task1", RepositoryID: "repo-primary", HeadBranch: "feature/foo", State: "closed"}, + {TaskID: "task1", RepositoryID: "repo-secondary", HeadBranch: "feature/foo", State: "open"}, + }}) + + err := errors.New("environment preparation failed: branch \"feature/foo\" not found locally or on remote") + svc.handleSessionLaunchFailed(ctx, "task1", "session1", "repo-secondary", err) + + if len(mc.sessionMessages) != 0 { + t.Fatalf("expected no missing-branch guidance for the secondary repository's open PR, got %d messages", len(mc.sessionMessages)) + } +} + +func TestHandleSessionLaunchFailed_BoundsBlockingPRStateLookup(t *testing.T) { + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateStarting) + + mc := &mockMessageCreator{} + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + svc.messageCreator = mc + var lookupHadDeadline bool + svc.SetGitHubService(&mockGitHubService{taskPRListFunc: func(ctx context.Context, _ []string) (map[string][]*github.TaskPR, error) { + _, lookupHadDeadline = ctx.Deadline() + if !lookupHadDeadline { + return nil, errors.New("missing lookup deadline") + } + <-ctx.Done() + return nil, ctx.Err() + }}) + + started := time.Now() + err := errors.New("environment preparation failed: branch \"feature/foo\" not found locally or on remote") + svc.handleSessionLaunchFailed(context.Background(), "task1", "session1", "repo-a", err) + + if !lookupHadDeadline { + t.Fatal("expected task PR lookup to have a deadline") + } + if elapsed := time.Since(started); elapsed > 1500*time.Millisecond { + t.Fatalf("blocking task PR lookup took %s, want no more than 1.5s", elapsed) + } + if len(mc.sessionMessages) != 1 { + t.Fatalf("expected neutral guidance after task PR lookup timeout, got %d messages", len(mc.sessionMessages)) + } + if _, ok := mc.sessionMessages[0].metadata["actions"]; ok { + t.Fatalf("expected no destructive actions after task PR lookup timeout, got %#v", mc.sessionMessages[0].metadata["actions"]) + } +} + +func TestHandleSessionLaunchFailed_ConflictingSameBranchPRStatesCreateNeutralGuidance(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateStarting) + + mc := &mockMessageCreator{} + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + svc.messageCreator = mc + svc.SetGitHubService(&mockGitHubService{taskPRs: []*github.TaskPR{ + {TaskID: "task1", RepositoryID: "repo-a", HeadBranch: "feature/foo", State: "open"}, + {TaskID: "task1", RepositoryID: "repo-a", HeadBranch: "feature/foo", State: "closed"}, + }}) + + err := errors.New("environment preparation failed: branch \"feature/foo\" not found locally or on remote") + svc.handleSessionLaunchFailed(ctx, "task1", "session1", "repo-a", err) + + if len(mc.sessionMessages) != 1 { + t.Fatalf("expected neutral guidance for ambiguous PR state, got %d messages", len(mc.sessionMessages)) + } + msg := mc.sessionMessages[0] + if !strings.Contains(strings.ToLower(msg.content), "retry") || strings.Contains(msg.content, "no longer exists") { + t.Fatalf("expected neutral fetch-failure guidance, got %q", msg.content) + } + if kind := msg.metadata["failure_kind"]; kind != "branch_fetch_failed" { + t.Fatalf("expected neutral failure kind, got %#v", kind) + } + if _, ok := msg.metadata["actions"]; ok { + t.Fatalf("expected no destructive actions for ambiguous PR state, got %#v", msg.metadata["actions"]) + } +} + +func TestHandleSessionLaunchFailed_MultipleTerminalMatchingPRsCreateNeutralGuidance(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateStarting) + + mc := &mockMessageCreator{} + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + svc.messageCreator = mc + svc.SetGitHubService(&mockGitHubService{taskPRs: []*github.TaskPR{ + {TaskID: "task1", RepositoryID: "repo-a", HeadBranch: "feature/foo", State: "closed"}, + {TaskID: "task1", RepositoryID: "repo-a", HeadBranch: "feature/foo", State: "merged"}, + }}) + + err := errors.New("environment preparation failed: branch \"feature/foo\" not found locally or on remote") + svc.handleSessionLaunchFailed(ctx, "task1", "session1", "repo-a", err) + + if len(mc.sessionMessages) != 1 { + t.Fatalf("expected neutral guidance for ambiguous terminal PRs, got %d messages", len(mc.sessionMessages)) + } + msg := mc.sessionMessages[0] + if !strings.Contains(strings.ToLower(msg.content), "retry") || strings.Contains(msg.content, "no longer exists") { + t.Fatalf("expected neutral fetch-failure guidance, got %q", msg.content) + } + if kind := msg.metadata["failure_kind"]; kind != "branch_fetch_failed" { + t.Fatalf("expected neutral failure kind, got %#v", kind) + } + if _, ok := msg.metadata["actions"]; ok { + t.Fatalf("expected no destructive actions for ambiguous terminal PRs, got %#v", msg.metadata["actions"]) + } +} + +func TestHandleSessionLaunchFailed_ClosedOrMergedMatchingPRCreatesMissingBranchGuidance(t *testing.T) { + for _, prState := range []string{"closed", "merged"} { + t.Run(prState, func(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateStarting) + + mc := &mockMessageCreator{} + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + svc.messageCreator = mc + svc.SetGitHubService(&mockGitHubService{taskPRs: []*github.TaskPR{ + {TaskID: "task1", RepositoryID: "repo-a", HeadBranch: "feature/foo", State: prState}, + }}) + + err := errors.New("environment preparation failed: branch \"feature/foo\" not found locally or on remote") + svc.handleSessionLaunchFailed(ctx, "task1", "session1", "repo-a", err) + + if len(mc.sessionMessages) != 1 { + t.Fatalf("expected 1 missing-branch guidance message, got %d", len(mc.sessionMessages)) + } + msg := mc.sessionMessages[0] + if !strings.Contains(msg.content, "feature/foo") || !strings.Contains(msg.content, "no longer exists") { + t.Fatalf("expected authoritative missing-branch guidance, got %q", msg.content) + } + actions, ok := msg.metadata["actions"].([]map[string]interface{}) + if !ok || len(actions) != 2 { + t.Fatalf("expected archive/delete actions, got %#v", msg.metadata["actions"]) + } + if suppressed, ok := svc.suppressToast.Load("session1"); !ok || suppressed != true { + t.Fatalf("expected duplicate launch toast to be suppressed, got ok=%v value=%v", ok, suppressed) + } + }) + } +} + +func TestHandleSessionLaunchFailed_UnrelatedOpenPRDoesNotSuppressNeutralGuidance(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateStarting) + + mc := &mockMessageCreator{} + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + svc.messageCreator = mc + svc.SetGitHubService(&mockGitHubService{taskPRs: []*github.TaskPR{ + {TaskID: "task1", RepositoryID: "repo-a", HeadBranch: "feature/other", State: "open"}, + }}) + + err := errors.New("environment preparation failed: branch \"feature/foo\" not found locally or on remote") + svc.handleSessionLaunchFailed(ctx, "task1", "session1", "repo-a", err) + + if len(mc.sessionMessages) != 1 { + t.Fatalf("expected neutral guidance for the unmatched branch, got %d messages", len(mc.sessionMessages)) + } + msg := mc.sessionMessages[0] + if !strings.Contains(strings.ToLower(msg.content), "retry") || strings.Contains(msg.content, "no longer exists") { + t.Fatalf("expected neutral fetch-failure guidance, got %q", msg.content) + } + if kind := msg.metadata["failure_kind"]; kind != "branch_fetch_failed" { + t.Fatalf("expected neutral failure kind, got %#v", kind) + } +} + +func TestHandleSessionLaunchFailed_ClosedPRInAnotherRepositoryCreatesNeutralGuidance(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateStarting) + + mc := &mockMessageCreator{} + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + svc.messageCreator = mc + svc.SetGitHubService(&mockGitHubService{taskPRs: []*github.TaskPR{ + {TaskID: "task1", RepositoryID: "repo-a", HeadBranch: "feature/foo", State: "closed"}, + }}) + + err := errors.New("environment preparation failed: branch \"feature/foo\" not found locally or on remote") + svc.handleSessionLaunchFailed(ctx, "task1", "session1", "repo-b", err) + + if len(mc.sessionMessages) != 1 { + t.Fatalf("expected neutral guidance when repo-b launch matches only repo-a's closed PR, got %d messages", len(mc.sessionMessages)) + } + msg := mc.sessionMessages[0] + if !strings.Contains(strings.ToLower(msg.content), "retry") || strings.Contains(msg.content, "no longer exists") { + t.Fatalf("expected neutral fetch-failure guidance, got %q", msg.content) + } + if kind := msg.metadata["failure_kind"]; kind != "branch_fetch_failed" { + t.Fatalf("expected neutral failure kind, got %#v", kind) + } + if _, ok := msg.metadata["actions"]; ok { + t.Fatalf("expected no destructive actions without a repository-scoped PR match, got %#v", msg.metadata["actions"]) + } +} + +func TestHandleSessionLaunchFailed_SingleRepositoryLegacyPRCreatesMissingBranchGuidance(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateStarting) + if err := repo.CreateRepository(ctx, &models.Repository{ + ID: "repo-a", WorkspaceID: "ws1", Name: "repo-a", SourceType: "local", + }); err != nil { + t.Fatalf("create repository: %v", err) + } + if err := repo.CreateTaskRepository(ctx, &models.TaskRepository{ + ID: "task-repo-a", TaskID: "task1", RepositoryID: "repo-a", Position: 0, + }); err != nil { + t.Fatalf("link task repository: %v", err) + } + + mc := &mockMessageCreator{} + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + svc.messageCreator = mc + svc.SetGitHubService(&mockGitHubService{taskPRs: []*github.TaskPR{ + {TaskID: "task1", HeadBranch: "feature/foo", State: "closed"}, + }}) + + err := errors.New("environment preparation failed: branch \"feature/foo\" not found locally or on remote") + svc.handleSessionLaunchFailed(ctx, "task1", "session1", "repo-a", err) + + if len(mc.sessionMessages) != 1 { + t.Fatalf("expected 1 missing-branch guidance message, got %d", len(mc.sessionMessages)) + } + msg := mc.sessionMessages[0] + if !strings.Contains(msg.content, "feature/foo") || !strings.Contains(msg.content, "no longer exists") { + t.Fatalf("expected authoritative missing-branch guidance, got %q", msg.content) + } + if actions, ok := msg.metadata["actions"].([]map[string]interface{}); !ok || len(actions) != 2 { + t.Fatalf("expected archive/delete actions, got %#v", msg.metadata["actions"]) + } +} + +func TestHandleSessionLaunchFailed_LegacyPRForDifferentRepositoryCreatesNeutralGuidance(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateStarting) + if err := repo.CreateRepository(ctx, &models.Repository{ + ID: "repo-a", WorkspaceID: "ws1", Name: "repo-a", SourceType: "local", + }); err != nil { + t.Fatalf("create repository: %v", err) + } + if err := repo.CreateTaskRepository(ctx, &models.TaskRepository{ + ID: "task-repo-a", TaskID: "task1", RepositoryID: "repo-a", Position: 0, + }); err != nil { + t.Fatalf("link task repository: %v", err) + } + + mc := &mockMessageCreator{} + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + svc.messageCreator = mc + svc.SetGitHubService(&mockGitHubService{taskPRs: []*github.TaskPR{ + {TaskID: "task1", HeadBranch: "feature/foo", State: "closed"}, + }}) + + err := errors.New("environment preparation failed: branch \"feature/foo\" not found locally or on remote") + svc.handleSessionLaunchFailed(ctx, "task1", "session1", "repo-b", err) + + if len(mc.sessionMessages) != 1 { + t.Fatalf("expected 1 neutral fetch-failure guidance message, got %d", len(mc.sessionMessages)) + } + msg := mc.sessionMessages[0] + if kind := msg.metadata["failure_kind"]; kind != "branch_fetch_failed" { + t.Fatalf("expected neutral failure kind, got %#v", kind) + } + if _, ok := msg.metadata["actions"]; ok { + t.Fatalf("expected no destructive actions for a legacy PR linked to another repository, got %#v", msg.metadata["actions"]) + } +} + +func TestHandleSessionLaunchFailed_MultipleRepositoriesLegacyPRCreatesNeutralGuidance(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateStarting) + for position, repositoryID := range []string{"repo-a", "repo-b"} { + if err := repo.CreateRepository(ctx, &models.Repository{ + ID: repositoryID, WorkspaceID: "ws1", Name: repositoryID, SourceType: "local", + }); err != nil { + t.Fatalf("create repository %q: %v", repositoryID, err) + } + if err := repo.CreateTaskRepository(ctx, &models.TaskRepository{ + ID: "task-repo-" + repositoryID, TaskID: "task1", RepositoryID: repositoryID, Position: position, + }); err != nil { + t.Fatalf("link task repository %q: %v", repositoryID, err) + } + } + + mc := &mockMessageCreator{} + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + svc.messageCreator = mc + svc.SetGitHubService(&mockGitHubService{taskPRs: []*github.TaskPR{ + {TaskID: "task1", HeadBranch: "feature/foo", State: "closed"}, + }}) + + err := errors.New("environment preparation failed: branch \"feature/foo\" not found locally or on remote") + svc.handleSessionLaunchFailed(ctx, "task1", "session1", "repo-a", err) + + if len(mc.sessionMessages) != 1 { + t.Fatalf("expected 1 neutral fetch-failure guidance message, got %d", len(mc.sessionMessages)) + } + msg := mc.sessionMessages[0] + if kind := msg.metadata["failure_kind"]; kind != "branch_fetch_failed" { + t.Fatalf("expected neutral failure kind, got %#v", kind) + } + if _, ok := msg.metadata["actions"]; ok { + t.Fatalf("expected no destructive actions for an unscoped multi-repository PR, got %#v", msg.metadata["actions"]) + } +} + +func TestHandleSessionLaunchFailed_EmptyRepositoryIDCreatesNeutralGuidance(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateStarting) + + mc := &mockMessageCreator{} + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + svc.messageCreator = mc + svc.SetGitHubService(&mockGitHubService{taskPRs: []*github.TaskPR{ + {TaskID: "task1", RepositoryID: "repo-a", HeadBranch: "feature/foo", State: "closed"}, + }}) + + err := errors.New("environment preparation failed: branch \"feature/foo\" not found locally or on remote") + svc.handleSessionLaunchFailed(ctx, "task1", "session1", "", err) + + if len(mc.sessionMessages) != 1 { + t.Fatalf("expected neutral guidance without a failed repository ID, got %d messages", len(mc.sessionMessages)) + } + msg := mc.sessionMessages[0] + if !strings.Contains(strings.ToLower(msg.content), "retry") || strings.Contains(msg.content, "no longer exists") { + t.Fatalf("expected neutral fetch-failure guidance, got %q", msg.content) + } + if kind := msg.metadata["failure_kind"]; kind != "branch_fetch_failed" { + t.Fatalf("expected neutral failure kind, got %#v", kind) + } + if _, ok := msg.metadata["actions"]; ok { + t.Fatalf("expected no destructive actions without a repository-scoped PR match, got %#v", msg.metadata["actions"]) } } @@ -144,7 +516,7 @@ func TestHandleSessionLaunchFailed_IgnoresUnrelatedErrors(t *testing.T) { svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) svc.messageCreator = mc - svc.handleSessionLaunchFailed(ctx, "task1", "session1", errors.New("failed to launch container")) + svc.handleSessionLaunchFailed(ctx, "task1", "session1", "repo-a", errors.New("failed to launch container")) if len(mc.sessionMessages) != 0 { t.Fatalf("expected no session messages for unrelated error, got %d", len(mc.sessionMessages))