diff --git a/apps/backend/internal/orchestrator/ceiling_boot_deadlock_test.go b/apps/backend/internal/orchestrator/ceiling_boot_deadlock_test.go new file mode 100644 index 0000000000..72d498c632 --- /dev/null +++ b/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) +} diff --git a/apps/backend/internal/orchestrator/ceiling_entry_lock.go b/apps/backend/internal/orchestrator/ceiling_entry_lock.go index f24643c687..a97ee01bd9 100644 --- a/apps/backend/internal/orchestrator/ceiling_entry_lock.go +++ b/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 +} diff --git a/apps/backend/internal/orchestrator/ceiling_replay.go b/apps/backend/internal/orchestrator/ceiling_replay.go index eb8e314a03..d0eb9686ad 100644 --- a/apps/backend/internal/orchestrator/ceiling_replay.go +++ b/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 { diff --git a/apps/backend/internal/orchestrator/dynamic_launch.go b/apps/backend/internal/orchestrator/dynamic_launch.go index e054ee0452..05bfac60b2 100644 --- a/apps/backend/internal/orchestrator/dynamic_launch.go +++ b/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 } diff --git a/apps/backend/internal/orchestrator/event_handlers_streaming.go b/apps/backend/internal/orchestrator/event_handlers_streaming.go index 02d10a1177..4899b9d265 100644 --- a/apps/backend/internal/orchestrator/event_handlers_streaming.go +++ b/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) { diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow.go b/apps/backend/internal/orchestrator/event_handlers_workflow.go index 33ca756bdb..10bf4df0fd 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow.go +++ b/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) diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index 045739e7d2..7f57c4eeb3 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/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 diff --git a/apps/backend/internal/orchestrator/task_operations.go b/apps/backend/internal/orchestrator/task_operations.go index b5419f7924..baed5f3a3e 100644 --- a/apps/backend/internal/orchestrator/task_operations.go +++ b/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, ) diff --git a/apps/backend/internal/orchestrator/workflow_session_target.go b/apps/backend/internal/orchestrator/workflow_session_target.go index d97af8c618..d82791f763 100644 --- a/apps/backend/internal/orchestrator/workflow_session_target.go +++ b/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 } diff --git a/docs/plans/ceiling-boot-deadlock/plan.md b/docs/plans/ceiling-boot-deadlock/plan.md new file mode 100644 index 0000000000..8955940e02 --- /dev/null +++ b/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. diff --git a/docs/plans/ceiling-boot-deadlock/task-01-break-lock-cycle.md b/docs/plans/ceiling-boot-deadlock/task-01-break-lock-cycle.md new file mode 100644 index 0000000000..2ffc10aec4 --- /dev/null +++ b/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. diff --git a/docs/specs/agents/system-design/session-concurrency-ceiling.md b/docs/specs/agents/system-design/session-concurrency-ceiling.md index eee6750bd3..aa0191fccc 100644 --- a/docs/specs/agents/system-design/session-concurrency-ceiling.md +++ b/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