fix: scope PR branch failure guidance (#1873)
This commit is contained in:
@@ -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")
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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))
|
||||
|
||||
Reference in New Issue
Block a user