fix: prevent ceiling replay lock cycles (#3779)

apps/backend/internal/orchestrator/ceiling_claim.go

@@ -148,49 +148,67 @@ func (c *ceilingDeferredLaunchClaim) settle(ctx context.Context) {
 // mutateCeilingClaim performs the compare-and-set settlement for one claim.
 // A changed deferral or claim is never modified by this older owner.
 func (s *Service) mutateCeilingClaim(ctx context.Context, claim *ceilingDeferredLaunchClaim, settle bool) error {
-	for attempt := 0; attempt < deferredLaunchCASRetryBudget; attempt++ {
-		record, prior, err := s.repo.GetTaskDeferredLaunch(ctx, claim.taskID)
-		if err != nil {
-			return err
-		}
-		current, err := models.ReadCeilingDeferral(record)
-		if err != nil {
-			return nil
-		}
-		equivalent, err := sameCeilingDeferralIdentity(current, claim.deferral)
-		if err != nil {
-			return err
-		}
-		if !equivalent {
-			// Capacity observations are mutable bookkeeping. The claim id below
-			// still binds the mutation to this exact in-flight owner, while the
-			// stable deferral identity prevents a successor from being touched.
-			return nil
-		}
-		claimID, _, ok := models.ReadCeilingLaunchClaim(record)
-		if !ok || claimID != claim.id {
-			return nil
-		}
-
-		updated := cloneCeilingRecord(record)
-		if settle {
-			updated = stripCeilingRecordKeys(updated)
-		} else {
-			delete(updated, models.CeilingLaunchClaimKey)
-		}
-		stored, lostCompare, err := s.repo.SetTaskDeferredLaunchIfUnchanged(ctx, claim.taskID, prior, updated)
-		if err != nil {
-			return err
-		}
-		if stored {
-			if settle {
-				s.publishTaskUpdatedByID(ctx, claim.taskID)
+	// Claim mutation is another task-admission operation. Keep the lock around
+	// the compare-and-set only; settling publishes after release because task
+	// event subscribers may re-enter runtime reconciliation.
+	admissionCtx, release := s.lockCeilingEntryAdmission(ctx, claim.taskID)
+	var publish bool
+	var result error
+	func() {
+		defer release()
+		for attempt := 0; attempt < deferredLaunchCASRetryBudget; attempt++ {
+			record, prior, err := s.repo.GetTaskDeferredLaunch(admissionCtx, claim.taskID)
+			if err != nil {
+				result = err
+				return
+			}
+			current, err := models.ReadCeilingDeferral(record)
+			if err != nil {
+				return
+			}
+			equivalent, err := sameCeilingDeferralIdentity(current, claim.deferral)
+			if err != nil {
+				result = err
+				return
+			}
+			if !equivalent {
+				// Capacity observations are mutable bookkeeping. The claim id below
+				// still binds the mutation to this exact in-flight owner, while the
+				// stable deferral identity prevents a successor from being touched.
+				return
+			}
+			claimID, _, ok := models.ReadCeilingLaunchClaim(record)
+			if !ok || claimID != claim.id {
+				return
+			}
+
+			updated := cloneCeilingRecord(record)
+			if settle {
+				updated = stripCeilingRecordKeys(updated)
+			} else {
+				delete(updated, models.CeilingLaunchClaimKey)
+			}
+			stored, lostCompare, err := s.repo.SetTaskDeferredLaunchIfUnchanged(admissionCtx, claim.taskID, prior, updated)
+			if err != nil {
+				result = err
+				return
+			}
+			if stored {
+				publish = settle
+				return
+			}
+			if !lostCompare {
+				result = fmt.Errorf("repository reported no claim settlement")
+				return
 			}
-			return nil
-		}
-		if !lostCompare {
-			return fmt.Errorf("repository reported no claim settlement")
 		}
+		result = fmt.Errorf("claim settlement compare-and-set retries exhausted")
+	}()
+	if result != nil {
+		return result
 	}
-	return fmt.Errorf("claim settlement compare-and-set retries exhausted")
+	if publish {
+		s.publishTaskUpdatedByID(context.WithoutCancel(ctx), claim.taskID)
+	}
+	return nil
 }

apps/backend/internal/orchestrator/ceiling_defer.go

@@ -51,93 +51,124 @@ func (s *Service) deferCeilingRefusal(
 	ctx context.Context, taskID, sessionID string, kind models.CeilingLaunchKind, payload map[string]interface{},
 	reasonCode string, population int, populationKnown bool, ceiling int,
 ) error {
-	ctx, release := s.lockCeilingEntryAdmission(ctx, taskID)
-	defer release()
-	// Capture the route/entry identity that is already committed at the
-	// admission boundary. The payload remains the first launch description;
-	// this only adds the durable binding needed to reject stale replay.
-	payload = s.enrichCeilingLaunchPayload(ctx, taskID, sessionID, payload)
-	for attempt := 0; attempt < deferredLaunchCASRetryBudget; attempt++ {
-		existingRaw, prior, err := s.repo.GetTaskDeferredLaunch(ctx, taskID)
-		if err != nil {
-			return fmt.Errorf("reading deferred launch for task %s: %w", taskID, err)
-		}
-
-		deferral := models.CeilingDeferral{
-			Kind:            kind,
-			Payload:         payload,
-			Origin:          string(launchOriginAutomatic),
-			ReasonCode:      reasonCode,
-			QueuedAt:        time.Now(),
-			Population:      population,
-			PopulationKnown: populationKnown,
-			Ceiling:         ceiling,
-		}
-
-		if existingCeiling, readErr := models.ReadCeilingDeferral(existingRaw); readErr == nil {
-			// A ceiling_deferred record already exists for this task (AC-12d).
-			equivalent, cmpErr := ceilingDeferralsEquivalentForAdmission(existingCeiling, deferral)
-			if cmpErr != nil {
-				return fmt.Errorf("comparing deferred launch payloads for task %s: %w", taskID, cmpErr)
+	// The admission lock protects only the read/merge/CAS sequence. State
+	// repair, card surfacing, and task publication can invoke other runtime
+	// paths, so they run after the durable record is released.
+	admissionCtx, release := s.lockCeilingEntryAdmission(ctx, taskID)
+	var (
+		result    error
+		reconcile bool
+		surface   bool
+		publish   bool
+	)
+	func() {
+		defer release()
+		// Capture the route/entry identity that is already committed at the
+		// admission boundary. The payload remains the first launch description;
+		// this only adds the durable binding needed to reject stale replay.
+		payload = s.enrichCeilingLaunchPayload(admissionCtx, taskID, sessionID, payload)
+		for attempt := 0; attempt < deferredLaunchCASRetryBudget; attempt++ {
+			existingRaw, prior, err := s.repo.GetTaskDeferredLaunch(admissionCtx, taskID)
+			if err != nil {
+				result = fmt.Errorf("reading deferred launch for task %s: %w", taskID, err)
+				return
 			}
-			if !equivalent {
-				// A different launch: retain the one already stored and report
-				// the collision rather than losing either payload silently.
-				s.logger.Zap().Warn("ceiling refusal superseded by an earlier pending deferral for the same task",
-					zap.String("task_id", taskID),
-					zap.String("stored_kind", string(existingCeiling.Kind)),
-					zap.String("superseded_kind", string(kind)),
-					zap.String(ceilingFieldReasonCode, ceilingReasonSuperseded))
-				return fmt.Errorf("%w: task %s already holds %s", ErrCeilingLaunchConflict, taskID, existingCeiling.Kind)
-			}
-			if existingCeiling.ReasonCode == reasonCode {
-				// AC-12a/AC-49e: the same launch, refused for the same reason,
-				// already recorded. Leave it exactly as stored, including its
-				// queued_at (AC-32) and its card-surface bookkeeping.
-				s.reconcileQueuedTaskState(ctx, taskID)
-				return nil
-			}
-			// AC-49f: the same pending launch, but the reason changed (for
-			// example a transient population read failure resolved into an
-			// ordinary refusal). Update the stored reason and this refusal's
-			// population/ceiling reading, preserve queued_at, and reset the
-			// surface stamps so the new reason gets its own note (AC-49g1).
-			updated := existingCeiling
-			updated.ReasonCode = reasonCode
-			updated.Population = population
-			updated.PopulationKnown = populationKnown
-			updated.Ceiling = ceiling
-			if s.writeCeilingDeferralUpdate(ctx, taskID, existingRaw, prior, updated) {
-				s.reconcileQueuedTaskState(ctx, taskID)
-				s.attemptCeilingSurfaceWrite(ctx, taskID)
-				s.publishTaskUpdatedByID(ctx, taskID)
-			}
-			return nil
-		}
 
-		record, discarded, replaced := models.MergeCeilingRecord(existingRaw, deferral)
-		if replaced {
-			s.logger.Zap().Warn("deferred_launch held a non-object value; the ceiling refusal replaced it",
-				zap.String("task_id", taskID), zap.Any("discarded_value", discarded))
-		}
+			deferral := models.CeilingDeferral{
+				Kind:            kind,
+				Payload:         payload,
+				Origin:          string(launchOriginAutomatic),
+				ReasonCode:      reasonCode,
+				QueuedAt:        time.Now(),
+				Population:      population,
+				PopulationKnown: populationKnown,
+				Ceiling:         ceiling,
+			}
 
-		stored, lostCompare, err := s.repo.SetTaskDeferredLaunchIfUnchanged(ctx, taskID, prior, record)
-		if err != nil {
-			return fmt.Errorf("writing deferred launch for task %s: %w", taskID, err)
+			if existingCeiling, readErr := models.ReadCeilingDeferral(existingRaw); readErr == nil {
+				// A ceiling_deferred record already exists for this task (AC-12d).
+				equivalent, cmpErr := ceilingDeferralsEquivalentForAdmission(existingCeiling, deferral)
+				if cmpErr != nil {
+					result = fmt.Errorf("comparing deferred launch payloads for task %s: %w", taskID, cmpErr)
+					return
+				}
+				if !equivalent {
+					// A different launch: retain the one already stored and report
+					// the collision rather than losing either payload silently.
+					s.logger.Zap().Warn("ceiling refusal superseded by an earlier pending deferral for the same task",
+						zap.String("task_id", taskID),
+						zap.String("stored_kind", string(existingCeiling.Kind)),
+						zap.String("superseded_kind", string(kind)),
+						zap.String(ceilingFieldReasonCode, ceilingReasonSuperseded))
+					result = fmt.Errorf("%w: task %s already holds %s", ErrCeilingLaunchConflict, taskID, existingCeiling.Kind)
+					return
+				}
+				if existingCeiling.ReasonCode == reasonCode {
+					// AC-12a/AC-49e: the same launch, refused for the same reason,
+					// already recorded. Leave it exactly as stored, including its
+					// queued_at (AC-32) and its card-surface bookkeeping.
+					reconcile = true
+					return
+				}
+				// AC-49f: the same pending launch, but the reason changed (for
+				// example a transient population read failure resolved into an
+				// ordinary refusal). Update the stored reason and this refusal's
+				// population/ceiling reading, preserve queued_at, and reset the
+				// surface stamps so the new reason gets its own note (AC-49g1).
+				updated := existingCeiling
+				updated.ReasonCode = reasonCode
+				updated.Population = population
+				updated.PopulationKnown = populationKnown
+				updated.Ceiling = ceiling
+				if s.writeCeilingDeferralUpdate(admissionCtx, taskID, existingRaw, prior, updated) {
+					reconcile = true
+					surface = true
+					publish = true
+				}
+				return
+			}
+
+			record, discarded, replaced := models.MergeCeilingRecord(existingRaw, deferral)
+			if replaced {
+				s.logger.Zap().Warn("deferred_launch held a non-object value; the ceiling refusal replaced it",
+					zap.String("task_id", taskID), zap.Any("discarded_value", discarded))
+			}
+
+			stored, lostCompare, err := s.repo.SetTaskDeferredLaunchIfUnchanged(admissionCtx, taskID, prior, record)
+			if err != nil {
+				result = fmt.Errorf("writing deferred launch for task %s: %w", taskID, err)
+				return
+			}
+			if stored {
+				reconcile = true
+				surface = true
+				publish = true
+				return
+			}
+			if lostCompare {
+				continue
+			}
+			result = fmt.Errorf("writing deferred launch for task %s: repository reported neither stored nor a lost compare", taskID)
+			return
 		}
-		if stored {
-			s.reconcileQueuedTaskState(ctx, taskID)
-			s.attemptCeilingSurfaceWrite(ctx, taskID)
-			s.publishTaskUpdatedByID(ctx, taskID)
-			return nil
-		}
-		if lostCompare {
-			continue
-		}
-		return fmt.Errorf("writing deferred launch for task %s: repository reported neither stored nor a lost compare", taskID)
+		result = fmt.Errorf("%s: could not persist deferred launch for task %s after %d attempts",
+			ceilingReasonDeferWriteFailed, taskID, deferredLaunchCASRetryBudget)
+	}()
+	if result != nil {
+		return result
 	}
-	return fmt.Errorf("%s: could not persist deferred launch for task %s after %d attempts",
-		ceilingReasonDeferWriteFailed, taskID, deferredLaunchCASRetryBudget)
+	if reconcile {
+		s.reconcileQueuedTaskState(ctx, taskID)
+	}
+	if surface {
+		s.attemptCeilingSurfaceWrite(ctx, taskID)
+	}
+	// A lost compare means a concurrent writer already moved the record; their
+	// admission path will fire the side effects.
+	if publish {
+		s.publishTaskUpdatedByID(ctx, taskID)
+	}
+	return nil
 }
 
 // reconcileQueuedTaskState repairs legacy task rows as soon as the durable

apps/backend/internal/orchestrator/ceiling_dispatch_admission.go

@@ -0,0 +1,87 @@
+package orchestrator
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/kandev/kandev/internal/task/models"
+)
+
+type ceilingDispatchClaimContextKey struct{}
+
+func withCeilingDispatchClaim(ctx context.Context, claim *ceilingDeferredLaunchClaim) context.Context {
+	return context.WithValue(ctx, ceilingDispatchClaimContextKey{}, claim)
+}
+
+// admitCeilingDispatch orders final route validation and renewal of the exact
+// dispatch claim with route mutation. The renewal is the local admission
+// boundary; later route changes use lifecycle cancellation. No admission lock
+// or lock-bearing context escapes into runtime I/O or callback publication.
+func (s *Service) admitCeilingDispatch(ctx context.Context, taskID string) error {
+	claim, _ := ctx.Value(ceilingDispatchClaimContextKey{}).(*ceilingDeferredLaunchClaim)
+	binding := ceilingEntryBindingFromContext(ctx)
+	if claim == nil && binding == nil {
+		return nil
+	}
+	ctx, release := s.lockCeilingEntryAdmission(ctx, taskID)
+	defer release()
+	if claim != nil {
+		if claim.taskID != taskID {
+			return ErrCeilingLaunchSuperseded
+		}
+		bound, present, err := models.ReadCeilingWorkflowEntryBinding(claim.deferral.Payload)
+		if err != nil {
+			return err
+		}
+		if present {
+			binding = &bound
+		}
+		ctx = withCeilingEntryKind(ctx, claim.deferral.Kind)
+	}
+	if err := s.validateClaimedCeilingBinding(ctx, taskID, binding); err != nil {
+		return err
+	}
+	if claim == nil {
+		return nil
+	}
+	return s.renewCeilingDispatchClaim(ctx, claim)
+}
+
+func (s *Service) renewCeilingDispatchClaim(ctx context.Context, claim *ceilingDeferredLaunchClaim) error {
+	for attempt := 0; attempt < deferredLaunchCASRetryBudget; attempt++ {
+		record, prior, err := s.repo.GetTaskDeferredLaunch(ctx, claim.taskID)
+		if err != nil {
+			return err
+		}
+		current, err := models.ReadCeilingDeferral(record)
+		if err != nil {
+			return ErrCeilingLaunchClaimed
+		}
+		equivalent, err := sameCeilingDeferralIdentity(current, claim.deferral)
+		if err != nil {
+			return err
+		}
+		details, exists := models.ReadCeilingLaunchClaimDetails(record)
+		now := time.Now().UTC()
+		if !equivalent || !exists || details.ID != claim.id || details.Expired(now) {
+			return ErrCeilingLaunchClaimed
+		}
+		updated := cloneCeilingRecord(record)
+		updated[models.CeilingLaunchClaimKey] = map[string]interface{}{
+			"id": claim.id, "owner": claim.owner,
+			models.CeilingLaunchClaimExpiresAtKey: now.Add(ceilingClaimLeaseDuration).Format(time.RFC3339Nano),
+		}
+		stored, lostCompare, err := s.repo.SetTaskDeferredLaunchIfUnchanged(ctx, claim.taskID, prior, updated)
+		if err != nil {
+			return err
+		}
+		if stored {
+			return nil
+		}
+		if !lostCompare {
+			return fmt.Errorf("admit ceiling dispatch: repository reported no write")
+		}
+	}
+	return fmt.Errorf("admit ceiling dispatch: compare-and-set retries exhausted")
+}

apps/backend/internal/orchestrator/ceiling_dispatch_admission_test.go

@@ -0,0 +1,217 @@
+package orchestrator
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/kandev/kandev/internal/task/models"
+	sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite"
+	wfmodels "github.com/kandev/kandev/internal/workflow/models"
+	v1 "github.com/kandev/kandev/pkg/api/v1"
+)
+
+type ceilingDispatchFixture struct {
+	svc      *Service
+	repo     *sqliterepo.Repository
+	agent    *mockAgentManager
+	steps    *mockStepGetter
+	task     *models.Task
+	route    models.WorkflowSessionRoute
+	binding  models.CeilingWorkflowEntryBinding
+	deferral models.CeilingDeferral
+}
+
+func newCeilingDispatchFixture(t *testing.T) *ceilingDispatchFixture {
+	t.Helper()
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedTaskAndSession(t, repo, "dispatch-task", "dispatch-session", models.TaskSessionStateWaitingForInput)
+	seedExecutorRunning(t, repo, "dispatch-session", "dispatch-task", "dispatch-execution")
+	task, err := repo.GetTask(ctx, "dispatch-task")
+	require.NoError(t, err)
+	task.WorkflowStepID = "dispatch-step"
+	require.NoError(t, repo.UpdateTask(ctx, task))
+	session, err := repo.GetTaskSession(ctx, "dispatch-session")
+	require.NoError(t, err)
+	session.AgentProfileID = "profile-1"
+	require.NoError(t, repo.UpdateTaskSession(ctx, session))
+	steps := newMockStepGetter()
+	steps.steps[task.WorkflowStepID] = &wfmodels.WorkflowStep{
+		ID: task.WorkflowStepID, WorkflowID: "wf1", Name: "Dispatch", Prompt: "continue",
+	}
+	tasks := newMockTaskRepo()
+	seedMockTaskState(tasks, task.ID, v1.TaskStateInProgress)
+	agent := &mockAgentManager{isAgentRunning: true, repoForExecutionLookup: repo}
+	svc := createTestServiceWithScheduler(repo, steps, tasks, agent)
+	svc.sessionCeiling = newSessionCeilingController(unlimitedSessionCeiling, nil, nil)
+	route := models.WorkflowSessionRoute{
+		OperationID: "dispatch-route", DestinationStepID: task.WorkflowStepID,
+		EntryIdentity: svc.workflowEntryIdentity(ctx, task.ID), TargetKind: "new_session",
+		DestinationID: session.ID, Phase: "committed",
+	}
+	require.NoError(t, repo.SetTaskMetadataKey(ctx, task.ID, models.MetaKeyWorkflowSessionRoute, route))
+	binding := models.CeilingWorkflowEntryBinding{
+		WorkflowID: "wf1", DestinationStepID: route.DestinationStepID,
+		RouteOperationID: route.OperationID, EntryIdentity: route.EntryIdentity, DestinationSessionID: session.ID,
+	}
+	deferral := models.CeilingDeferral{
+		Kind: models.CeilingLaunchWorkflowStepEnsure, Origin: string(launchOriginAutomatic),
+		ReasonCode: ceilingReasonRefused, QueuedAt: time.Now().UTC(),
+		Payload: map[string]interface{}{
+			metaKeySessionID: session.ID, metaKeyWorkflowStepID: task.WorkflowStepID,
+			models.CeilingLaunchEntryBindingKey: ceilingEntryBindingValue(binding),
+		},
+	}
+	require.NoError(t, repo.SetTaskMetadataKey(ctx, task.ID, models.MetaKeyDeferredLaunch, models.CeilingRecordKeys(deferral)))
+	return &ceilingDispatchFixture{svc, repo, agent, steps, task, route, binding, deferral}
+}
+
+// @covers AC-TASKS-QUEUED-SESSION-OWNERSHIP-002.6
+func TestCeilingReplayRejectsSuccessorClaimAfterPreparation(t *testing.T) {
+	for _, changeRoute := range []bool{false, true} {
+		t.Run(fmt.Sprintf("route_changed=%t", changeRoute), func(t *testing.T) {
+			testCeilingReplaySuccessorClaim(t, changeRoute)
+		})
+	}
+}
+
+func testCeilingReplaySuccessorClaim(t *testing.T, changeRoute bool) {
+	f := newCeilingDispatchFixture(t)
+	ctx := context.Background()
+	replaced := false
+	f.steps.getStepFunc = func(ctx context.Context, id string) (*wfmodels.WorkflowStep, error) {
+		record, _, err := f.repo.GetTaskDeferredLaunch(ctx, f.task.ID)
+		require.NoError(t, err)
+		if _, _, claimed := models.ReadCeilingLaunchClaim(record); claimed && !replaced {
+			replaced = true
+			record[models.CeilingLaunchClaimKey] = map[string]interface{}{
+				"id": "successor-claim", "owner": ceilingClaimOwnerSendNow,
+				models.CeilingLaunchClaimExpiresAtKey: time.Now().Add(time.Hour).UTC().Format(time.RFC3339Nano),
+			}
+			require.NoError(t, f.repo.SetTaskMetadataKey(ctx, f.task.ID, models.MetaKeyDeferredLaunch, record))
+			if changeRoute {
+				successor := f.route
+				successor.OperationID = "successor-route"
+				require.NoError(t, f.repo.SetTaskMetadataKey(ctx, f.task.ID, models.MetaKeyWorkflowSessionRoute, successor))
+			}
+		}
+		return f.steps.steps[id], nil
+	}
+	f.svc.retryOneDeferredCeilingLaunch(ctx, f.task)
+	require.True(t, replaced)
+	require.Empty(t, f.agent.capturedPrompts, "the replaced claim must not dispatch")
+	record, _, err := f.repo.GetTaskDeferredLaunch(ctx, f.task.ID)
+	require.NoError(t, err)
+	id, _, exists := models.ReadCeilingLaunchClaim(record)
+	require.True(t, exists)
+	require.Equal(t, "successor-claim", id)
+}
+
+func TestCeilingPromptRevalidatesAfterDispatchReceipt(t *testing.T) {
+	f := newCeilingDispatchFixture(t)
+	ctx := withCeilingEntryKind(withCeilingEntryBinding(context.Background(), &f.binding), f.deferral.Kind)
+	var afterAdmissionCalled bool
+	_, err := f.svc.promptTask(ctx, f.task.ID, f.route.DestinationID, "old prompt", "", false, nil, false,
+		launchOriginAutomatic, promptTaskOptions{beforeDispatch: func() error {
+			successor := f.route
+			successor.OperationID = "successor-route"
+			return f.svc.persistWorkflowSessionRoute(context.Background(), f.task.ID, successor)
+		}, afterDispatchAdmission: func() error {
+			afterAdmissionCalled = true
+			return nil
+		}})
+	require.ErrorIs(t, err, ErrCeilingLaunchSuperseded)
+	require.Empty(t, f.agent.capturedPrompts)
+	require.False(t, afterAdmissionCalled, "post-admission hooks must not run for a stale claim")
+}
+
+func TestCeilingModelSwitchRevalidatesAfterDispatchReceipt(t *testing.T) {
+	f := newCeilingDispatchFixture(t)
+	ctx := withCeilingEntryKind(withCeilingEntryBinding(context.Background(), &f.binding), f.deferral.Kind)
+	session, err := f.repo.GetTaskSession(ctx, f.route.DestinationID)
+	require.NoError(t, err)
+	session.AgentProfileSnapshot = map[string]interface{}{"model": "old-model"}
+	_, _, err = f.svc.attemptModelSwitchForPrompt(ctx, f.task.ID, session.ID, "new-model", "old prompt", session, &foregroundDispatch{}, func() error {
+		successor := f.route
+		successor.OperationID = "successor-route"
+		return f.svc.persistWorkflowSessionRoute(context.Background(), f.task.ID, successor)
+	}, nil)
+	require.ErrorIs(t, err, ErrCeilingLaunchSuperseded)
+	require.Empty(t, f.agent.capturedPrompts)
+}
+
+type ceilingRenewalBarrier struct {
+	*sqliterepo.Repository
+	entered, release chan struct{}
+}
+
+func (r *ceilingRenewalBarrier) SetTaskDeferredLaunchIfUnchanged(ctx context.Context, id string, prior interface{}, record map[string]interface{}) (bool, bool, error) {
+	close(r.entered)
+	<-r.release
+	return r.Repository.SetTaskDeferredLaunchIfUnchanged(ctx, id, prior, record)
+}
+
+// Reviewer-requested contract coverage: every kind uses the same final local
+// admission boundary, whose claim renewal serializes with route mutation.
+func TestCeilingDispatchAdmissionSerializesRouteMutation(t *testing.T) {
+	for _, kind := range []models.CeilingLaunchKind{
+		models.CeilingLaunchStart, models.CeilingLaunchStartCreated, models.CeilingLaunchPromptEnsure,
+		models.CeilingLaunchWorkflowStepEnsure, models.CeilingLaunchQueueDrainEnsure,
+		models.CeilingLaunchResume, models.CeilingLaunchDynamicRelaunch,
+	} {
+		t.Run(string(kind), func(t *testing.T) {
+			testCeilingDispatchAdmissionSerialization(t, kind)
+		})
+	}
+}
+
+func testCeilingDispatchAdmissionSerialization(t *testing.T, kind models.CeilingLaunchKind) {
+	f := newCeilingDispatchFixture(t)
+	ctx := context.Background()
+	f.deferral.Kind = kind
+	require.NoError(t, f.repo.SetTaskMetadataKey(ctx, f.task.ID, models.MetaKeyDeferredLaunch, models.CeilingRecordKeys(f.deferral)))
+	claim, found, err := f.svc.claimCeilingDeferredLaunch(ctx, f.task.ID, f.route.DestinationID, ceilingClaimOwnerReplay)
+	require.NoError(t, err)
+	require.True(t, found)
+	require.NotNil(t, claim)
+	before, _, err := f.repo.GetTaskDeferredLaunch(ctx, f.task.ID)
+	require.NoError(t, err)
+	oldClaim, ok := models.ReadCeilingLaunchClaimDetails(before)
+	require.True(t, ok)
+	barrier := &ceilingRenewalBarrier{Repository: f.repo, entered: make(chan struct{}), release: make(chan struct{})}
+	f.svc.repo = barrier
+	var release sync.Once
+	t.Cleanup(func() { release.Do(func() { close(barrier.release) }) })
+	admitted := make(chan error, 1)
+	go func() { admitted <- f.svc.admitCeilingDispatch(withCeilingDispatchClaim(ctx, claim), f.task.ID) }()
+	awaitCeilingProgress(t, barrier.entered, "final claim renewal")
+	routeChanged := make(chan error, 1)
+	successor := f.route
+	successor.OperationID = "successor-route"
+	go func() { routeChanged <- f.svc.persistWorkflowSessionRoute(ctx, f.task.ID, successor) }()
+	require.Eventually(t, func() bool {
+		f.svc.ceilingEntryAdmissionLocksMu.Lock()
+		defer f.svc.ceilingEntryAdmissionLocksMu.Unlock()
+		entry := f.svc.ceilingEntryAdmissionLocks[f.task.ID]
+		return entry != nil && entry.refs == 2
+	}, time.Second, time.Millisecond, "route mutation must wait for final claim renewal")
+	release.Do(func() { close(barrier.release) })
+	require.NoError(t, <-admitted)
+	require.NoError(t, <-routeChanged)
+	after, _, err := f.repo.GetTaskDeferredLaunch(ctx, f.task.ID)
+	require.NoError(t, err)
+	newClaim, ok := models.ReadCeilingLaunchClaimDetails(after)
+	require.True(t, ok)
+	require.Equal(t, oldClaim.ID, newClaim.ID)
+	require.True(t, newClaim.ExpiresAt.After(oldClaim.ExpiresAt))
+	current, err := f.repo.GetTask(ctx, f.task.ID)
+	require.NoError(t, err)
+	route, ok := models.LoadWorkflowSessionRoute(current.Metadata)
+	require.True(t, ok)
+	require.Equal(t, successor.OperationID, route.OperationID)
+}

apps/backend/internal/orchestrator/ceiling_entry.go

@@ -628,12 +628,20 @@ func ceilingDeferralsEquivalentForAdmission(a, b models.CeilingDeferral) (bool,
 // validateCeilingEntry compares a claimed record with the current task-owned
 // route and destination. It is deliberately read-only. Callers retain the
 // record on unavailable reads and only terminally dispose superseded entries.
+// It acquires task admission; a context that already owns that admission is
+// re-entrant.
 func (s *Service) validateCeilingEntry(
 	ctx context.Context,
 	task *models.Task,
 	deferral models.CeilingDeferral,
 ) (ceilingEntryDisposition, string, error) {
-	return s.validateCeilingEntryWithDestinationState(ctx, task, deferral, true)
+	taskID := ""
+	if task != nil {
+		taskID = task.ID
+	}
+	admissionCtx, release := s.lockCeilingEntryAdmission(ctx, taskID)
+	defer release()
+	return s.validateCeilingEntryWithDestinationState(admissionCtx, task, deferral, true)
 }
 
 // validateCeilingEntryWithDestinationState checks the task-owned workflow
@@ -826,6 +834,12 @@ func (s *Service) validateClaimedCeilingBinding(
 	if binding == nil {
 		return nil
 	}
+	// Binding validation is the final route read before a concrete launch. Use
+	// the same short admission section as route mutation. Nested callers that
+	// already own the section are re-entrant through the context marker.
+	admissionCtx, release := s.lockCeilingEntryAdmission(ctx, taskID)
+	defer release()
+	ctx = admissionCtx
 	task, err := s.repo.GetTask(ctx, taskID)
 	if err != nil {
 		return fmt.Errorf("reload task for deferred workflow entry: %w", err)

apps/backend/internal/orchestrator/ceiling_replay.go

@@ -186,6 +186,7 @@ func (s *Service) retryOneDeferredCeilingLaunch(ctx context.Context, task *model
 		return
 	}
 	deferral = claim.deferral
+	ctx = withCeilingDispatchClaim(ctx, claim)
 
 	defer claim.releaseIfHeld(ctx)
 	currentTask, currentErr := s.repo.GetTask(ctx, task.ID)
@@ -352,6 +353,14 @@ func (s *Service) clearCeilingDeferredRecord(ctx context.Context, taskID string,
 				return
 			}
 		}
+		if claim, ok := ctx.Value(ceilingDispatchClaimContextKey{}).(*ceilingDeferredLaunchClaim); ok && claim.taskID == taskID {
+			claimID, _, held := models.ReadCeilingLaunchClaim(existingRaw)
+			if !held || claimID != claim.id {
+				s.logger.Zap().Debug("skipping ceiling record clear: claim replaced by successor",
+					zap.String("task_id", taskID))
+				return
+			}
+		}
 		record := stripCeilingRecordKeys(existingRaw)
 		stored, lostCompare, err := s.repo.SetTaskDeferredLaunchIfUnchanged(ctx, taskID, prior, record)
 		if err != nil {

apps/backend/internal/orchestrator/ceiling_replay_lock_order_test.go

@@ -0,0 +1,418 @@
+package orchestrator
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/kandev/kandev/internal/orchestrator/executor"
+	"github.com/kandev/kandev/internal/task/models"
+	sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite"
+	wfmodels "github.com/kandev/kandev/internal/workflow/models"
+	v1 "github.com/kandev/kandev/pkg/api/v1"
+	"github.com/stretchr/testify/require"
+)
+
+func TestCeilingReplaySynchronousReconciliationCallback(t *testing.T) {
+	ctx := context.Background()
+	const (
+		taskID    = "ceiling-replay-lock-order-task"
+		sessionID = "ceiling-replay-lock-order-session"
+		stepID    = "ceiling-replay-lock-order-step"
+	)
+
+	repo := setupTestRepo(t)
+	seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateWaitingForInput)
+	seedExecutorRunning(t, repo, sessionID, taskID, "ceiling-replay-lock-order-execution")
+
+	task, err := repo.GetTask(ctx, taskID)
+	require.NoError(t, err)
+	task.WorkflowStepID = stepID
+	require.NoError(t, repo.UpdateTask(ctx, task))
+	session, err := repo.GetTaskSession(ctx, sessionID)
+	require.NoError(t, err)
+	session.AgentProfileID = "profile-1"
+	require.NoError(t, repo.UpdateTaskSession(ctx, session))
+
+	stepGetter := newMockStepGetter()
+	stepGetter.steps[stepID] = &wfmodels.WorkflowStep{
+		ID: stepID, WorkflowID: "wf1", Name: "Replay step", Prompt: "continue",
+	}
+	taskRepo := newMockTaskRepo()
+	seedMockTaskState(taskRepo, taskID, v1.TaskStateInProgress)
+
+	var svc *Service
+	promptEntered := make(chan struct{})
+	agentMgr := &mockAgentManager{
+		isAgentRunning:         true,
+		repoForExecutionLookup: repo,
+		promptAgentFunc: func(_ context.Context, _ string, _ string, _ []v1.MessageAttachment, _ bool) (*executor.PromptResult, error) {
+			close(promptEntered)
+			// A synchronous provider callback is allowed to reconcile the task
+			// while replay is dispatching. This is the callback cycle that must
+			// not inherit replay's task-admission ownership.
+			svc.writeTaskReviewState(context.Background(), taskID, sessionID)
+			return &executor.PromptResult{}, nil
+		},
+	}
+	svc = createTestServiceWithScheduler(repo, stepGetter, taskRepo, agentMgr)
+	svc.sessionCeiling = newSessionCeilingController(unlimitedSessionCeiling, nil, nil)
+
+	entryIdentity := svc.workflowEntryIdentity(ctx, taskID)
+	route := models.WorkflowSessionRoute{
+		OperationID:       "ceiling-replay-lock-order-route",
+		DestinationStepID: stepID,
+		EntryIdentity:     entryIdentity,
+		TargetKind:        "new_session",
+		DestinationID:     sessionID,
+		Phase:             "committed",
+	}
+	require.NoError(t, repo.SetTaskMetadataKey(ctx, taskID, models.MetaKeyWorkflowSessionRoute, route))
+	deferral := models.CeilingDeferral{
+		Kind: models.CeilingLaunchWorkflowStepEnsure,
+		Payload: map[string]interface{}{
+			metaKeySessionID:      sessionID,
+			metaKeyWorkflowStepID: stepID,
+			models.CeilingLaunchEntryBindingKey: map[string]interface{}{
+				"workflow_id":            "wf1",
+				"destination_step_id":    stepID,
+				"route_operation_id":     route.OperationID,
+				"entry_identity":         entryIdentity,
+				"destination_session_id": sessionID,
+			},
+		},
+		Origin:     string(launchOriginAutomatic),
+		ReasonCode: ceilingReasonRefused,
+		QueuedAt:   time.Now().UTC(),
+	}
+
+	done := make(chan ceilingReplayOutcome, 1)
+	go func() {
+		done <- svc.replayCeilingDeferral(ctx, &models.Task{ID: taskID}, deferral)
+	}()
+
+	select {
+	case outcome := <-done:
+		require.Equal(t, ceilingReplaySucceeded, outcome)
+	case <-time.After(500 * time.Millisecond):
+		t.Fatal("ceiling replay blocked while a synchronous dispatch callback reconciled task state")
+	}
+
+	select {
+	case <-promptEntered:
+	case <-time.After(500 * time.Millisecond):
+		t.Fatal("ceiling replay did not reach the provider dispatch boundary")
+	}
+}
+
+func TestCeilingReplayRevalidatesAfterRuntimePreparation(t *testing.T) {
+	ctx := context.Background()
+	const (
+		taskID    = "ceiling-replay-preparation-task"
+		sessionID = "ceiling-replay-preparation-session"
+		stepID    = "ceiling-replay-preparation-step"
+	)
+
+	repo := setupTestRepo(t)
+	seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateWaitingForInput)
+	seedExecutorRunning(t, repo, sessionID, taskID, "ceiling-replay-preparation-execution")
+	task, err := repo.GetTask(ctx, taskID)
+	require.NoError(t, err)
+	task.WorkflowStepID = stepID
+	require.NoError(t, repo.UpdateTask(ctx, task))
+	session, err := repo.GetTaskSession(ctx, sessionID)
+	require.NoError(t, err)
+	session.AgentProfileID = "profile-1"
+	require.NoError(t, repo.UpdateTaskSession(ctx, session))
+
+	step := &wfmodels.WorkflowStep{
+		ID: stepID, WorkflowID: "wf1", Name: "Replay step", Prompt: "continue",
+	}
+	stepGetter := newMockStepGetter()
+	stepGetter.steps[stepID] = step
+	stepGetter.steps[stepID+"-successor"] = &wfmodels.WorkflowStep{
+		ID: stepID + "-successor", WorkflowID: "wf1", Name: "Successor", Prompt: "new",
+	}
+	taskRepo := newMockTaskRepo()
+	seedMockTaskState(taskRepo, taskID, v1.TaskStateInProgress)
+	agentMgr := &mockAgentManager{
+		isAgentRunning:         true,
+		repoForExecutionLookup: repo,
+	}
+	svc := createTestServiceWithScheduler(repo, stepGetter, taskRepo, agentMgr)
+	svc.sessionCeiling = newSessionCeilingController(unlimitedSessionCeiling, nil, nil)
+
+	entryIdentity := svc.workflowEntryIdentity(ctx, taskID)
+	route := models.WorkflowSessionRoute{
+		OperationID:       "ceiling-replay-preparation-route",
+		DestinationStepID: stepID,
+		EntryIdentity:     entryIdentity,
+		TargetKind:        "new_session",
+		DestinationID:     sessionID,
+		Phase:             "committed",
+	}
+	require.NoError(t, repo.SetTaskMetadataKey(ctx, taskID, models.MetaKeyWorkflowSessionRoute, route))
+
+	var stepReads int
+	stepGetter.getStepFunc = func(stepCtx context.Context, requestedStepID string) (*wfmodels.WorkflowStep, error) {
+		stepReads++
+		if stepReads == 2 {
+			successor := route
+			successor.OperationID = "ceiling-replay-preparation-successor-route"
+			successor.DestinationStepID = stepID + "-successor"
+			successor.EntryIdentity = "successor-entry"
+			require.NoError(t, repo.SetTaskMetadataKey(stepCtx, taskID, models.MetaKeyWorkflowSessionRoute, successor))
+		}
+		return stepGetter.steps[requestedStepID], nil
+	}
+	deferral := models.CeilingDeferral{
+		Kind: models.CeilingLaunchWorkflowStepEnsure,
+		Payload: map[string]interface{}{
+			metaKeySessionID:      sessionID,
+			metaKeyWorkflowStepID: stepID,
+			models.CeilingLaunchEntryBindingKey: map[string]interface{}{
+				"workflow_id":            "wf1",
+				"destination_step_id":    stepID,
+				"route_operation_id":     route.OperationID,
+				"entry_identity":         entryIdentity,
+				"destination_session_id": sessionID,
+			},
+		},
+		Origin:     string(launchOriginAutomatic),
+		ReasonCode: ceilingReasonRefused,
+		QueuedAt:   time.Now().UTC(),
+	}
+
+	outcome := svc.replayCeilingDeferral(ctx, &models.Task{ID: taskID}, deferral)
+	require.Equal(t, ceilingReplaySuperseded, outcome)
+	agentMgr.mu.Lock()
+	require.Empty(t, agentMgr.capturedPrompts)
+	agentMgr.mu.Unlock()
+}
+
+func TestCeilingReplayPreservesReconciliationPrecedence(t *testing.T) {
+	tests := []struct {
+		name            string
+		configure       func(*testing.T, *sqliterepo.Repository, string, string, string)
+		wantState       v1.TaskState
+		wantStateWrites int
+	}{
+		{
+			name: "working sibling preserves active task",
+			configure: func(t *testing.T, repo *sqliterepo.Repository, taskID, sessionID, _ string) {
+				now := time.Now().UTC()
+				require.NoError(t, repo.CreateTaskSession(context.Background(), &models.TaskSession{
+					ID: "reconciliation-working-sibling", TaskID: taskID,
+					State: models.TaskSessionStateRunning, StartedAt: now, UpdatedAt: now,
+				}))
+			},
+			wantState:       v1.TaskStateInProgress,
+			wantStateWrites: 0,
+		},
+		{
+			name: "valid destination preserves scheduling",
+			configure: func(t *testing.T, repo *sqliterepo.Repository, taskID, _, destinationID string) {
+				deferral := models.CeilingDeferral{
+					Kind:    models.CeilingLaunchStartCreated,
+					Payload: map[string]interface{}{metaKeySessionID: destinationID},
+					Origin:  string(launchOriginAutomatic), ReasonCode: ceilingReasonRefused,
+					QueuedAt: time.Now().UTC(),
+				}
+				require.NoError(t, repo.SetTaskMetadataKey(context.Background(), taskID,
+					models.MetaKeyDeferredLaunch, models.CeilingRecordKeys(deferral)))
+			},
+			wantState:       v1.TaskStateScheduling,
+			wantStateWrites: 1,
+		},
+		{
+			name: "idle sibling permits review",
+			configure: func(t *testing.T, repo *sqliterepo.Repository, taskID, _, _ string) {
+				now := time.Now().UTC()
+				require.NoError(t, repo.CreateTaskSession(context.Background(), &models.TaskSession{
+					ID: "reconciliation-idle-sibling", TaskID: taskID,
+					State: models.TaskSessionStateIdle, StartedAt: now, UpdatedAt: now,
+				}))
+			},
+			wantState:       v1.TaskStateReview,
+			wantStateWrites: 1,
+		},
+		{
+			name: "failed sibling permits review",
+			configure: func(t *testing.T, repo *sqliterepo.Repository, taskID, _, _ string) {
+				now := time.Now().UTC()
+				require.NoError(t, repo.CreateTaskSession(context.Background(), &models.TaskSession{
+					ID: "reconciliation-failed-sibling", TaskID: taskID,
+					State: models.TaskSessionStateFailed, StartedAt: now, UpdatedAt: now,
+				}))
+			},
+			wantState:       v1.TaskStateReview,
+			wantStateWrites: 1,
+		},
+		{
+			name: "terminal destination falls back to review",
+			configure: func(t *testing.T, repo *sqliterepo.Repository, taskID, _, destinationID string) {
+				deferral := models.CeilingDeferral{
+					Kind:    models.CeilingLaunchStartCreated,
+					Payload: map[string]interface{}{metaKeySessionID: destinationID},
+					Origin:  string(launchOriginAutomatic), ReasonCode: ceilingReasonRefused,
+					QueuedAt: time.Now().UTC(),
+				}
+				require.NoError(t, repo.SetTaskMetadataKey(context.Background(), taskID,
+					models.MetaKeyDeferredLaunch, models.CeilingRecordKeys(deferral)))
+				require.NoError(t, repo.UpdateTaskSessionState(context.Background(), destinationID,
+					models.TaskSessionStateCompleted, ""))
+			},
+			wantState:       v1.TaskStateReview,
+			wantStateWrites: 1,
+		},
+		{
+			name: "archived task is unchanged",
+			configure: func(t *testing.T, repo *sqliterepo.Repository, taskID, _, _ string) {
+				require.NoError(t, repo.ArchiveTask(context.Background(), taskID))
+			},
+			wantState:       v1.TaskStateInProgress,
+			wantStateWrites: 0,
+		},
+		{
+			name: "office task is unchanged",
+			configure: func(t *testing.T, repo *sqliterepo.Repository, taskID, _, _ string) {
+				task, err := repo.GetTask(context.Background(), taskID)
+				require.NoError(t, err)
+				task.ProjectID = "office-project"
+				require.NoError(t, repo.UpdateTask(context.Background(), task))
+			},
+			wantState:       v1.TaskStateInProgress,
+			wantStateWrites: 0,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := context.Background()
+			taskID := "reconciliation-" + strings.ReplaceAll(tc.name, " ", "-")
+			sessionID := taskID + "-session"
+			repo := setupTestRepo(t)
+			seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateWaitingForInput)
+			tc.configure(t, repo, taskID, sessionID, sessionID)
+
+			taskRepo := newMockTaskRepo()
+			seedMockTaskState(taskRepo, taskID, v1.TaskStateInProgress)
+			svc := createTestService(repo, newMockStepGetter(), taskRepo)
+			svc.sessionCeiling = newSessionCeilingController(unlimitedSessionCeiling, nil, nil)
+			svc.writeTaskReviewState(ctx, taskID, sessionID)
+
+			taskRepo.mu.Lock()
+			gotState := taskRepo.updatedStates[taskID]
+			writes := taskRepo.stateWrites[taskID]
+			taskRepo.mu.Unlock()
+			if writes == 0 {
+				gotState = v1.TaskStateInProgress
+			}
+			require.Equal(t, tc.wantState, gotState)
+			require.Equal(t, tc.wantStateWrites, writes)
+		})
+	}
+}
+
+func TestCeilingReplayClaimAndShutdownProgress(t *testing.T) {
+	ctx := context.Background()
+	svc, repo := newServiceWithRealRepo(t)
+	const (
+		taskID    = "ceiling-replay-claim-shutdown-task"
+		sessionID = "ceiling-replay-claim-shutdown-session"
+	)
+	now := time.Now().UTC()
+	require.NoError(t, repo.CreateTask(ctx, &models.Task{
+		ID: taskID, Title: "claim and shutdown", State: v1.TaskStateScheduling,
+		CreatedAt: now, UpdatedAt: now,
+	}))
+	deferral := models.CeilingDeferral{
+		Kind:    models.CeilingLaunchStartCreated,
+		Payload: map[string]interface{}{metaKeySessionID: sessionID},
+		Origin:  string(launchOriginAutomatic), ReasonCode: ceilingReasonRefused,
+		QueuedAt: now,
+	}
+	_, _, err := repo.SetTaskDeferredLaunchIfUnchanged(
+		ctx, taskID, sqliterepo.AbsentDeferredLaunch(), models.CeilingRecordKeys(deferral),
+	)
+	require.NoError(t, err)
+
+	// Both dispatcher classes may race for one durable record, but only one
+	// receives the claim. The losing path must leave the record for retry.
+	type claimResult struct {
+		claim *ceilingDeferredLaunchClaim
+		found bool
+		err   error
+	}
+	ready := make(chan struct{}, 2)
+	start := make(chan struct{})
+	results := make(chan claimResult, 2)
+	for _, owner := range []string{ceilingClaimOwnerReplay, ceilingClaimOwnerSendNow} {
+		go func(owner string) {
+			ready <- struct{}{}
+			<-start
+			claim, found, claimErr := svc.claimCeilingDeferredLaunch(ctx, taskID, sessionID, owner)
+			results <- claimResult{claim: claim, found: found, err: claimErr}
+		}(owner)
+	}
+	<-ready
+	<-ready
+	close(start)
+	first := <-results
+	second := <-results
+	require.NoError(t, first.err)
+	require.NoError(t, second.err)
+	require.True(t, first.found)
+	require.True(t, second.found)
+	require.NotEqual(t, first.claim == nil, second.claim == nil)
+	ownerClaim := first.claim
+	if ownerClaim == nil {
+		ownerClaim = second.claim
+	}
+	require.NotNil(t, ownerClaim)
+
+	cancelledCtx, cancel := context.WithCancel(ctx)
+	cancel()
+	ownerClaim.releaseIfHeld(cancelledCtx)
+	claimAfterCancellation, found, err := svc.claimCeilingDeferredLaunch(
+		ctx, taskID, sessionID, ceilingClaimOwnerSendNow,
+	)
+	require.NoError(t, err)
+	require.True(t, found)
+	require.NotNil(t, claimAfterCancellation, "cancelled replay cleanup must return the record to the next dispatcher")
+	claimAfterCancellation.releaseIfHeld(ctx)
+
+	// Shutdown joins a sweep callback that is already in progress. Releasing
+	// the callback barrier must let stop complete without a detached worker.
+	svc.ceilingSweeper = newCeilingSweeper()
+	svc.ceilingSweeper.interval = time.Hour
+	tickStarted := make(chan struct{})
+	releaseTick := make(chan struct{})
+	require.True(t, svc.ceilingSweeper.start(ctx, func(context.Context) {
+		close(tickStarted)
+		<-releaseTick
+	}))
+	svc.ceilingSweeper.signalNow()
+	select {
+	case <-tickStarted:
+	case <-time.After(500 * time.Millisecond):
+		t.Fatal("sweeper did not enter its barrier-controlled callback")
+	}
+	stopped := make(chan struct{})
+	go func() {
+		svc.stopCeilingSweeper()
+		close(stopped)
+	}()
+	select {
+	case <-stopped:
+		t.Fatal("sweeper shutdown returned before its callback released")
+	default:
+	}
+	close(releaseTick)
+	select {
+	case <-stopped:
+	case <-time.After(500 * time.Millisecond):
+		t.Fatal("sweeper shutdown remained blocked after callback release")
+	}
+}

apps/backend/internal/orchestrator/ceiling_replay_progress_test.go

@@ -0,0 +1,254 @@
+package orchestrator
+
+import (
+	"context"
+	"encoding/json"
+	"os"
+	"os/exec"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/kandev/kandev/internal/agent/runtime/lifecycle"
+	"github.com/kandev/kandev/internal/events"
+	"github.com/kandev/kandev/internal/events/bus"
+	"github.com/kandev/kandev/internal/orchestrator/watcher"
+	"github.com/kandev/kandev/internal/task/models"
+	sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite"
+	v1 "github.com/kandev/kandev/pkg/api/v1"
+)
+
+// Isolate an actual lock cycle so a regression cannot leak blocked production
+// goroutines into the parent test process or hold its SQLite cleanup open.
+// @covers AC-TASKS-WORKFLOW-CANCELLED-TURN-COMPLETION-001.2
+func TestCeilingReplayBootReadyAndCancellationProgress(t *testing.T) {
+	const childKey = "KANDEV_TEST_CEILING_PROGRESS_CHILD"
+	if mode := os.Getenv(childKey); mode != "" {
+		runCeilingReplayProgress(t, mode == "async")
+		return
+	}
+	for _, mode := range []string{"sync", "async"} {
+		t.Run(mode, func(t *testing.T) {
+			ctx, cancel := context.WithTimeout(context.Background(), 25*time.Second)
+			defer cancel()
+			binary, err := os.Executable()
+			require.NoError(t, err)
+			cmd := exec.CommandContext(ctx, binary,
+				"-test.run=^TestCeilingReplayBootReadyAndCancellationProgress$", "-test.timeout=20s")
+			cmd.Env = append(os.Environ(), childKey+"="+mode)
+			output, err := cmd.CombinedOutput()
+			require.NoError(t, err, "%s", output)
+		})
+	}
+}
+
+type ceilingReplayReadBarrier struct {
+	*sqliterepo.Repository
+	taskID           string
+	entered, release chan struct{}
+	once             sync.Once
+}
+
+func (r *ceilingReplayReadBarrier) GetTask(ctx context.Context, id string) (*models.Task, error) {
+	if id == r.taskID && ceilingEntryAdmissionLockHeld(ctx, id) {
+		record, _, err := r.GetTaskDeferredLaunch(ctx, id)
+		if err != nil {
+			return nil, err
+		}
+		if _, _, claimed := models.ReadCeilingLaunchClaim(record); claimed {
+			r.once.Do(func() { close(r.entered); <-r.release })
+		}
+	}
+	return r.Repository.GetTask(ctx, id)
+}
+
+// As with NATS, publication acknowledges delivery scheduling, not completion
+// of the local subscriber. Different subjects can execute concurrently.
+type ceilingAsyncBus struct {
+	bus.EventBus
+	wg sync.WaitGroup
+}
+
+func (b *ceilingAsyncBus) dispatch(handler bus.EventHandler) bus.EventHandler {
+	return func(ctx context.Context, event *bus.Event) error {
+		copy := *event
+		b.wg.Add(1)
+		go func() { defer b.wg.Done(); _ = handler(ctx, &copy) }()
+		return nil
+	}
+}
+
+func (b *ceilingAsyncBus) Subscribe(subject string, handler bus.EventHandler) (bus.Subscription, error) {
+	return b.EventBus.Subscribe(subject, b.dispatch(handler))
+}
+
+func (b *ceilingAsyncBus) QueueSubscribe(subject, queue string, handler bus.EventHandler) (bus.Subscription, error) {
+	return b.EventBus.QueueSubscribe(subject, queue, b.dispatch(handler))
+}
+
+func runCeilingReplayProgress(t *testing.T, async bool) {
+	f := newCeilingDispatchFixture(t)
+	ctx := context.Background()
+	seedCeilingProgressCompanions(t, f)
+	var eventBus bus.EventBus = bus.NewMemoryEventBus(testLogger())
+	if async {
+		asyncBus := &ceilingAsyncBus{EventBus: eventBus}
+		eventBus = asyncBus
+		t.Cleanup(asyncBus.wg.Wait)
+	}
+	t.Cleanup(eventBus.Close)
+	f.svc.eventBus = eventBus
+	f.svc.taskRepo = newTaskServiceStateRepository(f.repo, eventBus)
+	f.svc.turnService = &repoTurnService{repo: f.repo}
+	_, err := f.svc.turnService.StartTurn(ctx, "cancel-session")
+	require.NoError(t, err)
+	f.agent.cancelAgentEntered = make(chan struct{}, 1)
+	stream := &blockingToolCallMessageCreator{entered: make(chan struct{}), release: make(chan struct{})}
+	f.svc.messageCreator = stream
+	replay := &ceilingReplayReadBarrier{Repository: f.repo, taskID: f.task.ID,
+		entered: make(chan struct{}), release: make(chan struct{})}
+	f.svc.repo = replay
+	streamDone, bootDone := make(chan struct{}), make(chan struct{})
+	w := watcher.NewWatcher(eventBus, watcher.EventHandlers{
+		OnAgentBootReady: func(ctx context.Context, data watcher.AgentEventData) {
+			defer close(bootDone)
+			f.svc.handleAgentBootReady(ctx, data)
+		},
+		OnAgentStreamEvent: func(ctx context.Context, data *lifecycle.AgentStreamEventPayload) {
+			defer close(streamDone)
+			f.svc.handleAgentStreamEvent(ctx, data)
+		},
+	}, "ceiling-progress", testLogger())
+	require.NoError(t, w.Start(ctx))
+	t.Cleanup(func() { require.NoError(t, w.Stop()) })
+	reviewPublished := observeCeilingCancellationReview(t, eventBus)
+
+	streamPublish := make(chan error, 1)
+	go func() {
+		streamPublish <- eventBus.Publish(ctx, events.BuildAgentStreamSubject(f.route.DestinationID),
+			bus.NewEvent(events.AgentStream, "test", lifecycle.AgentStreamEventPayload{
+				TaskID: f.task.ID, SessionID: f.route.DestinationID, ExecutionID: "dispatch-execution",
+				Data: &lifecycle.AgentStreamEventData{Type: agentEventToolCall, ToolCallID: "progress-tool", ToolTitle: "Read", ToolStatus: "running"},
+			}))
+	}()
+	awaitCeilingProgress(t, stream.entered, "stream owns its cancellation guard")
+	sweepDone := make(chan struct{})
+	go func() { defer close(sweepDone); f.svc.drainDeferredCeilingLaunches(ctx) }()
+	awaitCeilingProgress(t, replay.entered, "claimed replay owns task admission")
+	bootPublish := make(chan error, 1)
+	go func() {
+		bootPublish <- eventBus.Publish(ctx, events.AgentBootReady, bus.NewEvent(events.AgentBootReady, "test",
+			watcher.AgentEventData{TaskID: f.task.ID, SessionID: "boot-session", AgentExecutionID: "boot-execution"}))
+	}()
+	// Only replay and boot-ready can own/request this task's admission here.
+	// Observing the second reference proves boot-ready reached the contention.
+	require.Eventually(t, func() bool {
+		f.svc.ceilingEntryAdmissionLocksMu.Lock()
+		defer f.svc.ceilingEntryAdmissionLocksMu.Unlock()
+		entry := f.svc.ceilingEntryAdmissionLocks[f.task.ID]
+		return entry != nil && entry.refs == 2
+	}, 5*time.Second, time.Millisecond, "boot-ready did not reach task admission")
+	cancelDone := make(chan error, 1)
+	go func() { cancelDone <- f.svc.CancelAgent(ctx, "cancel-session") }()
+	awaitCeilingProgress(t, f.agent.cancelAgentEntered, "unrelated cancellation reached runtime")
+	close(replay.release)
+	close(stream.release)
+	awaitCeilingProgress(t, streamDone, "stream completed")
+	awaitCeilingProgress(t, bootDone, "boot-ready completed")
+	awaitCeilingProgress(t, sweepDone, "sweep reached the second task")
+	require.NoError(t, <-cancelDone)
+	require.NoError(t, <-streamPublish)
+	require.NoError(t, <-bootPublish)
+	awaitCeilingProgress(t, reviewPublished, "cancellation published REVIEW")
+	assertCeilingProgressSettled(t, f)
+}
+
+func seedCeilingProgressCompanions(t *testing.T, f *ceilingDispatchFixture) {
+	t.Helper()
+	ctx := context.Background()
+	now := time.Now().UTC()
+	require.NoError(t, f.repo.CreateTaskSession(ctx, &models.TaskSession{
+		ID: "boot-session", TaskID: f.task.ID, State: models.TaskSessionStateStarting, StartedAt: now, UpdatedAt: now,
+	}))
+	seedTaskAndSession(t, f.repo, "cancel-task", "cancel-session", models.TaskSessionStateRunning)
+	seedExecutorRunning(t, f.repo, "cancel-session", "cancel-task", "cancel-execution")
+	seedTaskAndSession(t, f.repo, "zz-second-task", "second-session", models.TaskSessionStateWaitingForInput)
+	seedExecutorRunning(t, f.repo, "second-session", "zz-second-task", "second-execution")
+	second, err := f.repo.GetTask(ctx, "zz-second-task")
+	require.NoError(t, err)
+	second.WorkflowStepID = f.task.WorkflowStepID
+	require.NoError(t, f.repo.UpdateTask(ctx, second))
+	session, err := f.repo.GetTaskSession(ctx, "second-session")
+	require.NoError(t, err)
+	session.AgentProfileID = "profile-1"
+	require.NoError(t, f.repo.UpdateTaskSession(ctx, session))
+	route := f.route
+	route.OperationID, route.DestinationID = "second-route", session.ID
+	route.EntryIdentity = f.svc.workflowEntryIdentity(ctx, second.ID)
+	require.NoError(t, f.repo.SetTaskMetadataKey(ctx, second.ID, models.MetaKeyWorkflowSessionRoute, route))
+	binding := f.binding
+	binding.RouteOperationID, binding.DestinationSessionID, binding.EntryIdentity = route.OperationID, session.ID, route.EntryIdentity
+	deferral := f.deferral
+	deferral.Payload = map[string]interface{}{
+		metaKeySessionID: session.ID, metaKeyWorkflowStepID: second.WorkflowStepID,
+		models.CeilingLaunchEntryBindingKey: ceilingEntryBindingValue(binding),
+	}
+	require.NoError(t, f.repo.SetTaskMetadataKey(ctx, second.ID, models.MetaKeyDeferredLaunch, models.CeilingRecordKeys(deferral)))
+	seedMockTaskState(f.svc.taskRepo.(*mockTaskRepo), second.ID, v1.TaskStateInProgress)
+}
+
+func observeCeilingCancellationReview(t *testing.T, eventBus bus.EventBus) <-chan struct{} {
+	t.Helper()
+	done := make(chan struct{})
+	var once sync.Once
+	_, err := eventBus.Subscribe(events.TaskStateChanged, func(_ context.Context, event *bus.Event) error {
+		data, err := json.Marshal(event.Data)
+		if err != nil {
+			return err
+		}
+		var update watcher.TaskEventData
+		if err := json.Unmarshal(data, &update); err != nil {
+			return err
+		}
+		if update.TaskID == "cancel-task" && update.NewState != nil && *update.NewState == v1.TaskStateReview {
+			once.Do(func() { close(done) })
+		}
+		return nil
+	})
+	require.NoError(t, err)
+	return done
+}
+
+func assertCeilingProgressSettled(t *testing.T, f *ceilingDispatchFixture) {
+	t.Helper()
+	ctx := context.Background()
+	task, err := f.repo.GetTask(ctx, "cancel-task")
+	require.NoError(t, err)
+	require.Equal(t, v1.TaskStateReview, task.State)
+	session, err := f.repo.GetTaskSession(ctx, "cancel-session")
+	require.NoError(t, err)
+	require.Equal(t, models.TaskSessionStateWaitingForInput, session.State)
+	turn, err := f.svc.turnService.GetActiveTurn(ctx, session.ID)
+	require.NoError(t, err)
+	require.Nil(t, turn)
+	require.False(t, f.svc.CancellationPending(session.ID))
+	f.agent.mu.Lock()
+	prompts := append([]string(nil), f.agent.capturedPrompts...)
+	f.agent.mu.Unlock()
+	require.Equal(t, []string{"continue"}, prompts, "only the second queued task dispatches")
+	record, _, err := f.repo.GetTaskDeferredLaunch(ctx, "zz-second-task")
+	require.NoError(t, err)
+	_, err = models.ReadCeilingDeferral(record)
+	require.Error(t, err, "accepted second task must be removed from the durable queue")
+}
+
+func awaitCeilingProgress(t *testing.T, signal <-chan struct{}, what string) {
+	t.Helper()
+	select {
+	case <-signal:
+	case <-time.After(5 * time.Second):
+		t.Fatalf("no progress: %s", what)
+	}
+}

apps/backend/internal/orchestrator/ceiling_seam3.go

@@ -176,6 +176,7 @@ func seam3PromptEnsureNonReconstructableOptionsSet(options promptTaskOptions) bo
 		options.expectedSessionIdentity != nil ||
 		options.afterDispatch != nil ||
 		options.beforeDispatch != nil ||
+		options.afterDispatchAdmission != nil ||
 		options.disableDispatchRetry
 }
 

apps/backend/internal/orchestrator/ceiling_surface.go

@@ -2,9 +2,12 @@ package orchestrator
 
 import (
 	"context"
+	"encoding/json"
+	"errors"
 	"fmt"
 	"time"
 
+	"github.com/google/uuid"
 	"go.uber.org/zap"
 
 	"github.com/kandev/kandev/internal/task/models"
@@ -16,6 +19,8 @@ import (
 // once at ERROR, leaving the record deferred and retryable.
 const ceilingSurfaceRetryBudget = 5
 
+var errCeilingSurfaceSuperseded = errors.New("ceiling surface belongs to a superseded deferral")
+
 // attemptCeilingSurfaceWrite is AC-49's single surfacing actor for the
 // ordinary refusal path (seams 2-5; seam 1 has no session and is excluded).
 // It writes the card note for a task's currently-stored ceiling deferral,
@@ -78,6 +83,9 @@ func (s *Service) writeCeilingSurfaceNoteAndStamp(
 	record map[string]interface{}, prior interface{},
 ) bool {
 	writeErr := s.writeCeilingSurfaceNote(ctx, taskID, sessionID, deferral)
+	if errors.Is(writeErr, errCeilingSurfaceSuperseded) {
+		return true
+	}
 
 	updated := make(map[string]interface{}, len(record)+1)
 	for key, value := range record {
@@ -112,14 +120,34 @@ func (s *Service) writeCeilingSurfaceNoteAndStamp(
 }
 
 // writeCeilingSurfaceNote is AC-49's carrier: one session-anchored status
-// message row, written through the existing CreateSessionMessage entry point
-// createRecoveryStatusMessage already uses for a different variant. It
-// carries no recovery actions (AC-49d): a ceiling deferral clears itself on
-// its own.
+// message row, written through the idempotent message entry point. The
+// deferral identity is checked while holding task admission, so replacement
+// cannot race the external write with an obsolete card note.
 func (s *Service) writeCeilingSurfaceNote(ctx context.Context, taskID, sessionID string, deferral models.CeilingDeferral) error {
 	if s.messageCreator == nil {
 		return fmt.Errorf("no message creator is configured")
 	}
+	admissionCtx, release := s.lockCeilingEntryAdmission(ctx, taskID)
+	defer release()
+	currentRaw, _, err := s.repo.GetTaskDeferredLaunch(admissionCtx, taskID)
+	if err != nil {
+		return fmt.Errorf("reading deferred launch before surfacing ceiling refusal: %w", err)
+	}
+	current, readErr := models.ReadCeilingDeferral(currentRaw)
+	if readErr != nil {
+		return errCeilingSurfaceSuperseded
+	}
+	identityMatches, compareErr := sameCeilingSurfaceIdentity(current, deferral)
+	if compareErr != nil {
+		return fmt.Errorf("comparing deferred launch before surfacing ceiling refusal: %w", compareErr)
+	}
+	if !identityMatches {
+		return errCeilingSurfaceSuperseded
+	}
+	messageID, err := ceilingSurfaceMessageID(taskID, sessionID, deferral)
+	if err != nil {
+		return fmt.Errorf("building ceiling refusal message identity: %w", err)
+	}
 	metadata := map[string]interface{}{
 		metaKeyVariant:         metaVariantCeiling,
 		ceilingFieldReasonCode: deferral.ReasonCode,
@@ -133,7 +161,45 @@ func (s *Service) writeCeilingSurfaceNote(ctx context.Context, taskID, sessionID
 		content = fmt.Sprintf("This session's launch is queued: %d of %d concurrent sessions are in use.",
 			deferral.Population, deferral.Ceiling)
 	}
-	return s.messageCreator.CreateSessionMessage(
-		ctx, taskID, content, sessionID, string(v1.MessageTypeStatus), "", metadata, false,
+	return s.messageCreator.CreateSessionMessageIdempotent(
+		admissionCtx, messageID, taskID, content, sessionID, string(v1.MessageTypeStatus), "", metadata, false,
 	)
 }
+
+func sameCeilingSurfaceIdentity(a, b models.CeilingDeferral) (bool, error) {
+	if a.ReasonCode != b.ReasonCode || a.Population != b.Population ||
+		a.PopulationKnown != b.PopulationKnown || a.Ceiling != b.Ceiling {
+		return false, nil
+	}
+	return sameCeilingDeferralIdentity(a, b)
+}
+
+func ceilingSurfaceMessageID(taskID, sessionID string, deferral models.CeilingDeferral) (string, error) {
+	identity, err := json.Marshal(struct {
+		TaskID          string
+		SessionID       string
+		Kind            models.CeilingLaunchKind
+		Payload         map[string]interface{}
+		Origin          string
+		ReasonCode      string
+		QueuedAt        string
+		Population      int
+		PopulationKnown bool
+		Ceiling         int
+	}{
+		TaskID:          taskID,
+		SessionID:       sessionID,
+		Kind:            deferral.Kind,
+		Payload:         deferral.Payload,
+		Origin:          deferral.Origin,
+		ReasonCode:      deferral.ReasonCode,
+		QueuedAt:        deferral.QueuedAt.UTC().Format(time.RFC3339),
+		Population:      deferral.Population,
+		PopulationKnown: deferral.PopulationKnown,
+		Ceiling:         deferral.Ceiling,
+	})
+	if err != nil {
+		return "", err
+	}
+	return uuid.NewSHA1(uuid.NameSpaceOID, append([]byte("ceiling-surface:"), identity...)).String(), nil
+}

apps/backend/internal/orchestrator/ceiling_surface_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 
@@ -72,6 +73,30 @@ func TestDeferCeilingRefusal_SameReasonDoesNotRewriteTheNote(t *testing.T) {
 	require.Len(t, messages.sessionMessages, 1, "an unchanged reason must not write a second note")
 }
 
+func TestWriteCeilingSurfaceNote_IsIdentityGuardedAndIdempotent(t *testing.T) {
+	svc, repo := newServiceWithRealRepo(t)
+	ctx := context.Background()
+	seedTaskAndSession(t, repo, "surface-identity-task", "surface-identity-session", models.TaskSessionStateRunning)
+	messages := &mockMessageCreator{}
+	svc.messageCreator = messages
+
+	old := models.CeilingDeferral{
+		Kind: models.CeilingLaunchResume, Origin: string(launchOriginAutomatic),
+		ReasonCode: ceilingReasonRefused, QueuedAt: time.Now().UTC().Truncate(time.Second), Ceiling: 3,
+		Payload: map[string]interface{}{metaKeySessionID: "surface-identity-session"},
+	}
+	newer := old
+	newer.ReasonCode = ceilingReasonUnknownPopulation
+	require.NoError(t, repo.SetTaskMetadataKey(ctx, "surface-identity-task", models.MetaKeyDeferredLaunch, models.CeilingRecordKeys(newer)))
+	require.ErrorIs(t, svc.writeCeilingSurfaceNote(ctx, "surface-identity-task", "surface-identity-session", old), errCeilingSurfaceSuperseded)
+	require.Empty(t, messages.sessionMessages)
+
+	require.NoError(t, repo.SetTaskMetadataKey(ctx, "surface-identity-task", models.MetaKeyDeferredLaunch, models.CeilingRecordKeys(old)))
+	require.NoError(t, svc.writeCeilingSurfaceNote(ctx, "surface-identity-task", "surface-identity-session", old))
+	require.NoError(t, svc.writeCeilingSurfaceNote(ctx, "surface-identity-task", "surface-identity-session", old))
+	require.Len(t, messages.sessionMessages, 1)
+}
+
 // TestDeferCeilingRefusal_ReasonChangeRewritesTheRecordAndTheNote pins AC-49f:
 // the same pending launch refused again for a different reason updates the
 // stored reason code and gets its own note.

apps/backend/internal/orchestrator/dynamic_launch.go

@@ -52,6 +52,11 @@ func (d *dynamicTaskDownstream) Launch(
 	options.AgentProfileID = launch.ExecutionProfileID
 	options.Prompt = launch.Prompt
 	options.PriorACPSession = launch.PriorACPSession
+	if d.task != nil {
+		if err := d.service.admitCeilingDispatch(ctx, d.task.ID); err != nil {
+			return dynamicruntime.DownstreamExecution{}, err
+		}
+	}
 	d.service.beginDynamicAttempt(d.sessionID)
 	taskID := ""
 	if d.task != nil {
@@ -478,7 +483,7 @@ func (s *Service) launchConcretePreparedSession(
 	options executor.LaunchOptions,
 ) (*executor.TaskExecution, error) {
 	if task != nil {
-		if err := s.validateContextCeilingEntry(ctx, task.ID); err != nil {
+		if err := s.admitCeilingDispatch(ctx, task.ID); err != nil {
 			return nil, err
 		}
 	}

apps/backend/internal/orchestrator/queue_send_now.go

@@ -615,6 +615,7 @@ func (s *Service) promptSendNowClaim(ctx context.Context, claim *messagequeue.Se
 			return false, fmt.Errorf("%w: deferred launch changed before Send Now", ErrCeilingLaunchSuperseded)
 		}
 		ceilingClaim.deferral = boundDeferral
+		ctx = withCeilingDispatchClaim(ctx, ceilingClaim)
 		if disposition, detail, validationErr := s.validateCeilingEntry(ctx, currentTask, ceilingClaim.deferral); validationErr != nil {
 			return false, validationErr
 		} else if disposition != ceilingEntryValid {
@@ -714,10 +715,10 @@ func (s *Service) promptSendNowClaim(ctx context.Context, claim *messagequeue.Se
 
 	_, err = s.promptTask(ctx, claim.Dispatch.TaskID, sessionID, promptContent, claim.Dispatch.Model,
 		claim.Dispatch.PlanMode, attachments, false, launchOriginManual, promptTaskOptions{
-			claimEntryID:         claim.Dispatch.ID,
-			afterClaim:           s.sendNowAfterClaim(ctx, claim, attachments, durablePlanComments),
-			beforeDispatch:       s.sendNowDeliveryBoundary(ctx, claim, durablePlanComments, &deliveryAttempted),
-			disableDispatchRetry: durablePlanComments,
+			claimEntryID:           claim.Dispatch.ID,
+			afterClaim:             s.sendNowAfterClaim(ctx, claim, attachments, durablePlanComments),
+			afterDispatchAdmission: s.sendNowDeliveryBoundary(ctx, claim, durablePlanComments, &deliveryAttempted),
+			disableDispatchRetry:   durablePlanComments,
 			afterDispatch: func() error {
 				return s.markSendNowClaimAcceptedWithRetry(ctx, claim)
 			},

apps/backend/internal/orchestrator/task_operations.go

@@ -3744,7 +3744,7 @@ func (s *Service) attemptColdResume(
 	if executor.IsCancellableResumeContext(ctx) {
 		resumeCtx = ctx
 	}
-	if err := s.validateContextCeilingEntry(resumeCtx, session.TaskID); err != nil {
+	if err := s.admitCeilingDispatch(resumeCtx, session.TaskID); err != nil {
 		return false, err
 	}
 	dispatchCtx, releaseCeilingDispatch, err := s.commitCeilingEntryDispatch(
@@ -5605,12 +5605,15 @@ type promptTaskOptions struct {
 	lifecyclePrompt bool
 	afterClaim      func() error
 	afterDispatch   func() error
-	// beforeDispatch is the durable at-most-once boundary for caller-owned
-	// queue receipts. It runs once, immediately before the first provider or
-	// model-switch I/O that can carry the prompt.
-	beforeDispatch        func() error
-	disableDispatchRetry  bool
-	preservePromptContext bool
+	// beforeDispatch runs once before the final dispatch admission boundary.
+	beforeDispatch func() error
+	// afterDispatchAdmission runs once after final dispatch admission succeeds
+	// and before the first provider or model-switch I/O that can carry the
+	// prompt. Queue receipts use this boundary because admission can reject a
+	// stale claim.
+	afterDispatchAdmission func() error
+	disableDispatchRetry   bool
+	preservePromptContext  bool
 	// reserveTurnUntilDispatch persists detached-resume ownership before agentctl
 	// dispatch, while delaying the visible turn.started event until acceptance.
 	reserveTurnUntilDispatch  bool
@@ -5811,6 +5814,7 @@ func (s *Service) promptTask(ctx context.Context, taskID, sessionID string, prom
 		}
 	}
 	runBeforeDispatch := runBeforeDispatchOnce(options.beforeDispatch)
+	runAfterDispatchAdmission := runBeforeDispatchOnce(options.afterDispatchAdmission)
 
 	// Keep the replay payload only after this provider attempt is accepted.
 	// Admission and dispatch failures must not retain prompt attachments until
@@ -5841,6 +5845,7 @@ func (s *Service) promptTask(ctx context.Context, taskID, sessionID string, prom
 
 	switchResult, switchHandled, modelSwitchGuard, switchErr := s.resolveModelSwitchAttempt(
 		resumePromptCtx, taskID, sessionID, model, effectivePrompt, session, foregroundDispatch, runBeforeDispatch,
+		runAfterDispatchAdmission,
 		resumeAttempt, modelSwitchGuard,
 	)
 	if switchHandled {
@@ -5866,6 +5871,7 @@ func (s *Service) promptTask(ctx context.Context, taskID, sessionID string, prom
 	return s.runPromptTurn(
 		resumePromptCtx, taskID, sessionID, prompt, planMode, resumedForPrompt, attachments, dispatchOnly,
 		effectivePrompt, session, rollback, options, foregroundDispatch, runBeforeDispatch,
+		runAfterDispatchAdmission,
 		releaseDispatchGuard, resumeAttempt,
 	)
 }
@@ -5885,12 +5891,13 @@ func (s *Service) runPromptTurn(
 	options promptTaskOptions,
 	foregroundDispatch *foregroundDispatch,
 	runBeforeDispatch func() error,
+	runAfterDispatchAdmission func() error,
 	releaseDispatchGuard func(),
 	resumeAttempt *resumeAttempt,
 ) (*PromptResult, error) {
 	promptCtx, session, earlyResult, err := s.validateAndRunDispatchBoundary(
 		ctx, taskID, sessionID, prompt, planMode, resumedForPrompt, attachments,
-		session, rollback, options, foregroundDispatch, runBeforeDispatch, releaseDispatchGuard,
+		session, rollback, options, foregroundDispatch, runBeforeDispatch, runAfterDispatchAdmission, releaseDispatchGuard,
 	)
 	if err != nil || earlyResult != nil {
 		return earlyResult, err
@@ -5997,11 +6004,13 @@ func (s *Service) resolveModelSwitchAttempt(
 	session *models.TaskSession,
 	foregroundDispatch *foregroundDispatch,
 	runBeforeDispatch func() error,
+	runAfterDispatchAdmission func() error,
 	resumeAttempt *resumeAttempt,
 	modelSwitchGuard *lockedCancelInFlightGuard,
 ) (switchResult *PromptResult, handled bool, remainingGuard *lockedCancelInFlightGuard, switchErr error) {
 	result, handledSwitch, attemptErr := s.attemptModelSwitchForPrompt(
 		ctx, taskID, sessionID, model, effectivePrompt, session, foregroundDispatch, runBeforeDispatch,
+		runAfterDispatchAdmission,
 	)
 	if !handledSwitch {
 		if modelSwitchGuard != nil {
@@ -6221,8 +6230,8 @@ func (s *Service) finishPromptExecutorDispatch(
 }
 
 // validateAndRunDispatchBoundary resolves promptTask's queued-dispatch
-// identity check and the caller's beforeDispatch hook — the two checks that
-// must both pass immediately before the executor call — rolling back the
+// identity check, pre-admission hook, and post-admission hook — the checks
+// that must all pass immediately before the executor call — rolling back the
 // foreground dispatch and prompt claim on either failure. Only bounded-ack
 // callers preserve request context past this point; ordinary prompts can take
 // minutes, so the executor context is resolved here and threaded back out.
@@ -6230,6 +6239,7 @@ func (s *Service) validateAndRunDispatchBoundary(
 	ctx context.Context, taskID, sessionID, prompt string, planMode, resumedForPrompt bool,
 	attachments []v1.MessageAttachment, session *models.TaskSession, rollback promptClaimRollback,
 	options promptTaskOptions, foregroundDispatch *foregroundDispatch, runBeforeDispatch func() error,
+	runAfterDispatchAdmission func() error,
 	releaseDispatchGuard func(),
 ) (promptCtx context.Context, resolvedSession *models.TaskSession, earlyResult *PromptResult, err error) {
 	promptCtx = options.executorContext(ctx)
@@ -6259,7 +6269,14 @@ func (s *Service) validateAndRunDispatchBoundary(
 		s.rollbackPromptClaim(failureCtx, taskID, sessionID, rollback)
 		return promptCtx, nil, nil, bindingErr
 	}
-	if boundaryErr := runBeforeDispatch(); boundaryErr != nil {
+	boundaryErr := runBeforeDispatch()
+	if boundaryErr == nil {
+		boundaryErr = s.admitCeilingDispatch(promptCtx, taskID)
+	}
+	if boundaryErr == nil {
+		boundaryErr = runAfterDispatchAdmission()
+	}
+	if boundaryErr != nil {
 		if releaseDispatchGuard != nil {
 			releaseDispatchGuard()
 		}
@@ -6747,8 +6764,9 @@ func (s *Service) promptDispatchCallbackForIdentity(
 }
 
 // attemptModelSwitchForPrompt runs promptTask's pre-dispatch model-switch
-// handling: arming the initial-prompt-attempt evidence and the caller's
-// beforeDispatch hook when a switch is required, then delegating to
+// handling: arming the initial-prompt-attempt evidence, running the caller's
+// pre-admission hook, admitting the dispatch, and running the caller's
+// post-admission hook when a switch is required, then delegating to
 // trySwitchModelForPrompt. handled reports whether the caller already has its
 // full response — a dispatched switch, a switch failure, or a beforeDispatch
 // failure — in which case promptTask returns (result, err) directly without
@@ -6756,13 +6774,21 @@ func (s *Service) promptDispatchCallbackForIdentity(
 func (s *Service) attemptModelSwitchForPrompt(
 	ctx context.Context, taskID, sessionID, model, effectivePrompt string, session *models.TaskSession,
 	foregroundDispatch *foregroundDispatch, runBeforeDispatch func() error,
+	runAfterDispatchAdmission func() error,
 ) (result *PromptResult, handled bool, err error) {
 	if modelSwitchRequired(session, model) {
 		s.beginInitialPromptAttempt(sessionID, s.isDynamicPromptSession(session))
-		if err := runBeforeDispatch(); err != nil {
+		admissionErr := runBeforeDispatch()
+		if admissionErr == nil {
+			admissionErr = s.admitCeilingDispatch(ctx, taskID)
+		}
+		if admissionErr == nil {
+			admissionErr = runAfterDispatchAdmission()
+		}
+		if admissionErr != nil {
 			s.rollbackForegroundDispatchOnFailure(ctx, taskID, sessionID, foregroundDispatch)
 			s.clearPromptAttemptEvidence(sessionID, "", 0)
-			return nil, true, err
+			return nil, true, admissionErr
 		}
 	}
 	result, handled, switchErr := s.trySwitchModelForPrompt(ctx, taskID, sessionID, model, effectivePrompt, session, foregroundDispatch)

apps/backend/internal/orchestrator/task_operations_test.go

@@ -4869,6 +4869,7 @@ func (m *mockMessageCreator) CreateSessionMessage(_ context.Context, taskID, con
 func (m *mockMessageCreator) CreateSessionMessageIdempotent(_ context.Context, messageID, taskID, content, sessionID, messageType, turnID string, metadata map[string]interface{}, requestsInput bool) error {
 	m.mu.Lock()
 	defer m.mu.Unlock()
+	m.sessionMessageAttempts++
 	if m.sessionMessageErr != nil {
 		return m.sessionMessageErr
 	}
@@ -4888,6 +4889,9 @@ func (m *mockMessageCreator) CreateSessionMessageIdempotent(_ context.Context, m
 		metadata:      metadata,
 		requestsInput: requestsInput,
 	})
+	if m.sessionMessageDone != nil {
+		m.sessionMessageOnce.Do(func() { close(m.sessionMessageDone) })
+	}
 	return nil
 }
 

apps/backend/internal/orchestrator/workflow_session_target.go

@@ -238,38 +238,44 @@ func (s *Service) promoteWorkflowSessionRoute(
 	destination *models.TaskSession,
 	route *models.WorkflowSessionRoute,
 ) (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
 	}
-	destinationParkingStamp := s.captureWorkflowParkingStamp(ctx, destination.ID)
-	if promoter, ok := s.repo.(workflowSessionRoutePromoter); ok && route != nil {
+	// The task admission lock covers the primary/route mutation only. Parking
+	// cleanup and task.updated publication may re-enter runtime paths and must
+	// happen after the durable ownership boundary is released.
+	admissionCtx, release := s.lockCeilingEntryAdmission(ctx, taskID)
+	destinationParkingStamp := ""
+	promoted := false
+	var promoteErr error
+	func() {
+		defer release()
+		if err := s.workflowRouteMutationAllowed(admissionCtx, taskID); err != nil {
+			promoteErr = err
+			return
+		}
+		destinationParkingStamp = s.captureWorkflowParkingStamp(admissionCtx, destination.ID)
+		if promoter, ok := s.repo.(workflowSessionRoutePromoter); ok && route != nil {
+			committed := *route
+			committed.DestinationID = destination.ID
+			committed.Phase = workflowSessionRouteCommitted
+			promoted, promoteErr = promoter.SetSessionPrimaryWithWorkflowSessionRouteIfNonterminal(admissionCtx, destination.ID, committed)
+			return
+		}
+		promoted, promoteErr = s.setNonterminalSessionPrimary(admissionCtx, destination.ID)
+		if promoteErr != nil || !promoted || route == nil {
+			return
+		}
 		committed := *route
 		committed.DestinationID = destination.ID
 		committed.Phase = workflowSessionRouteCommitted
-		promoted, err := promoter.SetSessionPrimaryWithWorkflowSessionRouteIfNonterminal(ctx, destination.ID, committed)
-		if err != nil || !promoted {
-			return promoted, err
-		}
-		s.clearWorkflowParkingForExplicitExecution(ctx, destination.ID, destinationParkingStamp)
-		s.publishPrimarySessionUpdate(ctx, taskID, destination.ID)
-		return true, nil
-	}
-	promoted, err := s.setNonterminalSessionPrimary(ctx, destination.ID)
-	if err != nil || !promoted {
-		return promoted, err
-	}
-	if route != nil {
-		committed := *route
-		committed.DestinationID = destination.ID
-		committed.Phase = workflowSessionRouteCommitted
-		if err := s.persistWorkflowSessionRoute(ctx, taskID, committed); err != nil {
-			return false, err
+		if err := s.persistWorkflowSessionRoute(admissionCtx, taskID, committed); err != nil {
+			promoted = false
+			promoteErr = err
 		}
+	}()
+	if promoteErr != nil || !promoted {
+		return promoted, promoteErr
 	}
 	// Promotion is the workflow's explicit selection of this destination,
 	// whether the compatibility path has a route object or only a primary