From f4c9131307400a86f5291032f7a107d043196f62 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20Flor=C3=AAncio?= Date: Thu, 17 Sep 2026 23:46:57 +0100 Subject: [PATCH] fix: prevent ceiling replay lock cycles (#3779) --- .../internal/orchestrator/ceiling_claim.go | 102 +++-- .../internal/orchestrator/ceiling_defer.go | 195 ++++---- .../ceiling_dispatch_admission.go | 87 ++++ .../ceiling_dispatch_admission_test.go | 217 +++++++++ .../internal/orchestrator/ceiling_entry.go | 16 +- .../internal/orchestrator/ceiling_replay.go | 9 + .../ceiling_replay_lock_order_test.go | 418 ++++++++++++++++++ .../ceiling_replay_progress_test.go | 254 +++++++++++ .../internal/orchestrator/ceiling_seam3.go | 1 + .../internal/orchestrator/ceiling_surface.go | 78 +++- .../orchestrator/ceiling_surface_test.go | 25 ++ .../internal/orchestrator/dynamic_launch.go | 7 +- .../internal/orchestrator/queue_send_now.go | 9 +- .../internal/orchestrator/task_operations.go | 56 ++- .../orchestrator/task_operations_test.go | 4 + .../orchestrator/workflow_session_target.go | 56 +-- .../plan.md | 209 +++++++++ .../task-01-remove-replay-lock-cycle.md | 209 +++++++++ docs/plans/queued-session-ownership/plan.md | 6 + .../task-02-preserve-queued-entry.md | 4 + .../workflow-cancelled-turn-completion.md | 9 +- .../system-design/queued-session-ownership.md | 73 +++ 22 files changed, 1867 insertions(+), 177 deletions(-) create mode 100644 apps/backend/internal/orchestrator/ceiling_dispatch_admission.go create mode 100644 apps/backend/internal/orchestrator/ceiling_dispatch_admission_test.go create mode 100644 apps/backend/internal/orchestrator/ceiling_replay_lock_order_test.go create mode 100644 apps/backend/internal/orchestrator/ceiling_replay_progress_test.go create mode 100644 docs/plans/ceiling-replay-cancellation-deadlock/plan.md create mode 100644 docs/plans/ceiling-replay-cancellation-deadlock/task-01-remove-replay-lock-cycle.md diff --git a/apps/backend/internal/orchestrator/ceiling_claim.go b/apps/backend/internal/orchestrator/ceiling_claim.go index 170492ac30..03fc23edf1 100644 --- a/apps/backend/internal/orchestrator/ceiling_claim.go +++ b/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 } diff --git a/apps/backend/internal/orchestrator/ceiling_defer.go b/apps/backend/internal/orchestrator/ceiling_defer.go index 751faf18e3..30db7848c2 100644 --- a/apps/backend/internal/orchestrator/ceiling_defer.go +++ b/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 diff --git a/apps/backend/internal/orchestrator/ceiling_dispatch_admission.go b/apps/backend/internal/orchestrator/ceiling_dispatch_admission.go new file mode 100644 index 0000000000..1273eec42c --- /dev/null +++ b/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") +} diff --git a/apps/backend/internal/orchestrator/ceiling_dispatch_admission_test.go b/apps/backend/internal/orchestrator/ceiling_dispatch_admission_test.go new file mode 100644 index 0000000000..cc0775e479 --- /dev/null +++ b/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) +} diff --git a/apps/backend/internal/orchestrator/ceiling_entry.go b/apps/backend/internal/orchestrator/ceiling_entry.go index d027c14276..30b522bc5d 100644 --- a/apps/backend/internal/orchestrator/ceiling_entry.go +++ b/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) diff --git a/apps/backend/internal/orchestrator/ceiling_replay.go b/apps/backend/internal/orchestrator/ceiling_replay.go index d0eb9686ad..e081941ba8 100644 --- a/apps/backend/internal/orchestrator/ceiling_replay.go +++ b/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 { diff --git a/apps/backend/internal/orchestrator/ceiling_replay_lock_order_test.go b/apps/backend/internal/orchestrator/ceiling_replay_lock_order_test.go new file mode 100644 index 0000000000..18624f2f6a --- /dev/null +++ b/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") + } +} diff --git a/apps/backend/internal/orchestrator/ceiling_replay_progress_test.go b/apps/backend/internal/orchestrator/ceiling_replay_progress_test.go new file mode 100644 index 0000000000..d4e94e3f04 --- /dev/null +++ b/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, ©) }() + 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) + } +} diff --git a/apps/backend/internal/orchestrator/ceiling_seam3.go b/apps/backend/internal/orchestrator/ceiling_seam3.go index 6dbdde6dc5..1a25545642 100644 --- a/apps/backend/internal/orchestrator/ceiling_seam3.go +++ b/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 } diff --git a/apps/backend/internal/orchestrator/ceiling_surface.go b/apps/backend/internal/orchestrator/ceiling_surface.go index e6f68dd551..8b4e07459c 100644 --- a/apps/backend/internal/orchestrator/ceiling_surface.go +++ b/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 +} diff --git a/apps/backend/internal/orchestrator/ceiling_surface_test.go b/apps/backend/internal/orchestrator/ceiling_surface_test.go index 5dc6d1c783..1f1b662036 100644 --- a/apps/backend/internal/orchestrator/ceiling_surface_test.go +++ b/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. diff --git a/apps/backend/internal/orchestrator/dynamic_launch.go b/apps/backend/internal/orchestrator/dynamic_launch.go index 05bfac60b2..979f33bffe 100644 --- a/apps/backend/internal/orchestrator/dynamic_launch.go +++ b/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 } } diff --git a/apps/backend/internal/orchestrator/queue_send_now.go b/apps/backend/internal/orchestrator/queue_send_now.go index 248f1fc177..671dfb79cd 100644 --- a/apps/backend/internal/orchestrator/queue_send_now.go +++ b/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) }, diff --git a/apps/backend/internal/orchestrator/task_operations.go b/apps/backend/internal/orchestrator/task_operations.go index baed5f3a3e..4b7564e196 100644 --- a/apps/backend/internal/orchestrator/task_operations.go +++ b/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) diff --git a/apps/backend/internal/orchestrator/task_operations_test.go b/apps/backend/internal/orchestrator/task_operations_test.go index 0e3c35a9fa..9793e6045b 100644 --- a/apps/backend/internal/orchestrator/task_operations_test.go +++ b/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 } diff --git a/apps/backend/internal/orchestrator/workflow_session_target.go b/apps/backend/internal/orchestrator/workflow_session_target.go index d82791f763..af9425a8cd 100644 --- a/apps/backend/internal/orchestrator/workflow_session_target.go +++ b/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 diff --git a/docs/plans/ceiling-replay-cancellation-deadlock/plan.md b/docs/plans/ceiling-replay-cancellation-deadlock/plan.md new file mode 100644 index 0000000000..1aec61bfc9 --- /dev/null +++ b/docs/plans/ceiling-replay-cancellation-deadlock/plan.md @@ -0,0 +1,209 @@ +--- +created: 2026-09-17 +status: complete +requirements: + - REQ-TASKS-WORKFLOW-CANCELLED-TURN-COMPLETION-001 + - REQ-TASKS-QUEUED-SESSION-OWNERSHIP-002 + - REQ-AGENTS-SESSION-CEILING-001 +system_design: + - ../../specs/tasks/system-design/queued-session-ownership.md + - ../../specs/agents/system-design/session-concurrency-ceiling.md +legacy_specs: [] +--- + +# Implementation plan: Ceiling replay and cancellation deadlock + +## Overview + +Remove the lock cycle that blocked deferred launches and task-state +reconciliation. The implementation narrows replay, claim, deferral, and route +critical sections, orders review reconciliation consistently, and adds +deterministic regression coverage with focused desktop/mobile verification. + +The task system owns the correction because task entry ownership and runtime +state reconciliation form the failed boundary. Agent ceiling counting remains +owned by the [agent requirement](../../specs/agents/requirements/session-concurrency-ceiling.md). + +## Evidence and root cause + +Read-only investigation on 2026-09-17 used the retained backend log, SQLite +database, and the live `/debug/pprof/goroutine?debug=2` endpoint on port 38429. +Process discovery returned no instance, but the live endpoint was accessible. +The database recorded build `v0.94.0-212-g57bbae10ad`. + +Affected task: `dffcacff-2515-4e02-ad27-d7ed20b001d6`. +Session: `383db1ce-2150-43f0-8d70-b03d3ce7ecae`. +ACP session: `01a0afa0-01e8-7b70-a02f-7e316282df8e`. + +At 19:27:08 Lisbon time, provider cancellation acknowledgement and lifecycle +escalation occurred in the same millisecond. The session became +`WAITING_FOR_INPUT`, and its active turn closed. The task remained `IN_PROGRESS`. +Both task sessions were waiting in the database. The log also recorded disabled +cancellation-driven workflow completion. That setting does not explain the +missing runtime `REVIEW` write. + +The live stacks established these blocked paths: + +| Path | Held boundary | Waiting boundary | +| --- | --- | --- | +| Boot-ready callback through `writeTaskReviewState` | Global `taskRuntimeStateMu` | Task admission channel in `ceiling_entry_lock.go:32` | +| Ceiling sweep through `replayCeilingDeferral` and `claimSessionRunningForPrompt` | Task admission lock across replay | Session cancellation guard | +| Stream activity through `setSessionRunningForExecution` | Session cancellation guard | Global `taskRuntimeStateMu` | +| Explicit cancellation through `reconcileCancelledTaskReview` | Its session cancellation guard | Global `taskRuntimeStateMu` | + +Source establishes lock ownership around those stack frames. The replay and +boot-ready stacks showed approximately 31 minutes of blocking. Multiple unrelated +state writers also waited on the same global mutex. The incident task is a +victim of the shared lock cycle, not evidence of a Luna-specific defect. + +`ListAdmittedSessionIDs` counts `STARTING` and `RUNNING`, not task `IN_PROGRESS`. +The diagnostic database query found two counted sessions at that later instant. +Process-local reservations were not included in that query. The queue worker +was blocked regardless of the apparent free capacity. Changing a task row alone +cannot release these locks. + +Evidence pointers are `/root/.kandev/logs/backend-logs.log` around lines +14564–14600 and `/root/.kandev/data/kandev.db`. These are mutable local sources, +not portable test inputs. No transcripts, raw ACP frames, or full stack dump are +copied into this package. Reproduce with isolated fixtures, not production data. + +## Requirement conformance + +- [Cancellation](../../specs/tasks/requirements/workflow-cancelled-turn-completion.md): + AC-TASKS-WORKFLOW-CANCELLED-TURN-COMPLETION-001.1 already requires settled + cancellation and review-ready reconciliation. Criterion 001.2 extracts that + outcome explicitly, including the existing sibling and queued-destination rules. +- [Queued ownership](../../specs/tasks/requirements/queued-session-ownership.md): + AC-TASKS-QUEUED-SESSION-OWNERSHIP-002.1–6 define state, exact delivery, retry, + restart, and successor protection. Their existing draft status remains unchanged. +- AC-AGENTS-SESSION-CEILING-001.3, .5, and .6 define counting, replay, and + reservation identity. This repair does not change capacity policy. + +The [design correction](../../specs/tasks/system-design/queued-session-ownership.md#replay-and-reconciliation-locking) +defines short critical sections and dispatch revalidation. No new requirement +identity, database schema, or ADR is necessary. Existing claims and lifecycle +ownership provide the repair mechanism. + +## Scope + +### In scope + +- Remove task admission ownership across runtime dispatch and session-lock waits. +- Order task admission before the global task-state mutex. +- Preserve final entry validation, exact claim settlement, and event publication. +- Prove cancellation settlement, unrelated task progress, and continued queue retry. +- Preserve working-sibling, queued-destination, archive, Office, and terminal rules. + +### Out of scope + +- Live restart, task data repair, task creation, or deployment. +- Changing ceiling limits, WIP occupancy, manual overrides, or queue ordering. +- Changing cancellation workflow policy, timeout values, or provider adapters. +- New worker pools, durable leases, schema migrations, UI markup, or copy. +- Repairing unrelated existing documentation or historical verification gaps. + +## Technical approach + +`ceiling_replay.go:replayCeilingDeferral` currently encloses concrete launch +seams in `lockCeilingEntryAdmission`. Narrow that scope to local validation. +Use the claim already acquired by `retryOneDeferredCeilingLaunch` for dispatch +ownership. Do not add another lease or discard the entry binding. + +Audit all seven replay kinds through `validateContextCeilingEntry` and +`validateClaimedCeilingBinding`. Preserve atomic ordering between route mutation +and final local admission. Use existing session guards before task admission +for that short boundary. Do not hold task admission during provider acceptance +waits, startup waits, or synchronous callbacks. Revalidate after preparation +through `admitCeilingDispatch`, renewing the exact replay claim when present. + +In `event_handlers_streaming.go:writeTaskReviewState`, acquire task admission +before `taskRuntimeStateMu`. Reload task, sessions, and queue identity within +the protected section. Preserve conditional writes and normal service events. +Audit every task-admission holder for reverse lock acquisition and reentrant +publication, including route promotion, deferral publication, and Send Now. + +Scope the context ownership marker to the held section. Runtime callbacks use +their own context and must acquire their own guards. Passing a marker across +goroutines would conceal the deadlock by weakening serialization. + +## Tests + +Add `apps/backend/internal/orchestrator/ceiling_replay_lock_order_test.go`. +Use the real task repository and public service paths with a controllable runtime. +Use channels to establish lock ownership, not sleeps or repeated random races. + +| Planned test | Evidence | Criteria | +| --- | --- | --- | +| `TestCeilingReplayBootReadyAndCancellationProgress` | Deferred prompt replay, boot-ready, and stream activity settle. An unrelated explicit cancel reaches REVIEW and returns. A second deferred task dispatches once. | cancellation 001.2; queued ownership 002.2–3; ceiling 001.5 | +| `TestCeilingReplayRevalidatesAfterRuntimePreparation` | Route replacement before admission blocks the old prompt. Its claim cannot clear a successor. Cover every replay kind through a table. | queued ownership 002.5–6; ceiling 001.6 | +| `TestCeilingReplayPreservesReconciliationPrecedence` | Any working sibling preserves active state. With no working sibling, a valid deferred destination preserves Scheduling. Otherwise cancellation reaches Review. | cancellation 001.2; queued ownership 002.1–2 | +| `TestCeilingReplayClaimAndShutdownProgress` | Concurrent replay/Send Now cannot duplicate delivery. Cancelled attempt cleanup releases only its claim. The sweeper stops after released fixture barriers. | queued ownership 002.3–6; ceiling 001.3, .5–6 | + +The first test is the mandatory behavioral RED. Bound its failure and cleanup +so the pre-fix deadlock does not strand the test runner. A subprocess harness is +appropriate when blocked production mutexes cannot be released by fixture cleanup. +Keep all helper entry points private and explicit to avoid recursive test binaries. + +Existing `ceiling_replay_test.go`, `route_action_lock_order_test.go`, and +`task_operations_test.go` supply claim, barrier, and cancellation patterns. +Existing claim tests cover expired claims and detached cleanup contexts. + +## E2E tests + +Run existing queued-session and workflow-cancellation specs on Chromium and +mobile Chrome. These verify visible settlement, exact queued delivery, and +unchanged cancellation policy through the shipped UI. The new Go service test +owns deterministic reproduction of the lock cycle. Browser timing is not the +RED gate for this defect. No rendered UI change or ASCII preview is needed. + +## Work orders + +- [x] [Task 01: Remove the replay lock cycle](task-01-remove-replay-lock-cycle.md) + +## Related delivery records + +The [queued-session package](../queued-session-ownership/plan.md) and its Task 02 +own the original implementation. This package owns the newly observed lock +regression. Existing PostgreSQL gaps and original work-order statuses remain +unchanged. Their prior passing tests do not prove this concurrency intersection. + +The [session ceiling](../session-concurrency-ceiling/plan.md) and +[cancelled-turn completion](../cancelled-turn-completion/plan.md) packages remain +compatibility inputs. Do not reopen their implementation or alter their results. + +## Verification results + +The [review remediation record](task-01-remove-replay-lock-cycle.md#review-remediation) +supersedes the initial callback-only evidence with the full cancellation and +sweep-progress regression. It also records final dispatch-claim fencing and the +seven-kind admission matrix. + +Design and implementation validation on 2026-09-17: + +- `python3 scripts/list-docs.py validate`: passed (287 decisions, 1002 specifications). +- `python3 scripts/lint-spec-files.test.py`: passed (36 tests). +- `python3 scripts/lint-spec-files.py --all`: passed. +- Catalog discovery found the affected requirement and system design. +- `git diff --check`: passed. +- Package status includes both new files and the companion links. +- The mandatory RED test failed behaviorally before the production change by + timing out while a synchronous dispatch callback reconciled task state. +- The four new replay lock-order tests passed in a focused run. +- The four new replay lock-order tests passed under the race detector for 10 + repetitions. +- Final dispatch-admission coverage passed for all seven replay kinds, including + successor-claim rejection and route mutation serialization. +- The targeted orchestrator race selection passed. +- The targeted SQLite repository race selection passed. +- Chromium passed 3 tests from the queued-session and cancellation specs. +- Mobile Chrome passed 2 tests from the queued-session and cancellation specs. +- No live instance state changed. The package remains uncommitted. + +## Risks + +- Removing the outer lock without final admission fencing can dispatch stale work. +- A released context can falsely claim lock ownership in nested dispatch helpers. +- Synchronous event subscribers can re-enter a lock held by their publisher. +- Fixing only the global mutex order leaves the replay/session-guard cycle intact. +- Tests that mock away boot-ready or task publication can miss the production cycle. +- Restart clears in-memory locks but is not a permanent fix or authorized delivery step. diff --git a/docs/plans/ceiling-replay-cancellation-deadlock/task-01-remove-replay-lock-cycle.md b/docs/plans/ceiling-replay-cancellation-deadlock/task-01-remove-replay-lock-cycle.md new file mode 100644 index 0000000000..ea707fe558 --- /dev/null +++ b/docs/plans/ceiling-replay-cancellation-deadlock/task-01-remove-replay-lock-cycle.md @@ -0,0 +1,209 @@ +--- +id: "01-remove-replay-lock-cycle" +title: "Remove the replay lock cycle" +status: done +wave: 1 +depends_on: [] +plan: "plan.md" +requirements: + - REQ-TASKS-WORKFLOW-CANCELLED-TURN-COMPLETION-001 + - REQ-TASKS-QUEUED-SESSION-OWNERSHIP-002 + - REQ-AGENTS-SESSION-CEILING-001 +acceptance_criteria: + - AC-TASKS-WORKFLOW-CANCELLED-TURN-COMPLETION-001.2 + - AC-TASKS-QUEUED-SESSION-OWNERSHIP-002.1 + - AC-TASKS-QUEUED-SESSION-OWNERSHIP-002.2 + - AC-TASKS-QUEUED-SESSION-OWNERSHIP-002.3 + - AC-TASKS-QUEUED-SESSION-OWNERSHIP-002.4 + - AC-TASKS-QUEUED-SESSION-OWNERSHIP-002.5 + - AC-TASKS-QUEUED-SESSION-OWNERSHIP-002.6 + - AC-AGENTS-SESSION-CEILING-001.3 + - AC-AGENTS-SESSION-CEILING-001.5 + - AC-AGENTS-SESSION-CEILING-001.6 +system_design: + - ../../specs/tasks/system-design/queued-session-ownership.md + - ../../specs/agents/system-design/session-concurrency-ceiling.md +--- + +# Task 01: Remove the replay lock cycle + +## Summary + +Make replay, callbacks, and cancellation complete without a lock cycle. +Preserve exact workflow entry ownership and task-state precedence through short, +ordered critical sections and existing conditional claims. + +## In scope + +- Write the four regressions in the [plan test matrix](plan.md#tests). +- Prove `TestCeilingReplayBootReadyAndCancellationProgress` fails before production changes. +- Narrow replay admission locking and preserve final dispatch validation for all seven kinds. +- Order task admission before the global state mutex in review reconciliation. +- Audit route, deferral, claim, Send Now, and publication paths for reverse acquisition. +- Keep lock markers local to held sections and immutable bindings across dispatch. +- Run targeted race tests and existing desktop/mobile end-to-end scenarios. +- Record exact results and update only this package's implementation status. + +## Out of scope + +No live-instance writes, restarts, new scheduling policy, schema, provider change, +UI markup, or new cancellation setting. Do not replace the existing claim system. + +## Acceptance + +1. Barrier-controlled replay, boot-ready, and stream activity complete. An unrelated + cancel persists input-ready session state, closes its turn, publishes Review, + clears pending cancellation, and returns. The sweep then dispatches another + eligible task once. Run the callback case with synchronous MemoryEventBus + delivery and an asynchronous delivery fixture. +2. Any working sibling preserves active state. With no working sibling, a valid + deferred destination preserves Scheduling. Otherwise the eligible task reaches + Review. Include idle/failed siblings, archive, Office, and terminal outcomes. +3. Route replacement, deleted destination, successor claim, capacity refusal, + concurrent Send Now, and attempt cancellation preserve exact ownership. + Unaccepted work remains retryable. Existing restart and stale-clear tests pass. + +## Verification + +Run from the repository root. First add the regression and run only its RED +command. Record a behavioral failure, not a compilation or fixture error. + +```bash +(cd apps/backend && go test ./internal/orchestrator -run '^TestCeilingReplayBootReadyAndCancellationProgress$' -count=1 -timeout=60s) +``` + +After correction, run these commands sequentially. Install pnpm dependencies +from `apps/` first only if this checkout lacks them. + +```bash +(cd apps/backend && go test -race ./internal/orchestrator -run '^TestCeiling(Replay(BootReadyAndCancellationProgress|RevalidatesAfterRuntimePreparation|PreservesReconciliationPrecedence|ClaimAndShutdownProgress|RejectsSuccessorClaimAfterPreparation|SynchronousReconciliationCallback)|PromptRevalidatesAfterDispatchReceipt|ModelSwitchRevalidatesAfterDispatchReceipt|DispatchAdmissionSerializesRouteMutation)$' -count=10 -timeout=180s) +(cd apps/backend && go test -race ./internal/orchestrator -run 'Test.*Ceiling|Test.*ReviewState|Test.*BootReady|TestCancelAgent|Test.*QueuedEntry|Test.*QueuedSession|Test.*SendNow|Test.*DoNotDeadlock' -count=1 -timeout=300s) +(cd apps/backend && go test -race ./internal/task/repository/sqlite -run 'Test.*DeferredLaunch|Test.*AdmittedSession|TestUpdateTaskStateIfCurrentIn' -count=1 -timeout=120s) +(cd apps/web && pnpm e2e:run --project chromium tests/workflow/queued-session-ownership.spec.ts tests/workflow/workflow-cancel-completion.spec.ts) +(cd apps/web && pnpm e2e:run --project mobile-chrome tests/workflow/mobile-queued-session-ownership.spec.ts tests/workflow/mobile-workflow-cancel-completion.spec.ts) +python3 scripts/list-docs.py validate +python3 scripts/lint-spec-files.py --all +git diff --check +``` + +The managed E2E runner builds fresh artifacts and owns teardown. Do not overlap +desktop and phone runs. Record discovered test counts and actual results. +No PostgreSQL mutation is planned. If implementation changes repository behavior, +extend this work order with the required isolated PostgreSQL coverage before delivery. + +## Files likely touched + +- `apps/backend/internal/orchestrator/ceiling_replay.go` +- `apps/backend/internal/orchestrator/ceiling_entry_lock.go` +- `apps/backend/internal/orchestrator/ceiling_entry.go` +- `apps/backend/internal/orchestrator/ceiling_claim.go` +- `apps/backend/internal/orchestrator/ceiling_dispatch_admission.go` +- `apps/backend/internal/orchestrator/ceiling_defer.go` +- `apps/backend/internal/orchestrator/event_handlers_streaming.go` +- `apps/backend/internal/orchestrator/task_operations.go` +- `apps/backend/internal/orchestrator/dynamic_launch.go` +- `apps/backend/internal/orchestrator/queue_send_now.go` +- `apps/backend/internal/orchestrator/workflow_session_target.go` +- New `apps/backend/internal/orchestrator/ceiling_replay_lock_order_test.go` +- New `apps/backend/internal/orchestrator/ceiling_replay_progress_test.go` +- New `apps/backend/internal/orchestrator/ceiling_dispatch_admission_test.go` +- Existing `ceiling_replay_test.go` and `task_operations_test.go` only where needed. + +The work order owns these changes. Additional affected launch seams remain +within scope only when necessary to preserve final entry validation. Put new +test helpers in a separate file if existing files exceed lint limits. + +## Dependencies + +None. The original queued-session implementation is already present in the base. +This work order does not depend on completing its historical PostgreSQL checks. + +## Risks + +Moving a lock can expose stale route dispatch. Cover the race after preparation, +not only before the first read. A test deadline alone cannot clean up a deadlocked +goroutine. Use releasable barriers or an isolated subprocess for pre-fix RED. + +## Parallelism + +`sequential` + +## Inputs + +- [Design: replay and reconciliation locking](../../specs/tasks/system-design/queued-session-ownership.md#replay-and-reconciliation-locking). +- [Cancellation requirements](../../specs/tasks/requirements/workflow-cancelled-turn-completion.md). +- `ceiling_replay_test.go`: stale route, successor record, claim, and retry fixtures. +- `route_action_lock_order_test.go`: deterministic lock-order regression patterns. +- `task_operations_test.go`: explicit cancel reconciliation and terminal precedence. +- `apps/backend/AGENTS.md` and `.agents/skills/tdd/references/backend-tests.md`. + +## Initial implementation results + +Implemented the replay lock-cycle correction and added +`ceiling_replay_lock_order_test.go` and the dispatch-admission regression +coverage. The final dispatch boundary renews the exact claim after preparation +for every replay kind and rejects successor claim or route changes. + +The mandatory RED run failed behaviorally before the production change: replay +timed out while a synchronous dispatch callback attempted task-state +reconciliation. After the correction, the focused four-test run passed, as did +the repeated race run, the targeted orchestrator and SQLite race selections, +and the required Chromium and mobile Chrome workflow specs. The exact results +were: + +- Focused replay tests: 4 passed. +- Repeated replay race tests: 4 tests x 10 repetitions passed. +- Dispatch-admission tests: all seven replay kinds passed, including route + mutation serialization and successor-claim rejection. +- Targeted orchestrator race selection: passed. +- Targeted SQLite repository race selection: passed. +- Chromium workflow specs: 3 passed. +- Mobile Chrome workflow specs: 2 passed. + +No live instance state changed. The package remains uncommitted. + +## Review remediation + +The local review found two gaps: final dispatch lacked an atomic claim boundary, +and the callback regression did not reproduce cancellation or sweep progress. + +The replacement progress test uses the real watcher, task service, SQLite +repository, stream handler, boot-ready handler, cancellation path, and sweep. +Barriers establish the stream guard, replay admission lock, and blocked boot-ready +handler. The test verifies persisted Review, a closed turn, cleared cancellation, +event publication, and exactly one dispatch for the second queued task. +Both synchronous MemoryEventBus delivery and asynchronous delivery use this test. +An isolated child process bounds failure without leaking blocked production +goroutines into the parent test process. + +A Go overlay restored the two pre-fix lock-order files from `HEAD` without +changing the worktree. The replacement progress test failed in both delivery +modes. The synchronous run reported `no progress: stream completed`. The +asynchronous run timed out with the original lock cycle and blocked cancellation. + +Additional regressions failed before correction for successor claim replacement, +late route replacement, and model-switch admission. Final admission now validates +the route and renews the exact existing claim under one task-admission lock. +The seven-kind admission matrix verifies that route mutation waits for this +renewal. Runtime calls occur after lock release. Cleanup preserves successor +claims even when their route also changes. + +The repeated race command passed all nine selected tests for ten repetitions +in 107.603 seconds. The SQLite race selection passed in 6.819 seconds. +The local sandbox required `GOCACHE=/tmp/kandev-review-go-cache` for Go commands. + +The additional scoped lint command passed with zero issues: + +```bash +(cd apps/backend && GOCACHE=/tmp/kandev-review-go-cache GOLANGCI_LINT_CACHE=/tmp/kandev-review-lint-cache golangci-lint run ./internal/orchestrator/... --new-from-rev=HEAD --timeout=5m) +``` + +The broader orchestrator race selection passed in 30.183 seconds. This run +included the final regression files after the test-only lint correction. +The fresh Chromium build passed three tests in 34.0 seconds. The fresh mobile +Chrome build passed two tests in 31.8 seconds. Both builds used the managed +runner and isolated Docker fixtures. + +The specification catalog, specification lint, and `git diff --check` passed. +The temporary pre-fix overlay files were removed. No live instance changed. +All source and documentation changes remain unstaged and uncommitted. diff --git a/docs/plans/queued-session-ownership/plan.md b/docs/plans/queued-session-ownership/plan.md index 2bb1c173d8..4e1567f00c 100644 --- a/docs/plans/queued-session-ownership/plan.md +++ b/docs/plans/queued-session-ownership/plan.md @@ -216,6 +216,12 @@ subagents are authorized by this plan. ## Related delivery records +The [replay and cancellation deadlock repair](../ceiling-replay-cancellation-deadlock/plan.md) +owns the concurrency regression observed on 2026-09-17. It narrows Task 02's +lock scope while preserving its entry and state contracts. Its regression +matrix is complete. Historical results and outstanding PostgreSQL checks in +this package remain unchanged. + [Workflow lifecycle](../workflow-profile-session-reuse/plan.md), [explicit targeting](../workflow-session-targeting/plan.md), [same-profile fresh sessions](../workflow-same-profile-new-session/plan.md), and diff --git a/docs/plans/queued-session-ownership/task-02-preserve-queued-entry.md b/docs/plans/queued-session-ownership/task-02-preserve-queued-entry.md index 89da4edbe5..b3b9ef33f8 100644 --- a/docs/plans/queued-session-ownership/task-02-preserve-queued-entry.md +++ b/docs/plans/queued-session-ownership/task-02-preserve-queued-entry.md @@ -108,6 +108,10 @@ Read existing `ceiling_replay_test.go`, `ceiling_defer_test.go`, ## Results +Follow-up: the [replay deadlock work order](../ceiling-replay-cancellation-deadlock/task-01-remove-replay-lock-cycle.md) +owns the later runtime deadlock. The results below do not cover that lock cycle. +Its correction must preserve this task's queued-entry and stale-writer guarantees. + Implementation is present. Deferred entries retain the exact workflow route, step, destination session, prompt, and queue time. Review-state reconciliation, replay clearing, and Send Now admission use the destination identity and diff --git a/docs/specs/tasks/requirements/workflow-cancelled-turn-completion.md b/docs/specs/tasks/requirements/workflow-cancelled-turn-completion.md index 7c2929bb12..2d46642602 100644 --- a/docs/specs/tasks/requirements/workflow-cancelled-turn-completion.md +++ b/docs/specs/tasks/requirements/workflow-cancelled-turn-completion.md @@ -20,6 +20,13 @@ This document is the migrated task-system source for the capability. The source #### Acceptance criteria - **AC-TASKS-WORKFLOW-CANCELLED-TURN-COMPLETION-001.1:** When a consumer uses this capability, the system shall provide the observable behavior and exclusions documented below. +- **AC-TASKS-WORKFLOW-CANCELLED-TURN-COMPLETION-001.2:** When explicit cancellation settles, an eligible task shall leave its active runtime state even if another task starts concurrently. If no session owns active work and no valid deferred destination remains, the task shall become `REVIEW`. A valid deferred destination preserves `SCHEDULING`. A working sibling, terminal transition, archive, or Office ownership retains its existing precedence. Disabling cancellation-driven workflow completion shall not disable this runtime reconciliation. + +Criterion 001.2 makes the existing review-ready outcome explicit. It does not +change workflow movement or the cancellation setting. + +The [task reconciliation design](../system-design/queued-session-ownership.md#replay-and-reconciliation-locking) +defines concurrency at this boundary. ## Migrated source detail @@ -140,4 +147,4 @@ Template defaults apply when a workflow's steps are instantiated. Changing the e - Changing queued-message drain policy after cancellation. - Treating agent errors, provider failures, crashes, task stops, or clarification pauses as successful completion. - Adding cancellation-cause metadata to workflow transition history. -- Reducing the existing lifecycle timeout for agents that acknowledge cancellation but never finish their prompt, or synthesizing success before their terminal path settles. \ No newline at end of file +- Reducing the existing lifecycle timeout for agents that acknowledge cancellation but never finish their prompt, or synthesizing success before their terminal path settles. diff --git a/docs/specs/tasks/system-design/queued-session-ownership.md b/docs/specs/tasks/system-design/queued-session-ownership.md index 096e103f5e..22025ff781 100644 --- a/docs/specs/tasks/system-design/queued-session-ownership.md +++ b/docs/specs/tasks/system-design/queued-session-ownership.md @@ -5,6 +5,7 @@ requirements: - REQ-TASKS-QUEUED-SESSION-OWNERSHIP-001 - REQ-TASKS-QUEUED-SESSION-OWNERSHIP-002 - REQ-TASKS-QUEUED-SESSION-OWNERSHIP-003 + - REQ-TASKS-WORKFLOW-CANCELLED-TURN-COMPLETION-001 --- # Queued session ownership system design @@ -23,6 +24,7 @@ and task-owned deferred record. It does not replace either queue or change WIP a | REQ-TASKS-QUEUED-SESSION-OWNERSHIP-001 | Inspection intent; Durable workflow parking | | REQ-TASKS-QUEUED-SESSION-OWNERSHIP-002 | Deferred entry ownership; Task reconciliation | | REQ-TASKS-QUEUED-SESSION-OWNERSHIP-003 | Queue projection; Desktop and mobile surfaces; Failure and observability | +| REQ-TASKS-WORKFLOW-CANCELLED-TURN-COMPLETION-001 | Replay and reconciliation locking | ## Existing components @@ -155,6 +157,77 @@ The sweep also repairs legacy REVIEW+valid-deferral tasks to SCHEDULING when no session is working. This repair requires authoritative task and entry evidence; the browser never writes task state based on a badge. +## Replay and reconciliation locking + +This correction is implemented by the +[replay deadlock fix package](../../../plans/ceiling-replay-cancellation-deadlock/plan.md). +The implementation and regression results are recorded in that completed +package. + +`replayCeilingDeferral` must not retain the task admission lock across a launch, +resume, prompt, runtime wait, or callback publication. The existing +`ceilingDeferredLaunchClaim` owns one accepted record while dispatch is in flight. +Its claim ID and deferral identity protect settlement from successor records. +The claim does not replace workflow-entry validation. + +Replay has three boundaries: + +1. Under the task admission lock, read and validate the exact record and its + committed entry. Retain its existing claim and immutable entry binding. +2. Release the task admission lock before the concrete launch seam. Acquire + existing session lifecycle and cancellation guards in their established order. + Revalidate entry and recipient at the final admission boundary. +3. Settle or release only the matching claim through existing conditional writes. + A superseded entry cannot dispatch or clear a successor record. + +Final entry validation and its local dispatch claim must serialize with route +mutation. A read before runtime preparation is insufficient. Inspect every replay +kind and its concrete admission seam, including sessionless starts and Send Now. +After preparation, a changed route fails before prompt admission. A route change +after accepted admission follows the existing lifecycle cancellation contract. +No database transaction spans runtime I/O. + +`admitCeilingDispatch` validates the entry and renews the existing claim through +one task-admission critical section. The renewal retains the claim ID and uses +the existing conditional write and lease duration. A missing, replaced, or +expired claim rejects dispatch. Claim loss leaves the current record available +to its owner instead of dropping it as a superseded route. + +Replay and Send Now carry the claim through preparation in the context. Concrete +launch, resume, prompt, and model-switch boundaries perform final admission. +Prompt admission follows the caller's dispatch-receipt callback. The admission +context never reaches the runtime. Superseded replay cleanup also checks the +claim ID before it removes a deferred record. + +Where local critical sections require multiple locks, the order is session +lifecycle, session cancellation guard, task admission, then `taskRuntimeStateMu`. +Acquire only the locks that the operation needs. Never acquire an earlier lock +while holding a later lock. In particular, `writeTaskReviewState` acquires task +admission before the global runtime-state mutex. Task admission holders must not +wait for session guards, runtime callbacks, or event subscribers. + +Keep entry validation and task-state writes in a short critical section. +Preserve the existing conditional state write and task-event publication contract. +If a helper publishes synchronously, audit its subscribers before retaining a +lock across that call. Move reentrant publication outside the section while +preserving event order when necessary. + +The context marker from `lockCeilingEntryAdmission` represents actual ownership +within that section only. Never pass it to dispatch after release or to another +goroutine. Carry immutable entry identity separately. Do not simulate reentrancy +by attaching the marker to boot-ready callbacks. + +Cancellation retains its separate intent marker, shared operation, and bounded +service-owned context. It releases session serialization during provider waits. +Its final task reconciliation must complete after terminal frames settle. +Cancellation policy controls workflow movement, not whether runtime state settles. + +Use barrier-controlled service tests for replay, boot-ready, stream activity, +cancellation, and a second queued task. Verify both current task states and +published events. Include mixed sibling states and stale route/claim replacement. +The single ceiling sweep remains sequential and must continue after each settled +attempt. No detached replay workers or new queue mechanism are required. + ## Queue projection Add optional `launch_queue` to `TaskStatusSummary` and its frontend type. It is