fix(orchestrator): launch the dynamic fallback successor outside the agent.failed dispatch (#3356)

Co-authored-by: Artem Shtanov <artemshtanov@gmail.com>
Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>

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

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()
+}

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