From b4aa51d1304a7b0b1ac39c8bac0c6c100675f774 Mon Sep 17 00:00:00 2001 From: Henry Date: Mon, 7 Sep 2026 04:29:04 +0800 Subject: [PATCH] fix(backend): stop a manual route retry from deadlocking against a context reset (#3443) Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com> Co-authored-by: Carlos Florencio --- .../internal/backendapp/dynamic_routing.go | 84 ++++++- .../dynamic_routing_cancel_race_test.go | 174 ++++++++++++++ .../backendapp/dynamic_routing_result_test.go | 37 +++ .../internal/orchestrator/dynamic_launch.go | 17 +- .../orchestrator/dynamic_launch_test.go | 58 +++++ .../orchestrator/event_handlers_agent.go | 53 +++-- .../orchestrator/route_action_lock.go | 57 +++++ .../route_action_lock_order_test.go | 219 ++++++++++++++++++ .../route_action_serialization_test.go | 156 +++++++++++++ apps/backend/internal/orchestrator/service.go | 82 ++++++- .../internal/orchestrator/task_operations.go | 6 + .../internal/plugins/webapp_protocol_json.go | 2 +- 12 files changed, 912 insertions(+), 33 deletions(-) create mode 100644 apps/backend/internal/backendapp/dynamic_routing_cancel_race_test.go create mode 100644 apps/backend/internal/backendapp/dynamic_routing_result_test.go create mode 100644 apps/backend/internal/orchestrator/route_action_lock.go create mode 100644 apps/backend/internal/orchestrator/route_action_lock_order_test.go create mode 100644 apps/backend/internal/orchestrator/route_action_serialization_test.go diff --git a/apps/backend/internal/backendapp/dynamic_routing.go b/apps/backend/internal/backendapp/dynamic_routing.go index 84763a5bb4..26f5451c0d 100644 --- a/apps/backend/internal/backendapp/dynamic_routing.go +++ b/apps/backend/internal/backendapp/dynamic_routing.go @@ -47,6 +47,15 @@ func applyDynamicRouteAction( if err != nil { return nil, err } + // expectedState is the session lifecycle state last confirmed under this + // in-memory copy. The selection and error writes below are conditioned on + // it, so a terminal transition (e.g. cancellation) that lands + // concurrently with route selection is detected instead of clobbered by a + // whole-row write built from this stale snapshot. The post-launch + // recovery write is different: the launch attempt itself can mutate + // state, so that write re-reads and gates on the state current at + // recovery time instead (see recoverDynamicRouteAction). + expectedState := session.State if err := repairDynamicRouteAfterLaunchFailure( ctx, session, request.ExpectedGeneration, resolver.MarkRouteRecoveryActionRequired, ); err != nil { @@ -54,11 +63,15 @@ func applyDynamicRouteAction( } decision, err := resolveDynamicRouteAction(ctx, resolver, request, session) if err != nil { - return handleDynamicRouteActionError(ctx, repo, session, err) + return handleDynamicRouteActionError(ctx, repo, session, expectedState, err) } - if err := persistDynamicRouteSelection(ctx, repo, session, decision); err != nil { + changed, err := persistDynamicRouteSelection(ctx, repo, session, expectedState, decision) + if err != nil { return nil, err } + if !changed { + return reloadRouteActionResult(ctx, repo, session.ID) + } if request.Action == orchestrator.RouteActionCancelWait || request.Action == orchestrator.RouteActionStop { return routeActionResult(ctx, repo, session), nil } @@ -113,6 +126,7 @@ func handleDynamicRouteActionError( ctx context.Context, repo *sqliterepo.Repository, session *models.TaskSession, + expectedState models.TaskSessionState, err error, ) (*orchestrator.RouteActionResult, error) { var noCandidate *dynamicruntime.NoEligibleCandidateError @@ -123,9 +137,13 @@ func handleDynamicRouteActionError( session.RouteState = "waiting" session.RouteReason = "no_eligible_candidate" session.DownstreamACPSessionID = "" - if updateErr := repo.UpdateTaskSession(ctx, session); updateErr != nil { + changed, updateErr := repo.UpdateTaskSessionIfCurrentState(ctx, session, expectedState) + if updateErr != nil { return nil, routeActionPersistenceError(ctx, repo, session, updateErr) } + if !changed { + return reloadRouteActionResult(ctx, repo, session.ID) + } return routeActionResult(ctx, repo, session), nil } @@ -133,8 +151,9 @@ func persistDynamicRouteSelection( ctx context.Context, repo *sqliterepo.Repository, session *models.TaskSession, + expectedState models.TaskSessionState, decision agentruntime.ProfileExecution, -) error { +) (bool, error) { session.ExecutionProfileID = decision.ExecutionProfileID session.RouteGeneration = decision.Generation session.RouteState = decision.Decision.Status @@ -151,10 +170,11 @@ func persistDynamicRouteSelection( // A candidate change never carries a provider-native ACP identity across // profiles. The conductor will populate this after a fresh launch. session.DownstreamACPSessionID = "" - if err := repo.UpdateTaskSession(ctx, session); err != nil { - return routeActionPersistenceError(ctx, repo, session, err) + changed, err := repo.UpdateTaskSessionIfCurrentState(ctx, session, expectedState) + if err != nil { + return false, routeActionPersistenceError(ctx, repo, session, err) } - return nil + return changed, nil } func finishDynamicRouteAction( @@ -182,6 +202,16 @@ func finishDynamicRouteAction( return routeActionResult(ctx, repo, session), nil } +// recoverDynamicRouteAction records why a successor launch failed after a +// route action was accepted. The launcher itself may mutate session state +// before failing (a real launch failure resets the row to CREATED ahead of +// the launch attempt), so the write here is gated on a fresh reload rather +// than on any state captured before launchSuccessor ran: a stale pre-launch +// snapshot would never match the launcher's own mutation and this recovery +// write would be silently dropped. A reloaded state that is already terminal +// (CANCELLED, COMPLETED, or FAILED) means a concurrent handler has already +// settled the session, and that must win over resurrecting it into +// WAITING_FOR_INPUT. func recoverDynamicRouteAction( ctx context.Context, repo *sqliterepo.Repository, @@ -200,14 +230,48 @@ func recoverDynamicRouteAction( if session.RouteGeneration != expectedGeneration { return routeActionResult(ctx, repo, session), nil } + if isTerminalRouteActionSessionState(session.State) { + return routeActionResult(ctx, repo, session), nil + } + reloadedState := session.State session.RouteState = "action_required" session.RouteReason = orchestrator.RouteActionLaunchFailedReason session.State = models.TaskSessionStateWaitingForInput session.ErrorMessage = launchErr.Error() session.DownstreamACPSessionID = "" - if err := repo.UpdateTaskSession(ctx, session); err != nil { + changed, err := repo.UpdateTaskSessionIfCurrentState(ctx, session, reloadedState) + if err != nil { return nil, routeActionPersistenceError(ctx, repo, session, err) } + if !changed { + return reloadRouteActionResult(ctx, repo, sessionID) + } + return routeActionResult(ctx, repo, session), nil +} + +// isTerminalRouteActionSessionState reports whether a reloaded session state +// means a concurrent handler has already settled the session, mirroring +// orchestrator's unexported isTerminalSessionState (not reachable from this +// package). +func isTerminalRouteActionSessionState(state models.TaskSessionState) bool { + return state == models.TaskSessionStateCancelled || + state == models.TaskSessionStateCompleted || + state == models.TaskSessionStateFailed +} + +// reloadRouteActionResult reloads the session that a guarded write refused to +// overwrite (because its state moved since expectedState was captured) and +// reports that superseded/terminal state instead of the write the caller +// intended. +func reloadRouteActionResult( + ctx context.Context, + repo *sqliterepo.Repository, + sessionID string, +) (*orchestrator.RouteActionResult, error) { + session, err := repo.GetTaskSession(ctx, sessionID) + if err != nil { + return nil, err + } return routeActionResult(ctx, repo, session), nil } @@ -249,7 +313,11 @@ func routeActionResult( } if repo != nil { if state, err := repo.LoadRouteState(ctx, session.ID); err == nil && state != nil { + result.LogicalProfileID = state.LogicalProfileID + result.ExecutionProfileID = state.ExecutionProfileID + result.RouteGeneration = state.Generation result.ProfileVersion = state.ProfileVersion + result.State = state.Status var policyState dynamicruntime.PolicyState if jsonErr := json.Unmarshal([]byte(state.PolicyStateJSON), &policyState); jsonErr == nil { result.ErrorCode = string(policyState.FailureCode) diff --git a/apps/backend/internal/backendapp/dynamic_routing_cancel_race_test.go b/apps/backend/internal/backendapp/dynamic_routing_cancel_race_test.go new file mode 100644 index 0000000000..6f21c18431 --- /dev/null +++ b/apps/backend/internal/backendapp/dynamic_routing_cancel_race_test.go @@ -0,0 +1,174 @@ +package backendapp + +import ( + "context" + "errors" + "path/filepath" + "testing" + "time" + + "github.com/jmoiron/sqlx" + "github.com/stretchr/testify/require" + + "github.com/kandev/kandev/internal/db" + "github.com/kandev/kandev/internal/task/models" + taskrepo "github.com/kandev/kandev/internal/task/repository" + sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" +) + +// newDynamicRoutingCancelRaceRepo opens an in-memory sqlite repo and seeds the +// minimal workspace/workflow/task/session rows a dynamic route action reads. +func newDynamicRoutingCancelRaceRepo(t *testing.T) *sqliterepo.Repository { + t.Helper() + dbConn, err := db.OpenSQLite(filepath.Join(t.TempDir(), "kandev.db")) + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + sqlxDB := sqlx.NewDb(dbConn, "sqlite3") + t.Cleanup(func() { _ = sqlxDB.Close() }) + repo, cleanup, err := taskrepo.Provide(sqlxDB, sqlxDB, nil) + if err != nil { + t.Fatalf("repo provide: %v", err) + } + t.Cleanup(func() { _ = cleanup() }) + return repo +} + +func seedDynamicRoutingCancelRaceSession(t *testing.T, repo *sqliterepo.Repository, sessionID string, state models.TaskSessionState) { + t.Helper() + ctx := context.Background() + if err := repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-1", Name: "WS"}); err != nil { + t.Fatalf("CreateWorkspace: %v", err) + } + if err := repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf-1", WorkspaceID: "ws-1", Name: "WF"}); err != nil { + t.Fatalf("CreateWorkflow: %v", err) + } + if err := repo.CreateTask(ctx, &models.Task{ + ID: "task-1", WorkspaceID: "ws-1", WorkflowID: "wf-1", Title: "T", Priority: "medium", + }); err != nil { + t.Fatalf("CreateTask: %v", err) + } + now := time.Now().UTC() + if err := repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: sessionID, TaskID: "task-1", State: state, StartedAt: now, UpdatedAt: now, + }); err != nil { + t.Fatalf("CreateTaskSession: %v", err) + } +} + +// TestRecoverDynamicRouteActionDoesNotResurrectCancelledSession proves the +// concrete race F4 describes: a successor launch fails after a concurrent +// cancellation has already committed CANCELLED to the row. Before the fix, +// recoverDynamicRouteAction reloaded the row, unconditionally overwrote State +// with WAITING_FOR_INPUT, and whole-row-wrote it back — resurrecting a +// session a concurrent cancel had just terminated. The fix refuses to +// resurrect a reloaded CANCELLED state at all, so a cancellation landing +// during the launch attempt wins. +func TestRecoverDynamicRouteActionDoesNotResurrectCancelledSession(t *testing.T) { + ctx := context.Background() + repo := newDynamicRoutingCancelRaceRepo(t) + seedDynamicRoutingCancelRaceSession(t, repo, "session-1", models.TaskSessionStateRunning) + + // Simulate a concurrent cancellation committing while the successor + // launch (which recoverDynamicRouteAction is about to react to) was still + // in flight. + cancelled, err := repo.GetTaskSession(ctx, "session-1") + require.NoError(t, err) + cancelled.State = models.TaskSessionStateCancelled + require.NoError(t, repo.UpdateTaskSession(ctx, cancelled)) + + result, err := recoverDynamicRouteAction(ctx, repo, nil, "session-1", 0, errors.New("launch failed")) + require.NoError(t, err) + require.Equal(t, "session-1", result.SessionID) + + persisted, err := repo.GetTaskSession(ctx, "session-1") + require.NoError(t, err) + require.Equal(t, models.TaskSessionStateCancelled, persisted.State, + "a concurrently cancelled session must not be resurrected by a superseded route-action recovery write") +} + +// TestRecoverDynamicRouteActionDoesNotResurrectOtherTerminalStates extends the +// CANCELLED case above to COMPLETED and FAILED: any terminal state a +// concurrent handler already committed while the successor launch was in +// flight must win over recoverDynamicRouteAction's WAITING_FOR_INPUT write, +// not just CANCELLED. +func TestRecoverDynamicRouteActionDoesNotResurrectOtherTerminalStates(t *testing.T) { + for _, state := range []models.TaskSessionState{ + models.TaskSessionStateCompleted, + models.TaskSessionStateFailed, + } { + t.Run(string(state), func(t *testing.T) { + ctx := context.Background() + repo := newDynamicRoutingCancelRaceRepo(t) + seedDynamicRoutingCancelRaceSession(t, repo, "session-1", models.TaskSessionStateRunning) + + settled, err := repo.GetTaskSession(ctx, "session-1") + require.NoError(t, err) + settled.State = state + require.NoError(t, repo.UpdateTaskSession(ctx, settled)) + + result, err := recoverDynamicRouteAction(ctx, repo, nil, "session-1", 0, errors.New("launch failed")) + require.NoError(t, err) + require.Equal(t, "session-1", result.SessionID) + + persisted, err := repo.GetTaskSession(ctx, "session-1") + require.NoError(t, err) + require.Equal(t, state, persisted.State, + "a concurrently settled terminal session must not be resurrected by a superseded route-action recovery write") + }) + } +} + +// TestRecoverDynamicRouteActionWritesWhenStateUnchanged is the control case: +// with no concurrent state change, the recovery write must still apply. +func TestRecoverDynamicRouteActionWritesWhenStateUnchanged(t *testing.T) { + ctx := context.Background() + repo := newDynamicRoutingCancelRaceRepo(t) + seedDynamicRoutingCancelRaceSession(t, repo, "session-1", models.TaskSessionStateRunning) + + result, err := recoverDynamicRouteAction(ctx, repo, nil, "session-1", 0, errors.New("launch failed")) + require.NoError(t, err) + require.Equal(t, "session-1", result.SessionID) + + persisted, err := repo.GetTaskSession(ctx, "session-1") + require.NoError(t, err) + require.Equal(t, models.TaskSessionStateWaitingForInput, persisted.State) + require.Equal(t, "action_required", persisted.RouteState) +} + +// TestFinishDynamicRouteActionRecoversAfterLauncherMutatesState proves F5: +// the real launch-failure path does not gate the recovery write on a +// pre-launch snapshot. orchestrator.relaunchDynamicTaskAfterFailure resets +// the session row to CREATED before attempting the successor launch, then +// fails; a CAS built from the state captured before that launch attempt +// would never match CREATED and would silently drop the recovery write, +// stranding the session with no error surfaced. finishDynamicRouteAction +// must still land WAITING_FOR_INPUT/action_required with the launch error. +func TestFinishDynamicRouteActionRecoversAfterLauncherMutatesState(t *testing.T) { + ctx := context.Background() + repo := newDynamicRoutingCancelRaceRepo(t) + seedDynamicRoutingCancelRaceSession(t, repo, "session-1", models.TaskSessionStateRunning) + + launchErr := errors.New("successor launch failed") + launcher := func(ctx context.Context, sessionID string) error { + // Mirrors internal/orchestrator/dynamic_launch.go's + // relaunchDynamicTaskAfterFailure, which resets state to CREATED + // before attempting (and, here, failing) the successor launch. + if err := repo.UpdateTaskSessionState(ctx, sessionID, models.TaskSessionStateCreated, ""); err != nil { + return err + } + return launchErr + } + + result, err := finishDynamicRouteAction(ctx, repo, nil, "session-1", 0, launcher) + require.NoError(t, err) + require.Equal(t, "session-1", result.SessionID) + require.Equal(t, "action_required", result.State) + + persisted, err := repo.GetTaskSession(ctx, "session-1") + require.NoError(t, err) + require.Equal(t, models.TaskSessionStateWaitingForInput, persisted.State) + require.Equal(t, "action_required", persisted.RouteState) + require.Equal(t, "route_action_launch_failed", persisted.RouteReason) + require.Equal(t, launchErr.Error(), persisted.ErrorMessage) +} diff --git a/apps/backend/internal/backendapp/dynamic_routing_result_test.go b/apps/backend/internal/backendapp/dynamic_routing_result_test.go new file mode 100644 index 0000000000..a67158a5da --- /dev/null +++ b/apps/backend/internal/backendapp/dynamic_routing_result_test.go @@ -0,0 +1,37 @@ +package backendapp + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" + + dynamicruntime "github.com/kandev/kandev/internal/agent/runtime/dynamic" + "github.com/kandev/kandev/internal/task/models" +) + +func TestRouteActionResultUsesAuthoritativeRouteState(t *testing.T) { + ctx := context.Background() + repo := newDynamicRoutingCancelRaceRepo(t) + seedDynamicRoutingCancelRaceSession(t, repo, "session-1", models.TaskSessionStateWaitingForInput) + + require.NoError(t, repo.SaveRouteState(ctx, dynamicruntime.RouteState{ + SessionID: "session-1", + LogicalProfileID: "logical-authoritative", + ExecutionProfileID: "execution-authoritative", + Generation: 7, + ProfileVersion: 4, + Status: "retrying", + PolicyStateJSON: `{"retry_ordinal":2}`, + })) + + result, err := repo.GetTaskSession(ctx, "session-1") + require.NoError(t, err) + got := routeActionResult(ctx, repo, result) + + require.Equal(t, "logical-authoritative", got.LogicalProfileID) + require.Equal(t, "execution-authoritative", got.ExecutionProfileID) + require.Equal(t, int64(7), got.RouteGeneration) + require.Equal(t, int64(4), got.ProfileVersion) + require.Equal(t, "retrying", got.State) +} diff --git a/apps/backend/internal/orchestrator/dynamic_launch.go b/apps/backend/internal/orchestrator/dynamic_launch.go index c1acc3db60..25a02838a7 100644 --- a/apps/backend/internal/orchestrator/dynamic_launch.go +++ b/apps/backend/internal/orchestrator/dynamic_launch.go @@ -1055,7 +1055,22 @@ func (s *Service) relaunchDynamicTaskAfterFailure( return false } } - if err := s.repo.UpdateTaskSessionState(ctx, data.SessionID, models.TaskSessionStateCreated, ""); err != nil { + // Reload immediately before the reset: StopExecution is I/O and a + // coordinator stop can commit a terminal state while it runs. A stale + // pre-stop snapshot would let this write resurrect a session the user + // already cancelled, so refuse on a reloaded terminal state and CAS the + // write on that same reload to catch a cancellation landing after it. + preResetState, err := s.repo.GetTaskSession(ctx, data.SessionID) + if err != nil || preResetState == nil { + return false + } + if isTerminalSessionState(preResetState.State) { + return false + } + changed, _, err := s.repo.UpdateTaskSessionStateIfCurrent( + ctx, data.SessionID, preResetState.State, models.TaskSessionStateCreated, "", + ) + if err != nil || !changed { return false } s.completeTurnForSession(ctx, data.SessionID) diff --git a/apps/backend/internal/orchestrator/dynamic_launch_test.go b/apps/backend/internal/orchestrator/dynamic_launch_test.go index 2f862ce266..78aac97d93 100644 --- a/apps/backend/internal/orchestrator/dynamic_launch_test.go +++ b/apps/backend/internal/orchestrator/dynamic_launch_test.go @@ -67,6 +67,64 @@ func TestRelaunchDynamicTaskAfterFailure_DoesNotLaunchSuccessorWhenStopFails(t * } } +// TestRelaunchDynamicTaskAfterFailure_DoesNotResurrectCancelledSession proves +// F-A: a coordinator stop that commits CANCELLED while the predecessor +// teardown is in flight must win over the relaunch's own CREATED reset. +// Before the fix, the reset at dynamic_launch.go was an unconditional +// UpdateTaskSessionState write with no expected-state predicate, so it +// silently overwrote CANCELLED with CREATED and StartCreatedSession then +// launched a new agent on the session the user just stopped. This exercises +// relaunchDynamicTaskAfterFailure directly because LaunchDynamicRouteAction +// (the manual retry path) calls it without the detached-launch wrapper's +// shouldDropSessionFailure guard. +func TestRelaunchDynamicTaskAfterFailure_DoesNotResurrectCancelledSession(t *testing.T) { + ctx := context.Background() + const ( + taskID = "task-dynamic-cancel-race" + sessionID = "session-dynamic-cancel-race" + executionID = "execution-dynamic-cancel-race" + ) + + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateRunning) + seedExecutorRunning(t, repo, sessionID, taskID, executionID) + taskRepo := newMockTaskRepo() + seedMockTaskState(taskRepo, taskID, v1.TaskStateInProgress) + agentManager := &mockAgentManager{ + stopAgentWithReasonFunc: func(context.Context, string, string, bool) error { + // Simulate a coordinator stop committing CANCELLED between the + // route-selection CAS and this relaunch's own destructive reset. + return repo.UpdateTaskSessionState(ctx, sessionID, models.TaskSessionStateCancelled, "") + }, + } + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentManager) + svc.lastTurnPrompt.Store(sessionID, capturedPrompt{text: "retry the task"}) + + relaunched := svc.relaunchDynamicTaskAfterFailure( + ctx, + watcher.AgentEventData{ + TaskID: taskID, + SessionID: sessionID, + AgentExecutionID: executionID, + }, + "fallback-profile", + ) + + if relaunched { + t.Fatal("relaunchDynamicTaskAfterFailure returned success over a concurrently cancelled session") + } + session, err := repo.GetTaskSession(ctx, sessionID) + if err != nil { + t.Fatalf("get session: %v", err) + } + if session.State != models.TaskSessionStateCancelled { + t.Fatalf("session state = %q, want it to stay CANCELLED", session.State) + } + if len(agentManager.startAgentProcessCalls) != 0 { + t.Fatalf("successor launch started %d processes after a concurrent cancel", len(agentManager.startAgentProcessCalls)) + } +} + // The agent.failed event that selects a fallback route is dispatched on the // lifecycle completion goroutine while it holds the execution's prompt // lifecycle lock. Stopping the predecessor needs that lock again, so the diff --git a/apps/backend/internal/orchestrator/event_handlers_agent.go b/apps/backend/internal/orchestrator/event_handlers_agent.go index bf97fcaa2c..59c1c82568 100644 --- a/apps/backend/internal/orchestrator/event_handlers_agent.go +++ b/apps/backend/internal/orchestrator/event_handlers_agent.go @@ -1153,7 +1153,7 @@ func metadataWithoutEntityReferences(metadata map[string]interface{}) map[string // handleAgentCompleted handles agent completion events func (s *Service) handleAgentCompleted(ctx context.Context, data watcher.AgentEventData) { if data.SessionID == "" { - s.handleAgentCompletedLocked(ctx, data) + s.handleAgentCompletedLocked(ctx, data, nil) return } @@ -1174,22 +1174,22 @@ func (s *Service) handleAgentCompleted(ctx context.Context, data watcher.AgentEv // cancel/interrupt decision for this session. If coordinator stop won while // the event waited, the guarded state reload below observes CANCELLED and // suppresses all workflow/on_enter side effects. - lock, release := s.acquireCancelInFlightGuard(data.SessionID) - if !lock.TryLock() { + mutex, release := s.acquireCancelInFlightGuard(data.SessionID) + if !mutex.TryLock() { go func() { - defer release() - lock.Lock() - defer lock.Unlock() - s.handleAgentCompletedAfterGuard(context.WithoutCancel(ctx), data) + mutex.Lock() + guard := &lockedCancelInFlightGuard{mutex: mutex, releaseRef: release, locked: true} + defer guard.release() + s.handleAgentCompletedAfterGuard(context.WithoutCancel(ctx), data, guard) }() return } - defer release() - defer lock.Unlock() - s.handleAgentCompletedAfterGuard(ctx, data) + guard := &lockedCancelInFlightGuard{mutex: mutex, releaseRef: release, locked: true} + defer guard.release() + s.handleAgentCompletedAfterGuard(ctx, data, guard) } -func (s *Service) handleAgentCompletedAfterGuard(ctx context.Context, data watcher.AgentEventData) { +func (s *Service) handleAgentCompletedAfterGuard(ctx context.Context, data watcher.AgentEventData, guard *lockedCancelInFlightGuard) { ctx = withWorkflowProfileSwitchGuardHeld(ctx, data.SessionID, data.AgentExecutionID) if s.isCancelInFlight(data.SessionID) { s.logger.Debug("deferring agent.completed while cancellation is in progress", @@ -1199,10 +1199,19 @@ func (s *Service) handleAgentCompletedAfterGuard(ctx context.Context, data watch return } - s.handleAgentCompletedLocked(ctx, data) + s.handleAgentCompletedLocked(ctx, data, guard) } -func (s *Service) handleAgentCompletedLocked(ctx context.Context, data watcher.AgentEventData) { +// handleAgentCompletedLocked runs with the per-session cancel-in-flight guard +// held, except around calls that themselves acquire the session lifecycle +// lock (reclaimIdleSession, directly or via setSessionWaitingForInputIfRequested): +// those release the guard first and reacquire it after, mirroring +// quiesceActiveResetTurn's yield/reacquire protocol, to keep this lock pair's +// global acquisition order (lifecycle outer, cancel guard inner) intact. guard +// is nil when data.SessionID == "" (handleAgentCompletedLocked is called directly, +// with no cancel guard ever acquired); every unlock/relock is a safe no-op on +// a nil guard. +func (s *Service) handleAgentCompletedLocked(ctx context.Context, data watcher.AgentEventData, guard *lockedCancelInFlightGuard) { s.logger.Info("handling agent completed", zap.String("task_id", data.TaskID), zap.String("session_id", data.SessionID), @@ -1309,7 +1318,22 @@ func (s *Service) handleAgentCompletedLocked(ctx context.Context, data watcher.A } transitioned := s.processOnTurnCompleteViaEngine(ctx, data.TaskID, session) + s.finishAgentCompletedTurn(ctx, data, session, transitioned, guard) +} +// finishAgentCompletedTurn runs the settle steps after +// processOnTurnCompleteViaEngine has decided whether the turn advanced the +// workflow: the WAITING_FOR_INPUT / subtask-terminal-collapse decision, +// execution cleanup, automation finalize, and the reclaimIdleSession settle +// point. guard's unlock/relock calls around the lifecycle-lock-acquiring +// steps follow handleAgentCompletedLocked's doc comment. +func (s *Service) finishAgentCompletedTurn( + ctx context.Context, + data watcher.AgentEventData, + session *models.TaskSession, + transitioned bool, + guard *lockedCancelInFlightGuard, +) { // Agent-exit path: processOnTurnCompleteViaEngine handles normal // on_turn_complete transitions. If it did not transition, ensure the // completed session leaves RUNNING and let setSessionWaitingForInput perform @@ -1330,7 +1354,9 @@ func (s *Service) handleAgentCompletedLocked(ctx context.Context, data watcher.A // itself inspects the task row to pick the path; for sibling // sessions the call is a no-op pass-through to the unconditional // helper, preserving the pre-fix behavior. + guard.unlock() s.setSessionWaitingForInputIfRequested(ctx, data.TaskID, data.SessionID, session) + guard.relock() } // Capture a git status snapshot before cleanup so it can be served @@ -1353,6 +1379,7 @@ func (s *Service) handleAgentCompletedLocked(ctx context.Context, data watcher.A // collapsed to COMPLETED inside setSessionWaitingForInputIfRequested // reclaimed earlier; this call covers sibling/office flows whose // settled shape has no live runtime. + guard.unlock() if err := s.reclaimIdleSession(context.WithoutCancel(ctx), data.SessionID); err != nil { s.logger.Warn("agent.completed settle: reclaim failed; row preserved", zap.String("task_id", data.TaskID), diff --git a/apps/backend/internal/orchestrator/route_action_lock.go b/apps/backend/internal/orchestrator/route_action_lock.go new file mode 100644 index 0000000000..81f4e70622 --- /dev/null +++ b/apps/backend/internal/orchestrator/route_action_lock.go @@ -0,0 +1,57 @@ +package orchestrator + +import "sync" + +// routeActionOperationLock tracks active waiters so its map entry can be +// removed after the last route action releases the per-session mutex. +type routeActionOperationLock struct { + mu sync.Mutex + refs int +} + +// acquireRouteActionOperationLock serializes route actions for one session. +// The returned function must be called exactly once after the handler returns. +func (s *Service) acquireRouteActionOperationLock(sessionID string) func() { + if sessionID == "" { + return func() {} + } + + s.routeActionLocksMu.Lock() + if s.routeActionLocks == nil { + s.routeActionLocks = make(map[string]*routeActionOperationLock) + } + lock, ok := s.routeActionLocks[sessionID] + if !ok { + lock = &routeActionOperationLock{} + s.routeActionLocks[sessionID] = lock + } + lock.refs++ + s.routeActionLocksMu.Unlock() + + lock.mu.Lock() + var released bool + return func() { + if released { + return + } + released = true + lock.mu.Unlock() + s.routeActionLocksMu.Lock() + lock.refs-- + if lock.refs == 0 { + delete(s.routeActionLocks, sessionID) + } + s.routeActionLocksMu.Unlock() + } +} + +func (s *Service) isRouteActionInFlight(sessionID string) bool { + if sessionID == "" { + return false + } + s.routeActionLocksMu.Lock() + lock := s.routeActionLocks[sessionID] + inFlight := lock != nil && lock.refs > 0 + s.routeActionLocksMu.Unlock() + return inFlight +} diff --git a/apps/backend/internal/orchestrator/route_action_lock_order_test.go b/apps/backend/internal/orchestrator/route_action_lock_order_test.go new file mode 100644 index 0000000000..1fd7b1fd6c --- /dev/null +++ b/apps/backend/internal/orchestrator/route_action_lock_order_test.go @@ -0,0 +1,219 @@ +package orchestrator + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/kandev/kandev/internal/task/models" + v1 "github.com/kandev/kandev/pkg/api/v1" +) + +// TestApplyRouteActionAndLifecycleResetDoNotDeadlock proves that +// ApplyRouteAction and a session-lifecycle operation (modeled here the same +// way resetAgentContext acquires its locks) can run concurrently on the same +// session without an ABBA deadlock between acquireCancelInFlightGuard and +// acquireSessionLifecycleLock. +// +// Before the fix, ApplyRouteAction held the cancel-in-flight guard across the +// whole routeActionHandler call, and the handler here reaches +// acquireSessionLifecycleLock (as the real dynamic-route launch path does via +// startCreatedSession) — guard-then-lifecycle. The concurrent goroutine takes +// lifecycle-then-guard, exactly like resetAgentContext. With both orders live +// on the same sessionID, each goroutine can block on the lock the other +// holds, and neither ever completes. +func TestApplyRouteActionAndLifecycleResetDoNotDeadlock(t *testing.T) { + svc, _ := newTurnLifecycleTestService(t) + ctx := context.Background() + sessionID := "session1" + + handlerEntered := make(chan struct{}) + handlerMayLockLifecycle := make(chan struct{}) + svc.SetRouteActionHandler(func(_ context.Context, request RouteActionRequest) (*RouteActionResult, error) { + close(handlerEntered) + <-handlerMayLockLifecycle + release := svc.acquireSessionLifecycleLock(request.SessionID) + defer release() + return &RouteActionResult{SessionID: request.SessionID}, nil + }) + + var wg sync.WaitGroup + routeActionDone := make(chan error, 1) + wg.Add(1) + go func() { + defer wg.Done() + _, err := svc.ApplyRouteAction(ctx, RouteActionRequest{ + SessionID: sessionID, + Action: RouteActionRetry, + }) + routeActionDone <- err + }() + + select { + case <-handlerEntered: + case <-time.After(3 * time.Second): + t.Fatal("route action handler never started") + } + + resetHasLifecycleLock := make(chan struct{}) + resetDone := make(chan struct{}) + wg.Add(1) + go func() { + defer wg.Done() + releaseLifecycleLock := svc.acquireSessionLifecycleLock(sessionID) + defer releaseLifecycleLock() + close(resetHasLifecycleLock) + resetGuard := svc.lockCancelInFlightGuard(sessionID) + defer resetGuard.release() + close(resetDone) + }() + + // Barrier: wait for the reset goroutine to actually hold the lifecycle + // lock before letting the route action handler also request it. A fixed + // sleep here would only assume enough time passed; this proves it. + select { + case <-resetHasLifecycleLock: + case <-time.After(3 * time.Second): + t.Fatal("reset goroutine never acquired the session lifecycle lock") + } + close(handlerMayLockLifecycle) + + waitCh := make(chan struct{}) + go func() { + wg.Wait() + close(waitCh) + }() + + select { + case <-waitCh: + case <-time.After(3 * time.Second): + t.Fatal("deadlock: ApplyRouteAction and lifecycle reset did not both complete within 3s") + } + + select { + case <-resetDone: + default: + t.Fatal("lifecycle reset goroutine did not complete") + } + if err := <-routeActionDone; err != nil { + t.Fatalf("ApplyRouteAction: %v", err) + } +} + +// pausedSecondSessionLoadRepo blocks the *second* GetTaskSession call for a +// target session until released. handleAgentCompleted's first repo call +// (captureSessionCommitsSweep, via resolveArchiveBaseCommitAndBranch) runs +// before the cancel-in-flight guard is acquired; the second is +// handleAgentCompletedLocked's own session load, the first repo access made +// while that guard is held. Pausing there proves the guard is genuinely held +// at the moment of the pause, without relying on a real reachability hook +// production code doesn't have. +type pausedSecondSessionLoadRepo struct { + sessionExecutorStore + sessionID string + calls atomic.Int32 + entered chan struct{} + release chan struct{} + once sync.Once +} + +func (r *pausedSecondSessionLoadRepo) GetTaskSession(ctx context.Context, sessionID string) (*models.TaskSession, error) { + if sessionID == r.sessionID && r.calls.Add(1) == 2 { + r.once.Do(func() { close(r.entered) }) + <-r.release + } + return r.sessionExecutorStore.GetTaskSession(ctx, sessionID) +} + +// TestHandleAgentCompletedAndResetAgentContextDoNotDeadlock exercises the +// real reachability F1a fixed: handleAgentCompletedLocked reaches +// reclaimIdleSession (directly, and via setSessionWaitingForInputIfRequested) +// while it can still be holding the cancel-in-flight guard, and +// reclaimIdleSession's first act is acquireSessionLifecycleLock — the same +// lock resetAgentContext already holds when it (unconditionally) reacquires +// the cancel guard via quiesceActiveResetTurn's resetGuard.relock. Two +// goroutines taking these locks in opposite order on the same session is a +// live ABBA deadlock unless the guard is released before the lifecycle +// lock is requested. +func TestHandleAgentCompletedAndResetAgentContextDoNotDeadlock(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "task1", "session1", "") + + pausedRepo := &pausedSecondSessionLoadRepo{ + sessionExecutorStore: repo, + sessionID: "session1", + entered: make(chan struct{}), + release: make(chan struct{}), + } + + agentMgr := &mockAgentManager{repoForExecutionLookup: repo} + taskRepo := newMockTaskRepo() + seedMockTaskState(taskRepo, "task1", v1.TaskStateInProgress) + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentMgr) + svc.repo = pausedRepo + + handlerDone := make(chan struct{}) + go func() { + defer close(handlerDone) + svc.handleAgentCompleted(ctx, watcherAgentCompletedData("task1", "session1", "exec-1")) + }() + + // Barrier: wait for handleAgentCompleted to reach its own guarded session + // load. At this point it holds the cancel-in-flight guard and has not + // touched the lifecycle lock. + select { + case <-pausedRepo.entered: + case <-time.After(3 * time.Second): + t.Fatal("handleAgentCompleted never reached its guarded session load") + } + + session, err := repo.GetTaskSession(ctx, "session1") + require.NoError(t, err) + + resetDone := make(chan bool, 1) + go func() { + resetDone <- svc.resetAgentContext(ctx, "task1", session, "Successor") + }() + + // Barrier: prove resetAgentContext has genuinely acquired the lifecycle + // lock (and is therefore blocked requesting the guard held above, exactly + // like the real global lock order) before releasing the paused handler. + // A losing TryLock on the same mutex is a positive signal, not a timing + // guess. + require.Eventually(t, func() bool { + value, ok := svc.sessionLifecycleLocks.Load("session1") + if !ok { + return false + } + mu, ok := value.(*sync.Mutex) + if !ok { + return false + } + if mu.TryLock() { + mu.Unlock() + return false + } + return true + }, time.Second, time.Millisecond, "resetAgentContext never acquired the session lifecycle lock") + + close(pausedRepo.release) + + select { + case <-handlerDone: + case <-time.After(3 * time.Second): + t.Fatal("deadlock: handleAgentCompleted did not complete within 3s") + } + select { + case resetOK := <-resetDone: + if !resetOK { + t.Fatal("resetAgentContext returned false") + } + case <-time.After(3 * time.Second): + t.Fatal("deadlock: resetAgentContext did not complete within 3s") + } +} diff --git a/apps/backend/internal/orchestrator/route_action_serialization_test.go b/apps/backend/internal/orchestrator/route_action_serialization_test.go new file mode 100644 index 0000000000..2710b42597 --- /dev/null +++ b/apps/backend/internal/orchestrator/route_action_serialization_test.go @@ -0,0 +1,156 @@ +package orchestrator + +import ( + "context" + "errors" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/kandev/kandev/internal/task/models" +) + +func TestApplyRouteActionSerializesHandlersPerSession(t *testing.T) { + svc := &Service{} + const sessionID = "session1" + firstEntered := make(chan struct{}) + secondEntered := make(chan struct{}) + releaseFirst := make(chan struct{}) + var calls atomic.Int32 + + svc.SetRouteActionHandler(func(_ context.Context, request RouteActionRequest) (*RouteActionResult, error) { + if calls.Add(1) == 1 { + close(firstEntered) + <-releaseFirst + } else { + close(secondEntered) + } + return &RouteActionResult{SessionID: request.SessionID}, nil + }) + + firstDone := make(chan error, 1) + go func() { + _, err := svc.ApplyRouteAction(context.Background(), RouteActionRequest{ + SessionID: sessionID, + Action: RouteActionRetry, + }) + firstDone <- err + }() + + select { + case <-firstEntered: + case <-time.After(time.Second): + t.Fatal("first route action did not enter its handler") + } + + secondDone := make(chan error, 1) + go func() { + _, err := svc.ApplyRouteAction(context.Background(), RouteActionRequest{ + SessionID: sessionID, + Action: RouteActionTryNext, + }) + secondDone <- err + }() + + select { + case <-secondEntered: + t.Fatal("second route action entered before the first completed") + case <-time.After(100 * time.Millisecond): + } + + close(releaseFirst) + for _, done := range []<-chan error{firstDone, secondDone} { + select { + case err := <-done: + if err != nil { + t.Fatalf("ApplyRouteAction: %v", err) + } + case <-time.After(time.Second): + t.Fatal("route action did not complete") + } + } +} + +func TestApplyRouteActionAllowsDifferentSessionsConcurrently(t *testing.T) { + svc := &Service{} + entered := make(chan string, 2) + release := make(chan struct{}) + var calls atomic.Int32 + + svc.SetRouteActionHandler(func(_ context.Context, request RouteActionRequest) (*RouteActionResult, error) { + calls.Add(1) + entered <- request.SessionID + <-release + return &RouteActionResult{SessionID: request.SessionID}, nil + }) + + var wg sync.WaitGroup + for _, sessionID := range []string{"session1", "session2"} { + wg.Add(1) + go func(sessionID string) { + defer wg.Done() + _, _ = svc.ApplyRouteAction(context.Background(), RouteActionRequest{ + SessionID: sessionID, + Action: RouteActionRetry, + }) + }(sessionID) + } + + seen := map[string]bool{} + for range 2 { + select { + case sessionID := <-entered: + seen[sessionID] = true + case <-time.After(time.Second): + t.Fatal("route actions for different sessions did not run concurrently") + } + } + close(release) + wg.Wait() + if calls.Load() != 2 || len(seen) != 2 { + t.Fatalf("calls = %d, sessions = %#v, want one call per session", calls.Load(), seen) + } +} + +func TestRouteActionClaimBlocksPromptAdmission(t *testing.T) { + svc, repo := newTurnLifecycleTestService(t) + ctx := context.Background() + const sessionID = "session1" + if err := repo.UpdateTaskSessionState(ctx, sessionID, models.TaskSessionStateWaitingForInput, ""); err != nil { + t.Fatalf("set session promptable: %v", err) + } + + handlerEntered := make(chan struct{}) + releaseHandler := make(chan struct{}) + svc.SetRouteActionHandler(func(_ context.Context, _ RouteActionRequest) (*RouteActionResult, error) { + close(handlerEntered) + <-releaseHandler + return &RouteActionResult{SessionID: sessionID}, nil + }) + + routeDone := make(chan error, 1) + go func() { + _, err := svc.ApplyRouteAction(ctx, RouteActionRequest{ + SessionID: sessionID, + Action: RouteActionRetry, + }) + routeDone <- err + }() + select { + case <-handlerEntered: + case <-time.After(time.Second): + t.Fatal("route action handler did not start") + } + + _, _, _, _, _, err := svc.claimSessionRunningForPrompt( + ctx, "task1", sessionID, "", false, nil, nil, "", false, + ) + close(releaseHandler) + if routeErr := <-routeDone; routeErr != nil { + t.Fatalf("route action: %v", routeErr) + } + if !errors.Is(err, ErrAgentPromptInProgress) { + t.Fatalf("prompt admission error = %v, want route-action busy error", err) + } +} diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index 57ddf6d55e..24dd1149a1 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -607,6 +607,12 @@ type Service struct { // seam. routeActionHandler func(context.Context, RouteActionRequest) (*RouteActionResult, error) + // routeActionLocks serializes manual route actions for one session from the + // active-turn check through handler completion. It is separate from + // cancelInFlight because route handlers can reach the lifecycle lock. + routeActionLocksMu sync.Mutex + routeActionLocks map[string]*routeActionOperationLock + // profileExecutionResolver is the shared logical-to-concrete profile // boundary used by task and workflow launch paths. profileExecutionResolver *agentruntime.ProfileExecutionResolver @@ -1260,22 +1266,55 @@ func (s *Service) ApplyRouteAction(ctx context.Context, request RouteActionReque if err := s.authorizeSession(ctx, request.SessionID); err != nil { return nil, err } - lock, release := s.acquireCancelInFlightGuard(request.SessionID) + releaseRouteAction := s.acquireRouteActionOperationLock(request.SessionID) + defer releaseRouteAction() + if err := s.rejectRouteActionDuringActiveTurn(ctx, request.SessionID); err != nil { + return nil, err + } + return s.routeActionHandler(ctx, request) +} + +// rejectRouteActionDuringActiveTurn holds the cancel-in-flight guard only for +// the active-turn read, making it atomic against a concurrent cancel +// decision. The guard is released before ApplyRouteAction calls +// routeActionHandler: that handler can reach acquireSessionLifecycleLock +// (dynamic route launches relaunch through startCreatedSession), and holding +// the guard across it would invert the global lock order documented on +// acquireSessionLifecycleLock. +// +// route-action-vs-route-action is serialized locally by the per-session +// operation lock held by ApplyRouteAction. ResolveRouteAction's generation CAS +// remains the cross-process fence (for example, ErrStaleGeneration). +// route-action-vs-cancellation is protected for projection writes: every +// full-session-row write in dynamic_routing.go is conditioned via +// UpdateTaskSessionIfCurrentState on a state confirmed no earlier than the +// write itself, and relaunchDynamicTaskAfterFailure's CREATED reset reloads +// the session immediately before writing, refuses a reloaded terminal state, +// and CASes on that reload. Launch admission after that reset still needs the +// lifecycle cancellation fence described in the dynamic launch follow-up. +// route-action-vs-turn-admission is protected at the final RUNNING/turn claim: +// both direct and lifecycle prompt admission check the operation marker while +// holding cancelInFlight, so a new turn cannot be admitted after a route action +// claims the session. Pre-admission process resume can still race that handler +// and remains a separate lifecycle handoff concern. +func (s *Service) rejectRouteActionDuringActiveTurn(ctx context.Context, sessionID string) error { + if s.turnService == nil { + return nil + } + lock, release := s.acquireCancelInFlightGuard(sessionID) lock.Lock() defer func() { lock.Unlock() release() }() - if s.turnService != nil { - activeTurn, err := s.turnService.GetActiveTurn(ctx, request.SessionID) - if err != nil && !isNoActiveTurnError(err) { - return nil, fmt.Errorf("check active turn for route action: %w", err) - } - if activeTurn != nil { - return nil, ErrRouteActionActiveTurn - } + activeTurn, err := s.turnService.GetActiveTurn(ctx, sessionID) + if err != nil && !isNoActiveTurnError(err) { + return fmt.Errorf("check active turn for route action: %w", err) } - return s.routeActionHandler(ctx, request) + if activeTurn != nil { + return ErrRouteActionActiveTurn + } + return nil } // Status contains orchestrator status information @@ -2476,6 +2515,29 @@ func (s *Service) setSessionResetInProgress(sessionID string, inProgress bool) { // session's persisted conversation identity. Keep the critical section around // the complete reset/resume operation, including the bounded runtime wait, so // no caller can launch with a token that a concurrent reset is invalidating. +// +// Global lock order: this lock is OUTER to acquireCancelInFlightGuard. A +// caller that holds the cancel-in-flight guard for sessionID must release it +// (lockedCancelInFlightGuard's unlock/relock protocol) before acquiring this +// lock, or it can deadlock against every caller that already takes lifecycle +// first — resetAgentContext, ResumeTaskSessionWithOptions, +// ensureSessionRunning, reclaimIdleSession. handleAgentCompletedLocked and +// ApplyRouteAction's route-action dispatch follow this order. +// +// Known remaining inversions, both guard-then-lifecycle chains that reach +// this lock while still holding the cancel-in-flight guard, by design, to +// keep terminalization/failure-routing from racing launch admission. Closing +// them needs a redesign of that admission invariant, tracked separately +// rather than in this lock-order fix: +// - autoStartStepPrompt (event_handlers_workflow.go) holds the guard across +// the whole StartCreatedSession launch for a session in CREATED state. +// - runDetachedDynamicSuccessorLaunch (dynamic_launch.go) acquires its own +// guard and reaches this lock via relaunchDynamicTaskAfterFailure -> +// StartCreatedSession. Both handleAgentFailedLocked and +// handleAgentErrorEvent schedule it (via routeDynamicAgentFailure -> +// launchDynamicSuccessorAfterFailure -> launchDynamicSuccessorDetached) +// as a detached goroutine rather than reaching it while holding their own +// dispatch-level guard, so this is one call site, not two. func (s *Service) acquireSessionLifecycleLock(sessionID string) func() { if sessionID == "" { return func() {} diff --git a/apps/backend/internal/orchestrator/task_operations.go b/apps/backend/internal/orchestrator/task_operations.go index 8d1e8e38ce..68079533ee 100644 --- a/apps/backend/internal/orchestrator/task_operations.go +++ b/apps/backend/internal/orchestrator/task_operations.go @@ -5044,6 +5044,9 @@ func (s *Service) claimSessionRunningForPrompt( if s.isSessionResetInProgress(sessionID) { return nil, "", "", false, nil, ErrSessionResetInProgress } + if s.isRouteActionInFlight(sessionID) { + return nil, "", "", false, nil, fmt.Errorf("%w, route action is in progress", ErrAgentPromptInProgress) + } if expectedCurrentTurnID != "" { if s.turnService == nil { return nil, "", "", false, nil, errors.New("cannot verify expected prompt turn without turn service") @@ -5152,6 +5155,9 @@ func (s *Service) claimLifecycleSessionRunning( if s.isSessionResetInProgress(sessionID) { return nil, "", "", false, ErrSessionResetInProgress } + if s.isRouteActionInFlight(sessionID) { + return nil, "", "", false, fmt.Errorf("%w, route action is in progress", ErrAgentPromptInProgress) + } if claimEntryID != "" && !s.isCurrentQueuedDispatch(sessionID, claimEntryID) { return nil, "", "", false, errQueuedDispatchSuperseded diff --git a/apps/backend/internal/plugins/webapp_protocol_json.go b/apps/backend/internal/plugins/webapp_protocol_json.go index 640a4270e9..d18acda913 100644 --- a/apps/backend/internal/plugins/webapp_protocol_json.go +++ b/apps/backend/internal/plugins/webapp_protocol_json.go @@ -139,7 +139,7 @@ func webAppTaskFromSDK(task pluginsdk.Task) webAppTask { WorkflowStepID: task.WorkflowStepID, Position: task.Position, AssigneeAgentProfileID: task.AssigneeAgentProfileID, - Labels: task.Labels, + Labels: task.Labels, //nolint:staticcheck // retain API v1 label compatibility Autopilot: task.Autopilot, WIPAdmitted: task.WIPAdmitted, QueuedForStepID: task.QueuedForStepID,