fix(control-plane): gate every dispatch lane before persistence and complete the rejection contract

README.md

@@ -231,12 +231,12 @@ Most agent tools help you **write** agent logic. AgentField is what **runs** it
 
 ## How it scales
 
-The control plane is a stateless Go service. You put more of them behind a load balancer and the fleet grows horizontally. Work lands in a durable PostgreSQL queue with lease-based processing, so a crash or a restart resumes where it left off instead of dropping the job.
+The control plane is a stateless Go service. You put more of them behind a load balancer and the fleet grows horizontally. Work is admitted into a bounded in-process queue with backpressure (`429`/`503` plus `Retry-After`). On graceful shutdown, in-flight executions are terminated with `status_reason` `control_plane_shutdown` rather than silently dropped.
 
 | Property | What it means |
 |---|---|
 | Stateless Go control plane | Horizontal scaling behind a load balancer. Add replicas to add capacity. |
-| Durable PostgreSQL queue | Lease-based processing. Jobs survive crashes and restarts. |
+| Bounded in-process admission | Backpressure returns `429`/`503` with `Retry-After`; graceful shutdown records `control_plane_shutdown`. |
 | Async execution | Webhooks and SSE, no timeout limits. A single run can go for hours or days. |
 | Backpressure | Queue-depth limits and circuit breakers keep a fan-out from overwhelming downstream agents. |
 | Routing overhead | Roughly 100-200ms per cross-agent hop. It matters when a branch does little work per hop, so keep hops coarse when latency is tight. |
@@ -319,7 +319,7 @@ Two examples already run at this load. The [deep-research engine](https://agentf
 | Progress updates mid-execution | Intermediate payloads during long tasks |
 | Auto retries + exponential backoff | Transparent - control plane handles |
 | Backpressure + queue depth limits | Fair scheduling, circuit breakers |
-| Durable queue (PostgreSQL) | Atomic lease-based processing |
+| Bounded in-process queue | Backpressure and explicit graceful-shutdown termination |
 
 #### Memory (Distributed State)
 

control-plane/internal/handlers/coverage_handlers_90_additional_test.go

@@ -142,20 +142,20 @@ func TestExecutionController_CompletionAndFailureCoverage(t *testing.T) {
 
 		now := time.Now().UTC()
 		require.NoError(t, store.CreateExecutionRecord(context.Background(), &types.Execution{
-			ExecutionID: "exec-success",
-			RunID:       "run-success",
-			AgentNodeID: agent.ID,
-			Status:      types.ExecutionStatusRunning,
+			ExecutionID:  "exec-success",
+			RunID:        "run-success",
+			AgentNodeID:  agent.ID,
+			Status:       types.ExecutionStatusRunning,
 			InputPayload: []byte(`{"prompt":"hello"}`),
-			CreatedAt:   now,
-			StartedAt:   now,
-			UpdatedAt:   now,
+			CreatedAt:    now,
+			StartedAt:    now,
+			UpdatedAt:    now,
 		}))
 
 		plan := &preparedExecution{
 			exec: &types.Execution{
-				ExecutionID: "exec-success",
-				RunID:       "run-success",
+				ExecutionID:  "exec-success",
+				RunID:        "run-success",
 				InputPayload: []byte(`{"prompt":"hello"}`),
 			},
 			agent:  agent,
@@ -217,6 +217,9 @@ func TestExecutionController_CompletionAndFailureCoverage(t *testing.T) {
 }
 
 func TestPrepareExecution_AdditionalCoverage(t *testing.T) {
+	previousLimiter := concurrencyLimiter
+	concurrencyLimiter = nil
+	t.Cleanup(func() { concurrencyLimiter = previousLimiter })
 	gin.SetMode(gin.TestMode)
 
 	t.Run("versioned serverless agent registers webhook and preserves headers", func(t *testing.T) {
@@ -433,6 +436,9 @@ func TestAsyncExecutionJob_ProcessFallbackCoverage(t *testing.T) {
 					requestBody: []byte(`{"prompt":"hello"}`),
 					agent:       agent,
 					target:      &parsedTarget{NodeID: agent.ID, TargetName: "reasoner-a", TargetType: "reasoner"},
+					// The slot was acquired above on behalf of this plan, so the
+					// job owns it and process() must release it exactly once.
+					slotHeld: true,
 				},
 			}
 

control-plane/internal/handlers/coverage_raise_88_test.go

@@ -51,10 +51,10 @@ type statusManagerErrorStore struct {
 
 type registerCoverageStore struct {
 	*nodeRESTStorageStub
-	versioned       map[string]*types.AgentNode
-	deleteCalls     []string
-	updateLifeErr   error
-	registerErr     error
+	versioned     map[string]*types.AgentNode
+	deleteCalls   []string
+	updateLifeErr error
+	registerErr   error
 }
 
 func (s *statusManagerErrorStore) GetAgent(ctx context.Context, id string) (*types.AgentNode, error) {
@@ -258,10 +258,12 @@ func TestExecuteReasonerAndSkillHandlers_TransportFailures(t *testing.T) {
 		newHandle func(*reasonerHandlerStorage) gin.HandlerFunc
 	}{
 		{
-			name:      "reasoner",
-			route:     "/reasoners/:reasoner_id",
-			target:    "/reasoners/node-1.ping",
-			newStore:  func() *reasonerHandlerStorage { return newReasonerHandlerStorage(newReasonerAgent("http://127.0.0.1:1")) },
+			name:   "reasoner",
+			route:  "/reasoners/:reasoner_id",
+			target: "/reasoners/node-1.ping",
+			newStore: func() *reasonerHandlerStorage {
+				return newReasonerHandlerStorage(newReasonerAgent("http://127.0.0.1:1"))
+			},
 			newHandle: func(s *reasonerHandlerStorage) gin.HandlerFunc { return ExecuteReasonerHandler(s) },
 		},
 		{
@@ -934,6 +936,9 @@ func TestAsyncExecutionJob_ProcessAdditionalCoverage(t *testing.T) {
 			plan: preparedExecution{
 				exec:   &types.Execution{ExecutionID: "exec-1"},
 				target: &parsedTarget{NodeID: "node-1"},
+				// The slot was acquired above on behalf of this plan, so the job
+				// owns it and process() must release it exactly once.
+				slotHeld: true,
 			},
 		}
 

control-plane/internal/handlers/execute.go

@@ -269,6 +269,7 @@ func (c *executionController) handleSync(ctx *gin.Context) {
 		return
 	}
 	plan.executionMode = "sync"
+	defer plan.releaseSlot()
 
 	if plan.replayHit != nil {
 		if err := c.completeReplayHit(reqCtx, plan); err != nil {
@@ -290,14 +291,6 @@ func (c *executionController) handleSync(ctx *gin.Context) {
 		return
 	}
 
-	// Check LLM health and per-agent concurrency limits before proceeding
-	if err := CheckExecutionPreconditions(plan.target.NodeID, plan.llmEndpoint); err != nil {
-		_ = c.failExecution(reqCtx, plan, err, 0, nil)
-		writeExecutionError(ctx, err)
-		return
-	}
-	defer ReleaseExecutionSlot(plan.target.NodeID)
-
 	// Emit execution started event with full reasoner context
 	c.publishExecutionStartedEvent(plan)
 
@@ -437,9 +430,15 @@ func (c *executionController) handleSync(ctx *gin.Context) {
 func (c *executionController) handleAsync(ctx *gin.Context) {
 	reqCtx := ctx.Request.Context()
 	pool := getAsyncWorkerPool()
-	// A reservation covers both preparation and time waiting in the queue. The
-	// worker releases it on dequeue so only not-yet-started work consumes capacity.
-	if !pool.reserve() {
+	// A reservation covers preparation, queue wait, and the worker's dispatch. It
+	// is released when the worker job returns (early on an agent HTTP 202 ACK).
+	// Workers plus queue capacity bound admitted work; a paused execution pins a
+	// worker and reservation for up to 24 hours.
+	if reserved, stopped := pool.reserveForAdmission(); !reserved {
+		if stopped {
+			writeExecutionError(ctx, newControlPlaneShutdownError("async execution queue stopped; retry later"))
+			return
+		}
 		writeAsyncAdmissionError(ctx, http.StatusServiceUnavailable, "async execution queue is full; retry later")
 		return
 	}
@@ -456,9 +455,12 @@ func (c *executionController) handleAsync(ctx *gin.Context) {
 		return
 	}
 	plan.executionMode = "async"
+	// Preparation owns the slot until it is deliberately transferred to the
+	// worker job below. This also protects the recovered-panic interval between
+	// persistence and submission.
+	defer plan.releaseSlot()
 
 	if plan.replayHit != nil {
-		ReleaseExecutionSlot(plan.target.NodeID)
 		if err := c.completeReplayHit(reqCtx, plan); err != nil {
 			writeExecutionError(ctx, err)
 			return
@@ -497,12 +499,26 @@ func (c *executionController) handleAsync(ctx *gin.Context) {
 		controller: c,
 		plan:       *plan,
 	}
+	plan.slotHeld = false // ownership transferred to job
+	submitted := false
+	defer func() {
+		if !submitted {
+			job.plan.releaseSlot()
+		}
+	}()
 
 	if ok := pool.submitReserved(job); !ok {
-		ReleaseExecutionSlot(plan.target.NodeID) // Release since process() won't run
-		writeAsyncAdmissionError(ctx, http.StatusServiceUnavailable, "async execution queue stopped; retry later")
+		// The pool only refuses a reserved submission once it has stopped, i.e.
+		// the control plane is draining. Persist the outcome on a detached
+		// context: the request context is very likely being cancelled by the
+		// same shutdown, and a cancelled write would strand this row in
+		// "running" — exactly the orphan this branch exists to prevent.
+		shutdownErr := newControlPlaneShutdownError("async execution queue stopped; retry later")
+		job.terminateForControlPlaneShutdown(shutdownErr)
+		writeExecutionError(ctx, shutdownErr)
 		return
 	}
+	submitted = true
 	reserved = false
 
 	createdAt := plan.exec.CreatedAt.UTC().Format(time.RFC3339)

control-plane/internal/handlers/execute_async_test.go

@@ -36,6 +36,198 @@ func useAsyncPoolForTest(t *testing.T, pool *asyncWorkerPool) {
 	})
 }
 
+func TestExecuteAsyncHandler_PoolStoppedTerminatesPersistedRow(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	pool := newAsyncWorkerPool(0, 4)
+	useAsyncPoolForTest(t, pool)
+	oldLimiter := concurrencyLimiter
+	concurrencyLimiter = &AgentConcurrencyLimiter{maxPerAgent: 2}
+	defer func() { concurrencyLimiter = oldLimiter }()
+
+	agent := testRestartAgent("http://agent.example")
+	baseStore := newTestExecutionStorage(agent)
+	store := &stopPoolOnCreateStorage{testExecutionStorage: baseStore, pool: pool}
+	router := gin.New()
+	router.POST("/api/v1/execute/async/:target", ExecuteAsyncHandler(store, services.NewFilePayloadStore(t.TempDir()), nil, time.Second, ""))
+	req := httptest.NewRequest(http.MethodPost, "/api/v1/execute/async/node-1.reasoner-a", strings.NewReader(`{"input":{"foo":"bar"}}`))
+	req.Header.Set("Content-Type", "application/json")
+	resp := httptest.NewRecorder()
+	router.ServeHTTP(resp, req)
+
+	require.Equal(t, http.StatusServiceUnavailable, resp.Code)
+	require.Contains(t, resp.Body.String(), "async execution queue stopped")
+	require.Equal(t, "1", resp.Header().Get("Retry-After"))
+	var body map[string]any
+	require.NoError(t, json.Unmarshal(resp.Body.Bytes(), &body))
+	require.Equal(t, string(ErrorCategoryControlPlaneShutdown), body["error_category"])
+	require.Equal(t, float64(1), body["retry_after"])
+	records, err := baseStore.QueryExecutionRecords(context.Background(), types.ExecutionFilter{})
+	require.NoError(t, err)
+	require.Len(t, records, 1)
+	require.Equal(t, types.ExecutionStatusFailed, records[0].Status)
+	require.NotNil(t, records[0].StatusReason)
+	require.Equal(t, "control_plane_shutdown", *records[0].StatusReason)
+	workflows, err := baseStore.QueryWorkflowExecutions(context.Background(), types.WorkflowExecutionFilters{})
+	require.NoError(t, err)
+	require.Len(t, workflows, 1)
+	require.NotNil(t, workflows[0].StatusReason)
+	require.Equal(t, "control_plane_shutdown", *workflows[0].StatusReason)
+	require.Equal(t, string(records[0].Status), workflows[0].Status)
+	require.Zero(t, concurrencyLimiter.GetRunningCount("node-1"))
+}
+
+func TestExecuteAsyncHandler_AlreadyStoppedPoolReturnsShutdownWithoutPersistence(t *testing.T) {
+	pool := newAsyncWorkerPool(0, 2)
+	pool.mu.Lock()
+	pool.stopped = true
+	pool.mu.Unlock()
+	useAsyncPoolForTest(t, pool)
+	store := newTestExecutionStorage(testRestartAgent("http://agent.example"))
+	router := gin.New()
+	router.POST("/api/v1/execute/async/:target", ExecuteAsyncHandler(store, services.NewFilePayloadStore(t.TempDir()), nil, time.Second, ""))
+	req := httptest.NewRequest(http.MethodPost, "/api/v1/execute/async/node-1.reasoner-a", strings.NewReader(`{"input":{}}`))
+	req.Header.Set("Content-Type", "application/json")
+	resp := httptest.NewRecorder()
+	router.ServeHTTP(resp, req)
+
+	require.Equal(t, http.StatusServiceUnavailable, resp.Code)
+	require.Equal(t, "1", resp.Header().Get("Retry-After"))
+	var body map[string]interface{}
+	require.NoError(t, json.Unmarshal(resp.Body.Bytes(), &body))
+	require.Equal(t, string(ErrorCategoryControlPlaneShutdown), body["error_category"])
+	records, err := store.QueryExecutionRecords(context.Background(), types.ExecutionFilter{})
+	require.NoError(t, err)
+	require.Empty(t, records)
+}
+
+type stopPoolOnCreateStorage struct {
+	*testExecutionStorage
+	pool   *asyncWorkerPool
+	cancel context.CancelFunc
+}
+
+type cancelBeforeShutdownUpdateStore struct {
+	*testExecutionStorage
+	once          sync.Once
+	interleaveErr error
+}
+
+func (s *cancelBeforeShutdownUpdateStore) UpdateExecutionRecord(ctx context.Context, executionID string, update func(*types.Execution) (*types.Execution, error)) (*types.Execution, error) {
+	s.once.Do(func() {
+		reason := "cancelled_by_user"
+		_, s.interleaveErr = s.testExecutionStorage.UpdateExecutionRecord(ctx, executionID, func(current *types.Execution) (*types.Execution, error) {
+			current.Status = types.ExecutionStatusCancelled
+			current.StatusReason = &reason
+			return current, nil
+		})
+		if s.interleaveErr != nil {
+			return
+		}
+		s.interleaveErr = s.testExecutionStorage.UpdateWorkflowExecution(ctx, executionID, func(current *types.WorkflowExecution) (*types.WorkflowExecution, error) {
+			current.Status = string(types.ExecutionStatusCancelled)
+			current.StatusReason = &reason
+			return current, nil
+		})
+	})
+	if s.interleaveErr != nil {
+		return nil, s.interleaveErr
+	}
+	return s.testExecutionStorage.UpdateExecutionRecord(ctx, executionID, update)
+}
+
+func (s *stopPoolOnCreateStorage) CreateExecutionRecord(ctx context.Context, execution *types.Execution) error {
+	s.pool.mu.Lock()
+	s.pool.stopped = true
+	s.pool.mu.Unlock()
+	err := s.testExecutionStorage.CreateExecutionRecord(ctx, execution)
+	if s.cancel != nil {
+		s.cancel()
+	}
+	return err
+}
+
+func TestAsyncShutdownTerminalizationPreservesCancellationInterleaving(t *testing.T) {
+	oldLimiter := concurrencyLimiter
+	concurrencyLimiter = &AgentConcurrencyLimiter{maxPerAgent: 2}
+	t.Cleanup(func() { concurrencyLimiter = oldLimiter })
+	require.NoError(t, concurrencyLimiter.Acquire("node-1"))
+
+	base := newTestExecutionStorage(testRestartAgent("http://agent.example"))
+	now := time.Now().UTC()
+	exec := &types.Execution{
+		ExecutionID: "exec-cancel-race", RunID: "run-cancel-race", AgentNodeID: "node-1", NodeID: "node-1",
+		ReasonerID: "reasoner-a", Status: types.ExecutionStatusRunning,
+		CreatedAt: now, StartedAt: now, UpdatedAt: now,
+	}
+	require.NoError(t, base.CreateExecutionRecord(context.Background(), exec))
+	require.NoError(t, base.StoreWorkflowExecution(context.Background(), &types.WorkflowExecution{
+		ExecutionID: exec.ExecutionID, WorkflowID: exec.RunID, RunID: &exec.RunID,
+		AgentNodeID: "node-1", ReasonerID: "reasoner-a", Status: string(types.ExecutionStatusRunning),
+		CreatedAt: now, StartedAt: now, UpdatedAt: now,
+	}))
+	store := &cancelBeforeShutdownUpdateStore{testExecutionStorage: base}
+	target, err := parseTarget("node-1.reasoner-a")
+	require.NoError(t, err)
+	job := asyncExecutionJob{
+		controller: newExecutionController(store, nil, nil, time.Second, ""),
+		plan: preparedExecution{
+			exec: exec, target: target, slotHeld: true,
+		},
+	}
+
+	job.terminateForControlPlaneShutdown(newControlPlaneShutdownError("control plane stopped"))
+
+	stored, err := base.GetExecutionRecord(context.Background(), exec.ExecutionID)
+	require.NoError(t, err)
+	workflow, err := base.GetWorkflowExecution(context.Background(), exec.ExecutionID)
+	require.NoError(t, err)
+	require.Equal(t, types.ExecutionStatusCancelled, stored.Status)
+	require.Equal(t, string(types.ExecutionStatusCancelled), workflow.Status)
+	require.Equal(t, "cancelled_by_user", *stored.StatusReason)
+	require.Equal(t, "cancelled_by_user", *workflow.StatusReason)
+	require.Zero(t, concurrencyLimiter.GetRunningCount("node-1"))
+}
+
+func TestAsyncForcedWorkerCancellationReleasesOnlyOwnedSlot(t *testing.T) {
+	oldLimiter := concurrencyLimiter
+	concurrencyLimiter = &AgentConcurrencyLimiter{maxPerAgent: 2}
+	t.Cleanup(func() { concurrencyLimiter = oldLimiter })
+	require.NoError(t, concurrencyLimiter.Acquire("node-1"))
+	require.NoError(t, concurrencyLimiter.Acquire("node-1"))
+
+	agent := testRestartAgent("http://agent.example")
+	store := newTestExecutionStorage(agent)
+	now := time.Now().UTC()
+	exec := &types.Execution{
+		ExecutionID: "exec-forced-stop", RunID: "run-forced-stop", AgentNodeID: "node-1", NodeID: "node-1",
+		ReasonerID: "reasoner-a", Status: types.ExecutionStatusRunning,
+		CreatedAt: now, StartedAt: now, UpdatedAt: now,
+	}
+	require.NoError(t, store.CreateExecutionRecord(context.Background(), exec))
+	require.NoError(t, store.StoreWorkflowExecution(context.Background(), &types.WorkflowExecution{
+		ExecutionID: exec.ExecutionID, WorkflowID: exec.RunID, RunID: &exec.RunID,
+		AgentNodeID: "node-1", ReasonerID: "reasoner-a", Status: string(types.ExecutionStatusRunning),
+		CreatedAt: now, StartedAt: now, UpdatedAt: now,
+	}))
+	target, err := parseTarget("node-1.reasoner-a")
+	require.NoError(t, err)
+	job := asyncExecutionJob{
+		controller: newExecutionController(store, nil, nil, time.Second, ""),
+		plan: preparedExecution{
+			exec: exec, target: target, agent: agent, requestBody: []byte(`{}`), slotHeld: true,
+		},
+	}
+	workerCtx, cancel := context.WithCancel(context.Background())
+	cancel()
+	job.processWithContext(workerCtx)
+
+	// This job owned one of two live slots. Forced shutdown must not consume the
+	// other execution's count through failForControlPlaneShutdown plus defer.
+	require.EqualValues(t, 1, concurrencyLimiter.GetRunningCount("node-1"))
+	ReleaseExecutionSlot("node-1")
+	require.Zero(t, concurrencyLimiter.GetRunningCount("node-1"))
+}
+
 func TestExecuteAsyncHandler_QueueSaturation(t *testing.T) {
 	gin.SetMode(gin.TestMode)
 	useAsyncPoolForTest(t, newAsyncWorkerPool(1, 1))