diff --git a/apps/backend/internal/orchestrator/dynamic_launch.go b/apps/backend/internal/orchestrator/dynamic_launch.go index 27ea630a57..b32c385358 100644 --- a/apps/backend/internal/orchestrator/dynamic_launch.go +++ b/apps/backend/internal/orchestrator/dynamic_launch.go @@ -7,6 +7,7 @@ import ( "errors" "fmt" "strings" + "time" "github.com/kandev/kandev/internal/agent/planinjection" agentruntime "github.com/kandev/kandev/internal/agent/runtime" @@ -21,6 +22,12 @@ import ( const dynamicRouteStatusWaiting = "waiting" +// dynamicSuccessorDetachedTimeout bounds one detached fallback launch: the +// predecessor stop plus the successor launch. Without it the repository calls +// inside the launch could hold the session's cancel guard indefinitely and +// block every later stop or message for that session. +const dynamicSuccessorDetachedTimeout = 2 * time.Minute + // dynamicTaskDownstream adapts the task executor to the provider-neutral // conductor. The callback updates the task-session attribution before every // concrete launch, including a cross-profile fallback. @@ -575,7 +582,155 @@ func (s *Service) launchDynamicSuccessorAfterFailure( zap.String("session_id", session.ID), zap.Error(err)) return false } - return s.relaunchDynamicTaskAfterFailure(ctx, data, next.ExecutionProfileID) + return s.launchDynamicSuccessorDetached(ctx, data, next.ExecutionProfileID) +} + +// launchDynamicSuccessorDetached runs the predecessor stop and successor +// launch outside the agent.failed dispatch that decided the route. +// +// The failure event is published synchronously by the lifecycle completion +// handler while it still holds the execution's prompt lifecycle lock, and the +// in-memory event bus delivers it on the publisher's goroutine. Stopping the +// predecessor from that goroutine publishes agent.stopped for the same +// execution, which needs the same lock to snapshot the prompt turn: the +// dispatch deadlocks, the successor never starts, and the session stays +// RUNNING without a process. Leaving the dispatch first lets the completion +// handler release the lock before the stop runs. +// The worker runs under the service-owned dynamicSuccessorCtx rather than +// context.WithoutCancel(ctx): the launch must survive the dispatch that +// scheduled it, but it must not outlive Stop and keep mutating session state +// after shutdown has begun. Returning false means no worker was scheduled, so +// the caller falls through to the ordinary terminal-failure path instead of +// reporting a route that will never be taken. +func (s *Service) launchDynamicSuccessorDetached( + _ context.Context, + data watcher.AgentEventData, + executionProfileID string, +) bool { + s.dynamicSuccessorMu.Lock() + if s.dynamicSuccessorStopped { + s.dynamicSuccessorMu.Unlock() + s.logger.Warn("dynamic successor launch skipped; service is stopping", + zap.String("task_id", data.TaskID), + zap.String("session_id", data.SessionID), + zap.String("execution_profile_id", executionProfileID)) + return false + } + if s.dynamicSuccessorCtx == nil { + s.dynamicSuccessorCtx, s.dynamicSuccessorCancel = context.WithCancel(context.Background()) + } + workerCtx := s.dynamicSuccessorCtx + s.dynamicSuccessorWorkers.Add(1) + s.dynamicSuccessorMu.Unlock() + go func() { + defer s.dynamicSuccessorWorkers.Done() + launchCtx, cancel := context.WithTimeout(workerCtx, dynamicSuccessorDetachedTimeout) + defer cancel() + s.runDetachedDynamicSuccessorLaunch(launchCtx, data, executionProfileID) + }() + return true +} + +// runDetachedDynamicSuccessorLaunch serializes with the session's cancel guard +// like every other session-backed failure decision, then launches the +// successor. A launch that cannot complete falls back to the ordinary +// recoverable-failure surface instead of leaving the session RUNNING. +func (s *Service) runDetachedDynamicSuccessorLaunch( + ctx context.Context, + data watcher.AgentEventData, + executionProfileID string, +) { + lock, release := s.acquireCancelInFlightGuard(data.SessionID) + defer release() + lock.Lock() + defer lock.Unlock() + if ctx.Err() != nil { + s.logger.Warn("dynamic successor launch abandoned before validation", + zap.String("task_id", data.TaskID), + zap.String("session_id", data.SessionID), + zap.String("execution_profile_id", executionProfileID), + zap.Error(ctx.Err())) + return + } + // A coordinator stop can win the guard between the route decision and this + // goroutine: it persists the session as cancelled and releases the guard + // before the launch acquires it. Relaunching from the stale event would + // reset that session back to CREATED and resurrect work after an + // acknowledged stop, so re-read the current state under the guard and + // abort unless the session is still nonterminal and still owns this + // execution. + if drop, _ := s.shouldDropSessionFailure(ctx, data, "dynamic.successor.detached", true); drop { + s.logger.Debug("dropping detached dynamic successor launch for stale session", + zap.String("task_id", data.TaskID), + zap.String("session_id", data.SessionID), + zap.String("agent_execution_id", data.AgentExecutionID), + zap.String("execution_profile_id", executionProfileID)) + return + } + if ctx.Err() != nil { + s.logger.Warn("dynamic successor launch abandoned before start", + zap.String("task_id", data.TaskID), + zap.String("session_id", data.SessionID), + zap.String("execution_profile_id", executionProfileID), + zap.Error(ctx.Err())) + return + } + if s.relaunchDynamicTaskAfterFailure(ctx, data, executionProfileID) { + return + } + if errors.Is(ctx.Err(), context.Canceled) { + s.logger.Warn("dynamic successor launch abandoned after cancellation", + zap.String("task_id", data.TaskID), + zap.String("session_id", data.SessionID), + zap.String("execution_profile_id", executionProfileID), + zap.Error(ctx.Err())) + return + } + s.logger.Warn("dynamic successor launch failed; surfacing recoverable failure", + zap.String("task_id", data.TaskID), + zap.String("session_id", data.SessionID), + zap.String("agent_execution_id", data.AgentExecutionID), + zap.String("execution_profile_id", executionProfileID)) + // The synchronous failure path finalized the automation run before + // surfacing the recoverable failure. handleAgentFailedLocked already + // returned here because the route was accepted, so this branch owns that + // finalization: without it an automation run stays nonterminal and holds a + // max_concurrent_runs slot and its worktree forever. + failureCtx := context.WithoutCancel(ctx) + s.retireExecutionActivityAndPublish(failureCtx, data.TaskID, data.SessionID, data.AgentExecutionID) + errMsg := data.ErrorMessage + if errMsg == "" { + errMsg = "dynamic successor launch failed" + } + s.finalizeAutomationRun(failureCtx, data.TaskID, false, errMsg) + s.handleRecoverableFailureLocked(failureCtx, data) +} + +func (s *Service) resetDynamicSuccessorWorkers() { + s.dynamicSuccessorMu.Lock() + defer s.dynamicSuccessorMu.Unlock() + s.dynamicSuccessorStopped = false + s.dynamicSuccessorCtx, s.dynamicSuccessorCancel = context.WithCancel(context.Background()) +} + +func (s *Service) stopDynamicSuccessorWorkers() { + s.dynamicSuccessorMu.Lock() + s.dynamicSuccessorStopped = true + cancel := s.dynamicSuccessorCancel + s.dynamicSuccessorMu.Unlock() + if cancel != nil { + cancel() + } + done := make(chan struct{}) + go func() { + s.dynamicSuccessorWorkers.Wait() + close(done) + }() + select { + case <-done: + case <-time.After(sendNowClaimRecoveryTimeout): + s.logger.Warn("timed out waiting for dynamic successor workers during shutdown") + } } // LaunchDynamicRouteAction completes a manual retry/try-next operation after diff --git a/apps/backend/internal/orchestrator/dynamic_launch_test.go b/apps/backend/internal/orchestrator/dynamic_launch_test.go index 1f9d77962f..2f862ce266 100644 --- a/apps/backend/internal/orchestrator/dynamic_launch_test.go +++ b/apps/backend/internal/orchestrator/dynamic_launch_test.go @@ -3,7 +3,9 @@ package orchestrator import ( "context" "errors" + "sync" "testing" + "time" "github.com/kandev/kandev/internal/orchestrator/watcher" "github.com/kandev/kandev/internal/task/models" @@ -64,3 +66,343 @@ func TestRelaunchDynamicTaskAfterFailure_DoesNotLaunchSuccessorWhenStopFails(t * t.Fatalf("unexpected stop call: %#v", agentManager.stopAgentWithReasonArgs[0]) } } + +// The agent.failed event that selects a fallback route is dispatched on the +// lifecycle completion goroutine while it holds the execution's prompt +// lifecycle lock. Stopping the predecessor needs that lock again, so the +// successor launch must leave the dispatch before the stop runs. +func TestLaunchDynamicSuccessorDetached_ReturnsWhileStopIsBlocked(t *testing.T) { + ctx := context.Background() + const ( + taskID = "task-dynamic-detached" + sessionID = "session-dynamic-detached" + executionID = "execution-dynamic-detached" + ) + + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateRunning) + seedExecutorRunning(t, repo, sessionID, taskID, executionID) + taskRepo := newMockTaskRepo() + seedMockTaskState(taskRepo, taskID, v1.TaskStateInProgress) + + stopEntered := make(chan struct{}) + releaseStop := make(chan struct{}) + stopErr := errors.New("runtime teardown failed") + var firstStop sync.Once + agentManager := &mockAgentManager{ + // Only the fallback stop blocks; the recoverable-failure cleanup that + // follows a failed launch stops the same execution again and must not. + stopAgentWithReasonFunc: func(context.Context, string, string, bool) error { + firstStop.Do(func() { + close(stopEntered) + <-releaseStop + }) + return stopErr + }, + } + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentManager) + svc.lastTurnPrompt.Store(sessionID, capturedPrompt{text: "retry the task"}) + data := watcher.AgentEventData{ + TaskID: taskID, + SessionID: sessionID, + AgentExecutionID: executionID, + } + + returned := make(chan struct{}) + go func() { + svc.launchDynamicSuccessorDetached(ctx, data, "fallback-profile") + close(returned) + }() + select { + case <-returned: + case <-time.After(2 * time.Second): + t.Fatal("launchDynamicSuccessorDetached blocked on the predecessor stop") + } + select { + case <-stopEntered: + case <-time.After(2 * time.Second): + t.Fatal("predecessor stop never started after the dispatch returned") + } + session, err := repo.GetTaskSession(ctx, sessionID) + if err != nil { + t.Fatalf("get session: %v", err) + } + if session.State != models.TaskSessionStateRunning { + t.Fatalf("session state = %q before the stop resolved, want RUNNING", session.State) + } + + close(releaseStop) + waitForSessionState(t, repo, sessionID, models.TaskSessionStateWaitingForInput) + if len(agentManager.startAgentProcessCalls) != 0 { + t.Fatalf("successor launch started %d processes after stop failure", len(agentManager.startAgentProcessCalls)) + } +} + +func TestRunDetachedDynamicSuccessorLaunch_ParksSessionWhenRelaunchFails(t *testing.T) { + ctx := context.Background() + const ( + taskID = "task-dynamic-detached-failure" + sessionID = "session-dynamic-detached-failure" + executionID = "execution-dynamic-detached-failure" + ) + + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateRunning) + seedExecutorRunning(t, repo, sessionID, taskID, executionID) + taskRepo := newMockTaskRepo() + seedMockTaskState(taskRepo, taskID, v1.TaskStateInProgress) + agentManager := &mockAgentManager{ + stopAgentWithReasonErr: errors.New("runtime teardown failed"), + } + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentManager) + svc.lastTurnPrompt.Store(sessionID, capturedPrompt{text: "retry the task"}) + + svc.runDetachedDynamicSuccessorLaunch(ctx, watcher.AgentEventData{ + TaskID: taskID, + SessionID: sessionID, + AgentExecutionID: executionID, + ErrorMessage: "provider quota exhausted", + }, "fallback-profile") + + session, err := repo.GetTaskSession(ctx, sessionID) + if err != nil { + t.Fatalf("get session: %v", err) + } + if session.State != models.TaskSessionStateWaitingForInput { + t.Fatalf("session state = %q, want WAITING_FOR_INPUT so the user can resume", session.State) + } + if len(agentManager.startAgentProcessCalls) != 0 { + t.Fatalf("successor launch started %d processes after stop failure", len(agentManager.startAgentProcessCalls)) + } +} + +func waitForSessionState(t *testing.T, repo taskSessionStateReader, sessionID string, want models.TaskSessionState) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for { + session, err := repo.GetTaskSession(context.Background(), sessionID) + if err != nil { + t.Fatalf("get session: %v", err) + } + if session.State == want { + return + } + if time.Now().After(deadline) { + t.Fatalf("session state = %q, want %q", session.State, want) + } + time.Sleep(20 * time.Millisecond) + } +} + +type taskSessionStateReader interface { + GetTaskSession(ctx context.Context, id string) (*models.TaskSession, error) +} + +// A coordinator stop can win the session's cancel guard between the route +// decision and the detached launch: it persists CANCELLED and releases the +// guard before the launch acquires it. Relaunching from that stale event would +// reset the session to CREATED and resurrect work the user already stopped. +func TestRunDetachedDynamicSuccessorLaunch_DropsCancelledSession(t *testing.T) { + ctx := context.Background() + const ( + taskID = "task-dynamic-detached-cancelled" + sessionID = "session-dynamic-detached-cancelled" + executionID = "execution-dynamic-detached-cancelled" + ) + + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateCancelled) + seedExecutorRunning(t, repo, sessionID, taskID, executionID) + taskRepo := newMockTaskRepo() + seedMockTaskState(taskRepo, taskID, v1.TaskStateInProgress) + // Dropping a stale failure also retires the predecessor, and that teardown + // runs on its own goroutine with reason "agent completed". Count only the + // relaunch's own stop, under a mutex, so the assertion neither races that + // cleanup nor mistakes it for a resurrection. + var stopMu sync.Mutex + fallbackStops := 0 + agentManager := &mockAgentManager{ + stopAgentWithReasonFunc: func(_ context.Context, _ string, reason string, _ bool) error { + if reason == "dynamic route fallback" { + stopMu.Lock() + fallbackStops++ + stopMu.Unlock() + } + return nil + }, + } + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentManager) + svc.lastTurnPrompt.Store(sessionID, capturedPrompt{text: "retry the task"}) + + svc.runDetachedDynamicSuccessorLaunch(ctx, watcher.AgentEventData{ + TaskID: taskID, + SessionID: sessionID, + AgentExecutionID: executionID, + }, "fallback-profile") + + // The relaunch path opens by stopping the predecessor and then resets the + // session to CREATED, both before the launch can fail. That stop is + // synchronous, so its absence on return proves the launch was refused + // before it touched the cancelled session — which a final-state assertion + // cannot show, because the failure branch restores a state a resurrection + // has already passed through. + stopMu.Lock() + stops := fallbackStops + stopMu.Unlock() + if stops != 0 { + t.Fatalf("detached launch stopped the predecessor of a cancelled session %d times", stops) + } + session, err := repo.GetTaskSession(ctx, sessionID) + if err != nil { + t.Fatalf("get session: %v", err) + } + if session.State != models.TaskSessionStateCancelled { + t.Fatalf("session state = %q, want CANCELLED to survive the detached launch", session.State) + } +} + +// handleAgentFailedLocked returned early because the route was accepted, so +// the detached failure branch owns the automation finalization the synchronous +// path used to reach. Without it the run stays nonterminal and holds a +// max_concurrent_runs slot forever. +func TestRunDetachedDynamicSuccessorLaunch_FinalizesAutomationRunOnFailure(t *testing.T) { + ctx := context.Background() + const ( + taskID = "task-dynamic-detached-automation" + sessionID = "session-dynamic-detached-automation" + executionID = "execution-dynamic-detached-automation" + ) + + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateRunning) + task, err := repo.GetTask(ctx, taskID) + if err != nil { + t.Fatalf("get task: %v", err) + } + task.Origin = models.TaskOriginAutomationRun + if err := repo.UpdateTask(ctx, task); err != nil { + t.Fatalf("update task origin: %v", err) + } + seedExecutorRunning(t, repo, sessionID, taskID, executionID) + taskRepo := newMockTaskRepo() + seedMockTaskState(taskRepo, taskID, v1.TaskStateInProgress) + agentManager := &mockAgentManager{ + stopAgentWithReasonErr: errors.New("runtime teardown failed"), + } + autoSvc := &stubAutomationService{} + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentManager) + svc.SetAutomationService(autoSvc) + svc.lastTurnPrompt.Store(sessionID, capturedPrompt{text: "retry the task"}) + + svc.runDetachedDynamicSuccessorLaunch(ctx, watcher.AgentEventData{ + TaskID: taskID, + SessionID: sessionID, + AgentExecutionID: executionID, + ErrorMessage: "provider quota exhausted", + }, "fallback-profile") + + if autoSvc.failed[taskID] != "provider quota exhausted" { + t.Fatalf("automation run failure = %q, want the launch error so the run leaves task_created", + autoSvc.failed[taskID]) + } +} + +// A service shutdown can cancel the detached worker while predecessor teardown +// is in progress. The worker must not surface a second failure after shutdown, +// because that would mutate the session after the service-owned cancellation. +func TestRunDetachedDynamicSuccessorLaunch_DoesNotRecoverAfterCancellation(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + const ( + taskID = "task-dynamic-detached-shutdown" + sessionID = "session-dynamic-detached-shutdown" + executionID = "execution-dynamic-detached-shutdown" + ) + + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateRunning) + seedExecutorRunning(t, repo, sessionID, taskID, executionID) + taskRepo := newMockTaskRepo() + seedMockTaskState(taskRepo, taskID, v1.TaskStateInProgress) + + stopEntered := make(chan struct{}) + releaseStop := make(chan struct{}) + stopErr := errors.New("runtime teardown failed") + var stopStartOnce sync.Once + agentManager := &mockAgentManager{ + repoForExecutionLookup: repo, + stopAgentWithReasonFunc: func(_ context.Context, _ string, _ string, _ bool) error { + stopStartOnce.Do(func() { close(stopEntered) }) + <-releaseStop + return stopErr + }, + } + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentManager) + svc.lastTurnPrompt.Store(sessionID, capturedPrompt{text: "retry the task"}) + + done := make(chan struct{}) + go func() { + svc.runDetachedDynamicSuccessorLaunch(ctx, watcher.AgentEventData{ + TaskID: taskID, + SessionID: sessionID, + AgentExecutionID: executionID, + ErrorMessage: "provider quota exhausted", + }, "fallback-profile") + close(done) + }() + + select { + case <-stopEntered: + case <-time.After(2 * time.Second): + t.Fatal("predecessor stop did not start") + } + cancel() + close(releaseStop) + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("detached launch did not finish after cancellation") + } + + session, err := repo.GetTaskSession(context.Background(), sessionID) + if err != nil { + t.Fatalf("get session: %v", err) + } + if session.State != models.TaskSessionStateRunning { + t.Fatalf("session state = %q, want RUNNING after shutdown cancellation", session.State) + } + if len(agentManager.startAgentProcessCalls) != 0 { + t.Fatalf("successor launch started %d processes after shutdown cancellation", len(agentManager.startAgentProcessCalls)) + } +} + +// context.WithoutCancel would let the worker keep mutating session state after +// Stop began. The launch runs under a service-owned context instead, so a +// stopped service schedules nothing and reports the route as not taken. +func TestLaunchDynamicSuccessorDetached_SkipsWhenServiceStopped(t *testing.T) { + const ( + taskID = "task-dynamic-detached-stopped" + sessionID = "session-dynamic-detached-stopped" + ) + + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateRunning) + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), newMockTaskRepo(), &mockAgentManager{}) + svc.stopDynamicSuccessorWorkers() + + if svc.launchDynamicSuccessorDetached(context.Background(), watcher.AgentEventData{ + TaskID: taskID, + SessionID: sessionID, + }, "fallback-profile") { + t.Fatal("a stopped service reported a detached successor launch it will never run") + } + + svc.resetDynamicSuccessorWorkers() + if !svc.launchDynamicSuccessorDetached(context.Background(), watcher.AgentEventData{ + TaskID: taskID, + SessionID: sessionID, + }, "fallback-profile") { + t.Fatal("a restarted service refused to schedule the detached successor launch") + } + svc.stopDynamicSuccessorWorkers() +} diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index f03d31eee6..c72bf8fe91 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -1145,6 +1145,17 @@ type Service struct { ciAutomationCancel context.CancelFunc ciAutomationStopped bool ciAutomationWorkers sync.WaitGroup + + // dynamicSuccessorWorkers owns the detached dynamic fallback launches. The + // launch has to leave the agent.failed dispatch to avoid the prompt + // lifecycle deadlock, but a detached goroutine must still stop mutating + // session state once Stop begins, so it runs under a service-owned context + // instead of an unbounded context.WithoutCancel. + dynamicSuccessorMu sync.Mutex + dynamicSuccessorCtx context.Context + dynamicSuccessorCancel context.CancelFunc + dynamicSuccessorStopped bool + dynamicSuccessorWorkers sync.WaitGroup } func (s *Service) officeStallDependencies() ( @@ -1324,6 +1335,7 @@ func NewService( // Create the service (watcher will be created after we have handlers) sendNowCtx, sendNowCancel := context.WithCancel(context.Background()) ciAutomationCtx, ciAutomationCancel := context.WithCancel(context.Background()) + dynamicSuccessorCtx, dynamicSuccessorCancel := context.WithCancel(context.Background()) s := &Service{ config: cfg, logger: svcLogger, @@ -1345,6 +1357,8 @@ func NewService( sendNowCancel: sendNowCancel, ciAutomationCtx: ciAutomationCtx, ciAutomationCancel: ciAutomationCancel, + dynamicSuccessorCtx: dynamicSuccessorCtx, + dynamicSuccessorCancel: dynamicSuccessorCancel, idleReaper: newIdleSessionReaper(), } // Always publish queue-status after a task-scoped queue purge so the @@ -2497,6 +2511,7 @@ func (s *Service) Start(ctx context.Context) error { s.resetReservedPromptCallbacks() s.resetSendNowWorkers() s.resetCIAutomationWorkers() + s.resetDynamicSuccessorWorkers() // Reconcile session state from persisted runtime state on startup. // This does NOT launch any agent processes — sessions are recovered lazily @@ -2610,6 +2625,10 @@ func (s *Service) Stop() error { s.mu.Unlock() s.logger.Info("stopping orchestrator service") + // Stop detached dynamic successors before the scheduler and watcher. Their + // workers can otherwise observe the shutdown only after those components + // have already stopped, and may launch or recover a session during teardown. + s.stopDynamicSuccessorWorkers() s.stopDynamicPolicyRecovery() // Stop components in reverse order