fix: break ceiling replay boot deadlock (#3776)

apps/backend/internal/orchestrator/ceiling_boot_deadlock_test.go

@@ -0,0 +1,236 @@
+package orchestrator
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/kandev/kandev/internal/orchestrator/executor"
+	"github.com/kandev/kandev/internal/task/models"
+	wfmodels "github.com/kandev/kandev/internal/workflow/models"
+	v1 "github.com/kandev/kandev/pkg/api/v1"
+)
+
+type gatedPromptAgentManager struct {
+	*mockAgentManager
+	beforePrompt chan struct{}
+	allowPrompt  chan struct{}
+	allowOnce    sync.Once
+}
+
+func (m *gatedPromptAgentManager) releasePrompt() {
+	m.allowOnce.Do(func() { close(m.allowPrompt) })
+}
+
+func (m *gatedPromptAgentManager) PromptAgentWithDispatchCallback(
+	ctx context.Context,
+	executionID, prompt string,
+	attachments []v1.MessageAttachment,
+	dispatchOnly bool,
+	onDispatched func(),
+) (*executor.PromptResult, error) {
+	close(m.beforePrompt)
+	<-m.allowPrompt
+	return m.mockAgentManager.PromptAgentWithDispatchCallback(
+		ctx, executionID, prompt, attachments, dispatchOnly, onDispatched,
+	)
+}
+
+// @covers AC-AGENTS-SESSION-CEILING-001.5
+func TestCeilingReplayReleasesAdmissionBeforeProviderDispatch(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedTaskAndSession(t, repo, "replay-task", "replay-session", models.TaskSessionStateWaitingForInput)
+	seedExecutorRunning(t, repo, "replay-session", "replay-task", "replay-execution")
+	taskRepo := newMockTaskRepo()
+	seedMockTaskState(taskRepo, "replay-task", v1.TaskStateScheduling)
+	agent := &mockAgentManager{repoForExecutionLookup: repo, isAgentRunning: true}
+	svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agent)
+	providerEntered := make(chan bool, 1)
+	providerRelease := make(chan struct{})
+	var releaseOnce sync.Once
+	releaseProvider := func() { releaseOnce.Do(func() { close(providerRelease) }) }
+	agent.promptAgentFunc = func(callCtx context.Context, _, _ string, _ []v1.MessageAttachment, _ bool) (*executor.PromptResult, error) {
+		providerEntered <- ceilingEntryAdmissionLockHeld(callCtx, "replay-task")
+		<-providerRelease
+		return &executor.PromptResult{}, nil
+	}
+	result := make(chan ceilingReplayOutcome, 1)
+	go func() {
+		result <- svc.replayCeilingDeferral(ctx, &models.Task{ID: "replay-task"}, models.CeilingDeferral{
+			Kind: models.CeilingLaunchPromptEnsure,
+			Payload: map[string]interface{}{
+				metaKeySessionID: "replay-session", metaKeyPrompt: "resume the accepted work",
+			},
+		})
+	}()
+	t.Cleanup(func() {
+		releaseProvider()
+		select {
+		case <-result:
+		case <-time.After(5 * time.Second):
+			t.Error("replay did not finish during cleanup")
+		}
+	})
+	select {
+	case held := <-providerEntered:
+		if held {
+			t.Error("provider dispatch inherited ownership of the replay admission lock")
+		}
+	case outcome := <-result:
+		result <- outcome
+		t.Fatalf("replay ended before provider dispatch: %v", outcome)
+	case <-time.After(5 * time.Second):
+		t.Fatal("replay did not reach provider dispatch")
+	}
+	admitted := make(chan struct{})
+	go func() {
+		release := svc.acquireCeilingEntryAdmissionLock("replay-task")
+		release()
+		close(admitted)
+	}()
+	t.Cleanup(func() { releaseProvider(); <-admitted })
+	select {
+	case <-admitted:
+	case <-time.After(time.Second):
+		t.Error("provider dispatch blocks task admission needed by lifecycle callbacks")
+	}
+	releaseProvider()
+	outcome := <-result
+	result <- outcome
+	require.Equal(t, ceilingReplaySucceeded, outcome)
+}
+
+// @covers AC-AGENTS-SESSION-CEILING-001.2
+func TestReviewAdmissionWaitDoesNotBlockIndependentScheduling(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedTaskAndSession(t, repo, "review-task", "review-session", models.TaskSessionStateWaitingForInput)
+	seedTaskAndSession(t, repo, "new-task", "new-session", models.TaskSessionStateCreated)
+	taskRepo := newMockTaskRepo()
+	seedMockTaskState(taskRepo, "review-task", v1.TaskStateInProgress)
+	seedMockTaskState(taskRepo, "new-task", v1.TaskStateInProgress)
+	svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, &mockAgentManager{})
+	release := svc.acquireCeilingEntryAdmissionLock("review-task")
+	var releaseOnce sync.Once
+	releaseAdmission := func() { releaseOnce.Do(release) }
+	reviewDone := make(chan struct{})
+	go func() {
+		svc.writeTaskReviewState(ctx, "review-task", "review-session")
+		close(reviewDone)
+	}()
+	t.Cleanup(func() { releaseAdmission(); <-reviewDone })
+	// Observe the waiter at the exact admission boundary before scheduling.
+	require.Eventually(t, func() bool {
+		svc.ceilingEntryAdmissionLocksMu.Lock()
+		defer svc.ceilingEntryAdmissionLocksMu.Unlock()
+		return svc.ceilingEntryAdmissionLocks["review-task"].refs == 2
+	}, 5*time.Second, time.Millisecond)
+	scheduled := make(chan error, 1)
+	go func() { scheduled <- svc.scheduleTaskForSession(ctx, "new-task", "new-session") }()
+	t.Cleanup(func() { releaseAdmission(); <-scheduled })
+	select {
+	case err := <-scheduled:
+		scheduled <- err
+		require.NoError(t, err)
+		taskRepo.mu.Lock()
+		state := taskRepo.tasks["new-task"].State
+		taskRepo.mu.Unlock()
+		require.Equal(t, v1.TaskStateScheduling, state)
+	case <-time.After(time.Second):
+		t.Error("a task-local admission wait blocked independent scheduling")
+	}
+}
+
+// @covers AC-AGENTS-SESSION-CEILING-001.5
+func TestCeilingReplayCommitsRouteOwnershipBeforeProviderDispatch(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedTaskAndSessionWithStep(t, repo, "commit-task", "commit-session", "step-old")
+	if err := repo.UpdateTaskSessionState(ctx, "commit-session", models.TaskSessionStateWaitingForInput, ""); err != nil {
+		t.Fatalf("update session state: %v", err)
+	}
+	seedExecutorRunning(t, repo, "commit-session", "commit-task", "commit-execution")
+	task, err := repo.GetTask(ctx, "commit-task")
+	require.NoError(t, err)
+	entryIdentity := (&Service{repo: repo}).workflowEntryIdentity(ctx, task.ID)
+	oldRoute := models.WorkflowSessionRoute{
+		OperationID:       "route-old",
+		DestinationStepID: "step-old",
+		EntryIdentity:     entryIdentity,
+		TargetKind:        "new_session",
+		DestinationID:     "commit-session",
+		Phase:             "committed",
+	}
+	newRoute := oldRoute
+	newRoute.OperationID = "route-new"
+	newRoute.DestinationStepID = "step-new"
+	newRoute.EntryIdentity = "entry:new"
+	require.NoError(t, repo.SetTaskMetadataKey(ctx, task.ID, models.MetaKeyWorkflowSessionRoute, oldRoute))
+
+	deferral := models.CeilingDeferral{
+		Kind: models.CeilingLaunchPromptEnsure,
+		Payload: map[string]interface{}{
+			metaKeySessionID: "commit-session",
+			metaKeyPrompt:    "old workflow prompt",
+			models.CeilingLaunchEntryBindingKey: map[string]interface{}{
+				"workflow_id":            "wf1",
+				"destination_step_id":    "step-old",
+				"route_operation_id":     "route-old",
+				"entry_identity":         entryIdentity,
+				"destination_session_id": "commit-session",
+			},
+		},
+		Origin:   string(launchOriginAutomatic),
+		QueuedAt: time.Now().UTC(),
+	}
+
+	stepGetter := newMockStepGetter()
+	stepGetter.steps["step-old"] = &wfmodels.WorkflowStep{ID: "step-old", WorkflowID: "wf1"}
+	taskRepo := newMockTaskRepo()
+	seedMockTaskState(taskRepo, task.ID, v1.TaskStateScheduling)
+	baseAgent := &mockAgentManager{repoForExecutionLookup: repo, isAgentRunning: true}
+	agent := &gatedPromptAgentManager{
+		mockAgentManager: baseAgent,
+		beforePrompt:     make(chan struct{}),
+		allowPrompt:      make(chan struct{}),
+	}
+	svc := createTestServiceWithScheduler(repo, stepGetter, taskRepo, agent)
+	result := make(chan ceilingReplayOutcome, 1)
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		result <- svc.replayCeilingDeferral(ctx, &models.Task{ID: task.ID}, deferral)
+	}()
+	t.Cleanup(func() {
+		agent.releasePrompt()
+		select {
+		case <-done:
+		case <-time.After(5 * time.Second):
+			t.Error("replay did not finish during cleanup")
+		}
+	})
+	select {
+	case <-agent.beforePrompt:
+	case outcome := <-result:
+		t.Fatalf("replay ended before provider dispatch: %v", outcome)
+	case <-time.After(5 * time.Second):
+		t.Fatal("replay did not reach the provider dispatch boundary")
+	}
+
+	routeErr := svc.persistWorkflowSessionRoute(context.Background(), task.ID, newRoute)
+	require.Error(t, routeErr)
+	require.True(t, errors.Is(routeErr, ErrCeilingEntryDispatchCommitted), routeErr)
+	agent.releasePrompt()
+	require.Equal(t, ceilingReplaySucceeded, <-result)
+
+	reloaded, err := repo.GetTask(ctx, task.ID)
+	require.NoError(t, err)
+	route, ok := models.LoadWorkflowSessionRoute(reloaded.Metadata)
+	require.True(t, ok)
+	require.Equal(t, oldRoute.OperationID, route.OperationID)
+}

apps/backend/internal/orchestrator/ceiling_entry_lock.go

@@ -1,6 +1,12 @@
 package orchestrator
 
-import "context"
+import (
+	"context"
+	"errors"
+	"sync"
+
+	"github.com/kandev/kandev/internal/task/models"
+)
 
 type ceilingEntryAdmissionLock struct {
 	mu   chan struct{}
@@ -8,6 +14,19 @@ type ceilingEntryAdmissionLock struct {
 }
 
 type ceilingEntryAdmissionContextKey struct{}
+type ceilingEntryDispatchCommitContextKey struct{}
+
+// ErrCeilingEntryDispatchCommitted reports that a workflow route is already
+// owned by an admitted provider dispatch. A competing transition must leave
+// that route unchanged and retry after the dispatch settles.
+var ErrCeilingEntryDispatchCommitted = errors.New("workflow route dispatch is already committed")
+
+// ceilingEntryDispatchCommit is a process-local ownership fence. The durable
+// deferred-launch claim survives a restart, while this marker closes the
+// smaller gap between the final route read and provider I/O in this process.
+type ceilingEntryDispatchCommit struct {
+	binding models.CeilingWorkflowEntryBinding
+}
 
 // acquireCeilingEntryAdmissionLock serializes operations that read or write a
 // task's committed workflow route and its ceiling deferred-launch record. A
@@ -61,3 +80,87 @@ func (s *Service) lockCeilingEntryAdmission(ctx context.Context, taskID string)
 	release := s.acquireCeilingEntryAdmissionLock(taskID)
 	return context.WithValue(ctx, ceilingEntryAdmissionContextKey{}, taskID), release
 }
+
+func ceilingEntryDispatchCommitFromContext(ctx context.Context) *ceilingEntryDispatchCommit {
+	if ctx == nil {
+		return nil
+	}
+	commit, _ := ctx.Value(ceilingEntryDispatchCommitContextKey{}).(*ceilingEntryDispatchCommit)
+	return commit
+}
+
+// commitCeilingEntryDispatch validates and records immutable workflow-entry
+// ownership while the task admission lock is held. Provider and readiness I/O
+// run after the lock is released, carrying the process-local commit marker in
+// context so route writers can reject a successor during that I/O.
+func (s *Service) commitCeilingEntryDispatch(
+	ctx context.Context,
+	taskID string,
+	binding *models.CeilingWorkflowEntryBinding,
+) (context.Context, func(), error) {
+	if s == nil || taskID == "" || binding == nil {
+		return ctx, func() {}, nil
+	}
+	if !binding.Valid() {
+		return ctx, func() {}, errors.New("workflow entry dispatch binding is incomplete")
+	}
+	if existing := ceilingEntryDispatchCommitFromContext(ctx); existing != nil {
+		return ctx, func() {}, nil
+	}
+
+	admissionCtx, releaseAdmission := s.lockCeilingEntryAdmission(ctx, taskID)
+	validationCtx := withCeilingEntryBinding(admissionCtx, binding)
+	if err := s.validateClaimedCeilingBinding(validationCtx, taskID, binding); err != nil {
+		releaseAdmission()
+		return ctx, func() {}, err
+	}
+
+	commit := &ceilingEntryDispatchCommit{binding: *binding}
+	s.ceilingEntryDispatchCommitsMu.Lock()
+	if s.ceilingEntryDispatchCommits == nil {
+		s.ceilingEntryDispatchCommits = make(map[string]*ceilingEntryDispatchCommit)
+	}
+	if existing := s.ceilingEntryDispatchCommits[taskID]; existing != nil {
+		s.ceilingEntryDispatchCommitsMu.Unlock()
+		releaseAdmission()
+		return ctx, func() {}, ErrCeilingEntryDispatchCommitted
+	}
+	s.ceilingEntryDispatchCommits[taskID] = commit
+	s.ceilingEntryDispatchCommitsMu.Unlock()
+	// The marker, not the admission lock, spans provider I/O. Releasing here
+	// lets lifecycle and route readers acquire the task lock without allowing a
+	// successor route to replace this committed binding.
+	releaseAdmission()
+
+	var releaseOnce sync.Once
+	release := func() {
+		releaseOnce.Do(func() {
+			s.ceilingEntryDispatchCommitsMu.Lock()
+			if s.ceilingEntryDispatchCommits[taskID] == commit {
+				delete(s.ceilingEntryDispatchCommits, taskID)
+			}
+			s.ceilingEntryDispatchCommitsMu.Unlock()
+		})
+	}
+	// Do not return validationCtx itself: it carries the admission-lock marker,
+	// but that lock was released before provider I/O. Rebuild the dispatch
+	// context from the caller's context so nested route writers cannot mistake
+	// the released lock for a held one.
+	dispatchCtx := withCeilingEntryBinding(ctx, binding)
+	return context.WithValue(dispatchCtx, ceilingEntryDispatchCommitContextKey{}, commit), release, nil
+}
+
+// workflowRouteMutationAllowed lets the route owner proceed while rejecting
+// unrelated transitions after an admitted dispatch has committed its binding.
+func (s *Service) workflowRouteMutationAllowed(ctx context.Context, taskID string) error {
+	if s == nil || taskID == "" {
+		return nil
+	}
+	s.ceilingEntryDispatchCommitsMu.Lock()
+	commit := s.ceilingEntryDispatchCommits[taskID]
+	s.ceilingEntryDispatchCommitsMu.Unlock()
+	if commit == nil || ceilingEntryDispatchCommitFromContext(ctx) == commit {
+		return nil
+	}
+	return ErrCeilingEntryDispatchCommitted
+}

apps/backend/internal/orchestrator/ceiling_replay.go

@@ -404,6 +404,26 @@ func stripCeilingRecordKeys(existing interface{}) map[string]interface{} {
 	return stripped
 }
 
+// loadCeilingReplayTask releases admission before session guards or provider
+// callbacks can run. Only the immutable workflow binding crosses into dispatch;
+// the durable claim owns replay until settlement.
+func (s *Service) loadCeilingReplayTask(ctx context.Context, taskID string, deferral models.CeilingDeferral) (*models.Task, ceilingReplayOutcome) {
+	ctx, release := s.lockCeilingEntryAdmission(ctx, taskID)
+	defer release()
+	current, err := s.repo.GetTask(ctx, taskID)
+	if err != nil || current == nil {
+		return nil, ceilingReplayFailed
+	}
+	disposition, _, validationErr := s.validateCeilingEntry(ctx, current, deferral)
+	if validationErr != nil || disposition == ceilingEntryUnavailable {
+		return nil, ceilingReplayFailed
+	}
+	if disposition == ceilingEntrySuperseded {
+		return nil, ceilingReplaySuperseded
+	}
+	return current, ceilingReplaySucceeded
+}
+
 // replayCeilingDeferral dispatches to the one replay function matching the
 // deferral's kind. The closed set is exhaustive; a kind outside it was
 // already rejected as unreplayable by ReadCeilingDeferral before this point.
@@ -411,21 +431,9 @@ func (s *Service) replayCeilingDeferral(ctx context.Context, task *models.Task,
 	if task == nil || task.ID == "" {
 		return ceilingReplayFailed
 	}
-	ctx, release := s.lockCeilingEntryAdmission(ctx, task.ID)
-	defer release()
-	current, err := s.repo.GetTask(ctx, task.ID)
-	if err != nil || current == nil {
-		return ceilingReplayFailed
-	}
-	disposition, _, validationErr := s.validateCeilingEntry(ctx, current, deferral)
-	if validationErr != nil {
-		return ceilingReplayFailed
-	}
-	if disposition == ceilingEntrySuperseded {
-		return ceilingReplaySuperseded
-	}
-	if disposition == ceilingEntryUnavailable {
-		return ceilingReplayFailed
+	current, outcome := s.loadCeilingReplayTask(ctx, task.ID, deferral)
+	if current == nil {
+		return outcome
 	}
 	replayCtx := ctx
 	if binding, present, bindingErr := models.ReadCeilingWorkflowEntryBinding(deferral.Payload); bindingErr == nil && present {

apps/backend/internal/orchestrator/dynamic_launch.go

@@ -53,7 +53,18 @@ func (d *dynamicTaskDownstream) Launch(
 	options.Prompt = launch.Prompt
 	options.PriorACPSession = launch.PriorACPSession
 	d.service.beginDynamicAttempt(d.sessionID)
-	execution, err := d.service.executor.LaunchPreparedSession(ctx, d.task, d.sessionID, options)
+	taskID := ""
+	if d.task != nil {
+		taskID = d.task.ID
+	}
+	dispatchCtx, releaseDispatchCommit, err := d.service.commitCeilingEntryDispatch(
+		ctx, taskID, ceilingEntryBindingFromContext(ctx),
+	)
+	if err != nil {
+		return dynamicruntime.DownstreamExecution{}, err
+	}
+	defer releaseDispatchCommit()
+	execution, err := d.service.executor.LaunchPreparedSession(dispatchCtx, d.task, d.sessionID, options)
 	if err != nil {
 		var classified *routingerr.Error
 		if errors.As(err, &classified) {
@@ -72,7 +83,7 @@ func (d *dynamicTaskDownstream) Launch(
 		return dynamicruntime.DownstreamExecution{}, fmt.Errorf("%w: %v", classified, err)
 	}
 	d.service.bindDynamicAttemptExecution(d.sessionID, execution.AgentExecutionID)
-	d.service.bindPromptAttemptToExecution(ctx, d.sessionID, execution.AgentExecutionID)
+	d.service.bindPromptAttemptToExecution(dispatchCtx, d.sessionID, execution.AgentExecutionID)
 	d.execution = execution
 	acpSessionID := ""
 	if session, sessionErr := d.service.repo.GetTaskSession(ctx, d.sessionID); sessionErr == nil && session != nil {
@@ -471,12 +482,23 @@ func (s *Service) launchConcretePreparedSession(
 			return nil, err
 		}
 	}
+	taskID := ""
+	if task != nil {
+		taskID = task.ID
+	}
+	dispatchCtx, releaseDispatchCommit, err := s.commitCeilingEntryDispatch(
+		ctx, taskID, ceilingEntryBindingFromContext(ctx),
+	)
+	if err != nil {
+		return nil, err
+	}
+	defer releaseDispatchCommit()
 	if options.StartAgent && (options.Prompt != "" || len(options.Attachments) > 0) {
 		s.beginInitialPromptAttempt(sessionID, false)
 	}
-	execution, err := s.executor.LaunchPreparedSession(ctx, task, sessionID, options)
+	execution, err := s.executor.LaunchPreparedSession(dispatchCtx, task, sessionID, options)
 	if execution != nil {
-		s.bindPromptAttemptToExecution(ctx, sessionID, execution.AgentExecutionID)
+		s.bindPromptAttemptToExecution(dispatchCtx, sessionID, execution.AgentExecutionID)
 	}
 	return execution, err
 }

apps/backend/internal/orchestrator/event_handlers_streaming.go

@@ -2443,10 +2443,10 @@ func (s *Service) writeTaskReviewState(ctx context.Context, taskID, completedSes
 		return
 	}
 
-	s.taskRuntimeStateMu.Lock()
-	defer s.taskRuntimeStateMu.Unlock()
 	ctx, releaseCeilingEntry := s.lockCeilingEntryAdmission(ctx, taskID)
 	defer releaseCeilingEntry()
+	s.taskRuntimeStateMu.Lock()
+	defer s.taskRuntimeStateMu.Unlock()
 
 	if completedSessionID != "" {
 		if session, err := s.repo.GetTaskSession(ctx, completedSessionID); err == nil && session != nil && isWorkingSessionState(session.State) {

apps/backend/internal/orchestrator/event_handlers_workflow.go

@@ -3546,6 +3546,11 @@ func (s *Service) prepareWorkflowReplacementSession(
 	newAgentProfileID string,
 	workflowRoute *models.WorkflowSessionRoute,
 ) (*models.TaskSession, error) {
+	if workflowRoute != nil {
+		if err := s.workflowRouteMutationAllowed(ctx, taskID); err != nil {
+			return nil, err
+		}
+	}
 	task, err := s.scheduler.GetTask(ctx, taskID)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get task for session switch: %w", err)

apps/backend/internal/orchestrator/service.go

@@ -708,6 +708,12 @@ type Service struct {
 	// parallel while an old entry cannot race a successor route.
 	ceilingEntryAdmissionLocksMu sync.Mutex
 	ceilingEntryAdmissionLocks   map[string]*ceilingEntryAdmissionLock
+	// ceilingEntryDispatchCommits retain immutable workflow-entry ownership
+	// after the admission lock is released and while provider I/O is in flight.
+	// Route writers consult this map so a successor cannot replace the route in
+	// the validation-to-dispatch window.
+	ceilingEntryDispatchCommitsMu sync.Mutex
+	ceilingEntryDispatchCommits   map[string]*ceilingEntryDispatchCommit
 
 	// Message creator for saving agent responses
 	messageCreator MessageCreator

apps/backend/internal/orchestrator/task_operations.go

@@ -635,6 +635,9 @@ func (s *Service) startCreatedSession(
 	if options.ceilingEntryBinding == nil {
 		options.ceilingEntryBinding = ceilingEntryBindingFromContext(ctx)
 	}
+	if options.ceilingEntryBinding != nil {
+		ctx = withCeilingEntryBinding(ctx, options.ceilingEntryBinding)
+	}
 
 	// One GetWorkflowMeta read shared by profile resolution and prompt build.
 	ctx = withWorkflowMetaCache(ctx)
@@ -1304,6 +1307,9 @@ func (s *Service) promoteSelectedExplicitWorkflowSession(
 //nolint:cyclop,funlen,gocognit,nestif // launch path threads many orthogonal concerns (workflow-step / agent-profile / office-task / config-mode / route / system-prompt wrapping); splitting it would require shared mutable state across helpers
 func (s *Service) startTask(ctx context.Context, taskID string, agentProfileID string, executorID string, executorProfileID string, priority string, prompt string, workflowStepID string, planMode, autoStart bool, attachments []v1.MessageAttachment, opts startTaskOptions) (*executor.TaskExecution, error) {
 	ctx = executor.WithTaskRunnerProfileExplicit(ctx, strings.TrimSpace(executorProfileID) != "")
+	if opts.ceilingEntryBinding != nil {
+		ctx = withCeilingEntryBinding(ctx, opts.ceilingEntryBinding)
+	}
 	// One GetWorkflowMeta read shared by profile resolution and prompt build.
 	ctx = withWorkflowMetaCache(ctx)
 	// Fail before task-state or session mutations when the selected logical
@@ -2162,6 +2168,11 @@ func (s *Service) createStartSessionWithWorkflowRoute(
 	agentProfileID, officeAgentProfileID, executorID, executorProfileID, workflowStepID string,
 	workflowRoute *models.WorkflowSessionRoute,
 ) (string, bool, error) {
+	if workflowRoute != nil {
+		if err := s.workflowRouteMutationAllowed(ctx, task.ID); err != nil {
+			return "", false, err
+		}
+	}
 	dbTask, err := s.repo.GetTask(ctx, task.ID)
 	if err != nil || dbTask == nil || !dbTask.IsFromOffice {
 		var sessionID string
@@ -3017,7 +3028,15 @@ func (s *Service) resumeTaskSessionWithContinuation(
 		s.cleanupCancelledResumeAttempt(attempt)
 		return nil, err
 	}
-	execution, err := s.executor.ResumeSessionWithOptions(resumeCtx, session, true, options)
+	dispatchCtx, releaseCeilingDispatch, err := s.commitCeilingEntryDispatch(
+		resumeCtx, taskID, entryBinding,
+	)
+	if err != nil {
+		s.cleanupCancelledResumeAttempt(attempt)
+		return nil, err
+	}
+	execution, err := s.executor.ResumeSessionWithOptions(dispatchCtx, session, true, options)
+	releaseCeilingDispatch()
 	if execution != nil {
 		attempt.setExecutionID(execution.AgentExecutionID)
 	}
@@ -3181,7 +3200,14 @@ func (s *Service) recoverAlreadyRunningResume(
 		return nil, nil, fmt.Errorf("task session does not belong to task")
 	}
 
-	execution, err := s.executor.ResumeSessionWithOptions(resumeCtx, session, true, options)
+	dispatchCtx, releaseCeilingDispatch, err := s.commitCeilingEntryDispatch(
+		resumeCtx, taskID, ceilingEntryBindingFromContext(resumeCtx),
+	)
+	if err != nil {
+		return nil, nil, err
+	}
+	execution, err := s.executor.ResumeSessionWithOptions(dispatchCtx, session, true, options)
+	releaseCeilingDispatch()
 	if err != nil {
 		return nil, nil, err
 	}
@@ -3721,7 +3747,14 @@ func (s *Service) attemptColdResume(
 	if err := s.validateContextCeilingEntry(resumeCtx, session.TaskID); err != nil {
 		return false, err
 	}
-	execution, launchErr := s.executor.ResumeSession(resumeCtx, session, true)
+	dispatchCtx, releaseCeilingDispatch, err := s.commitCeilingEntryDispatch(
+		resumeCtx, session.TaskID, ceilingEntryBindingFromContext(resumeCtx),
+	)
+	if err != nil {
+		return false, err
+	}
+	execution, launchErr := s.executor.ResumeSession(dispatchCtx, session, true)
+	releaseCeilingDispatch()
 	if execution != nil && resumeAttempt != nil {
 		resumeAttempt.setExecutionID(execution.AgentExecutionID)
 	}
@@ -5862,6 +5895,19 @@ func (s *Service) runPromptTurn(
 	if err != nil || earlyResult != nil {
 		return earlyResult, err
 	}
+	dispatchCtx, releaseCeilingDispatch, err := s.commitCeilingEntryDispatch(
+		promptCtx, taskID, ceilingEntryBindingFromContext(promptCtx),
+	)
+	if err != nil {
+		if releaseDispatchGuard != nil {
+			releaseDispatchGuard()
+		}
+		return nil, s.failPromptDispatch(
+			ctx, taskID, sessionID, foregroundDispatch, rollback, options, resumeAttempt, err,
+		)
+	}
+	defer releaseCeilingDispatch()
+	promptCtx = dispatchCtx
 	onDispatched, dispatchOutcome := s.preparePromptDispatchCallback(
 		promptCtx, taskID, sessionID, session, rollback, options, foregroundDispatch, releaseDispatchGuard,
 	)

apps/backend/internal/orchestrator/workflow_session_target.go

@@ -167,6 +167,9 @@ func (s *Service) resolveBoundSourceWorkflowSession(
 func (s *Service) persistWorkflowSessionRoute(ctx context.Context, taskID string, route models.WorkflowSessionRoute) error {
 	ctx, release := s.lockCeilingEntryAdmission(ctx, taskID)
 	defer release()
+	if err := s.workflowRouteMutationAllowed(ctx, taskID); err != nil {
+		return err
+	}
 	setter, ok := s.repo.(taskMetadataKeySetter)
 	if !ok {
 		return nil
@@ -237,6 +240,9 @@ func (s *Service) promoteWorkflowSessionRoute(
 ) (bool, error) {
 	ctx, release := s.lockCeilingEntryAdmission(ctx, taskID)
 	defer release()
+	if err := s.workflowRouteMutationAllowed(ctx, taskID); err != nil {
+		return false, err
+	}
 	if destination == nil {
 		return false, nil
 	}

docs/plans/ceiling-boot-deadlock/plan.md

@@ -0,0 +1,80 @@
+---
+created: 2026-09-17
+status: done
+requirements:
+  - REQ-AGENTS-SESSION-CEILING-001
+system_design:
+  - ../../specs/agents/system-design/session-concurrency-ceiling.md
+legacy_specs: []
+---
+
+# Implementation Plan: Ceiling replay boot deadlock
+
+## Overview
+
+Restore deferred replay and independent task admission after an agent boots.
+Two live goroutine snapshots on September 17 confirmed a circular wait.
+Replay held task admission and waited for the session guard.
+The boot callback held the session guard and waited for task admission.
+It also held the global task-state mutex, which blocked unrelated launches.
+
+Existing replay and manual-admission criteria define the intended behavior.
+The user requested this package, implementation, and PR in the same turn.
+
+## Scope
+
+### In scope
+
+- Limit replay admission locking to validation.
+- Preserve durable claims and final workflow-binding validation.
+- Acquire task admission before the global task-state mutex.
+- Cover replay progress and independent scheduling with regression tests.
+
+### Out of scope
+
+- Live-instance restart, database changes, backup performance, and plugin warnings.
+- Changes to ceiling limits, manual overrides, or rendered UI.
+
+## Technical approach
+
+`ceiling_replay.go` releases its validation lock before concrete replay dispatch.
+The caller context must not retain the released lock's ownership marker.
+Durable claims still exclude concurrent replay and Send Now.
+Existing launch and prompt boundaries still validate the immutable workflow binding.
+
+`writeTaskReviewState` in `event_handlers_streaming.go` acquires task admission
+before `taskRuntimeStateMu`. A task-local wait cannot hold the global mutex.
+The agent system design records these lock lifetimes.
+
+## Tests
+
+`ceiling_boot_deadlock_test.go` covers:
+
+- `TestCeilingReplayReleasesAdmissionBeforeProviderDispatch`: AC-AGENTS-SESSION-CEILING-001.5.
+- `TestReviewAdmissionWaitDoesNotBlockIndependentScheduling`: AC-AGENTS-SESSION-CEILING-001.2.
+
+Existing ceiling replay tests cover stale workflow entries and durable claims.
+Existing boot-ready tests cover session settlement and queue draining.
+
+## E2E tests
+
+Service tests exercise scheduling and replay through the real repository and executor.
+Only the provider boundary is simulated. No rendered UI changes require browser coverage.
+
+## Work orders
+
+- [x] [Task 01: Break the replay lock cycle](task-01-break-lock-cycle.md)
+
+## Verification results
+
+Both new regression tests failed on the original code for the expected lock waits.
+Both passed after the correction (0.156 seconds).
+The work-order race command passed (4.449 seconds).
+The documentation catalog validated 288 decisions and 1002 specifications.
+The specification linter and git diff whitespace check passed.
+The package status check found both new documents before staging.
+
+## Risks
+
+Releasing admission before dispatch must preserve final workflow-binding checks
+and must not leak lock ownership into callbacks.

docs/plans/ceiling-boot-deadlock/task-01-break-lock-cycle.md

@@ -0,0 +1,92 @@
+---
+id: "01-break-lock-cycle"
+title: "Break the replay lock cycle"
+status: done
+wave: 1
+depends_on: []
+plan: "plan.md"
+requirements:
+  - REQ-AGENTS-SESSION-CEILING-001
+acceptance_criteria:
+  - AC-AGENTS-SESSION-CEILING-001.2
+  - AC-AGENTS-SESSION-CEILING-001.5
+system_design:
+  - ../../specs/agents/system-design/session-concurrency-ceiling.md
+---
+
+# Task 01: Break the replay lock cycle
+
+## Summary
+
+Release replay admission before provider work and session-guard waits.
+Keep unrelated scheduling available during task-local admission waits.
+
+## In scope
+
+- Add deterministic boundary tests before production changes.
+- Narrow replay lock scope and correct review-state lock order.
+- Preserve existing stale-entry rejection and claim settlement.
+
+## Out of scope
+
+Live data, process restart, UI changes, and ceiling configuration.
+
+## Acceptance
+
+- Provider dispatch permits an independent task-admission operation.
+- A review callback waiting for task admission does not block another task's scheduling.
+- Existing replay identity, claim, and boot-ready tests pass with the race detector.
+
+## Verification
+
+From the repository root:
+
+This session uses a writable cache because the host Go cache is read-only.
+
+```bash
+export GOCACHE=/root/kandev/.worktrees/go-cache
+(cd apps/backend && go test -race ./internal/orchestrator -run 'Test(Ceiling|ReplayCeiling|ClaimCeiling|ClearCeiling|WriteTaskReview|ReviewAdmission|HandleAgentBootReady|AgentBootReady|WorkflowEntryDispatch)' -count=1 -timeout=5m)
+python3 scripts/list-docs.py validate
+python3 scripts/lint-spec-files.py --all
+git diff --check
+```
+
+## Files likely touched
+
+- apps/backend/internal/orchestrator/ceiling_replay.go
+- apps/backend/internal/orchestrator/event_handlers_streaming.go
+- apps/backend/internal/orchestrator/ceiling_boot_deadlock_test.go
+- docs/specs/agents/system-design/session-concurrency-ceiling.md
+
+## Dependencies
+
+None.
+
+## Risks
+
+The released validation context must not represent continued lock ownership.
+Final workflow-binding validation and durable claims must remain intact.
+
+## Parallelism
+
+`sequential`
+
+## Inputs
+
+- REQ-AGENTS-SESSION-CEILING-001 and its replay and manual-admission criteria.
+- The linked session ceiling design.
+- Live goroutine evidence summarized in the plan.
+- Existing ceiling replay and boot-ready test fixtures.
+
+## Results
+
+- RED: Both new tests failed on the original code. Replay retained admission
+  during provider dispatch, and review reconciliation blocked independent scheduling.
+- GREEN: Both new tests passed after the correction (0.156 seconds).
+- The race command above passed (4.449 seconds).
+- `python3 scripts/list-docs.py validate` passed: 288 decisions and 1002 specifications.
+- `python3 scripts/lint-spec-files.py --all` passed.
+- `git diff --check` passed.
+- The package status check found the plan and work order before staging.
+- Public docs retain the existing automatic-retry contract. No public API, configuration,
+  or rendered UI changes require documentation or screenshots.

docs/specs/agents/system-design/session-concurrency-ceiling.md

@@ -63,6 +63,16 @@ origins never become manual during replay. The sweep clears a record only after
 successful dispatch. A repeated refusal leaves the original timestamp and
 payload in place.
 
+The task admission lock protects bounded record and route operations. Replay
+releases this lock after validation, before session-guard waits and provider
+dispatch. The dispatch context retains the workflow binding, but not ownership
+of the released lock. The durable claim excludes competing replay consumers.
+Launch and prompt boundaries revalidate the binding before provider work.
+
+Task-state reconciliation acquires task admission before the global task-state
+mutex. A task-local admission wait cannot hold the global mutex. Provider
+callbacks can complete while replay waits for agent readiness.
+
 The task-owned [queued session ownership design](../../tasks/system-design/queued-session-ownership.md)
 plans explicit passive-inspection classification at the caller boundary. Opening
 a conversation is not a manual override. Actual explicit execution retains this