fix(sdk/python): stop retrying permanent status callbacks

control-plane/internal/handlers/reasoners.go

@@ -96,7 +96,7 @@ func persistLegacyAcknowledgement(ctx context.Context, storageProvider storage.S
 // createLegacyExecutionRecord writes the executions row for a legacy
 // /reasoners or /skills call before it is dispatched, so the agent's
 // status callback (POST /api/v1/executions/:id/status) finds it.
-func createLegacyExecutionRecord(ctx context.Context, storageProvider storage.StorageProvider, wfExec *types.WorkflowExecution) {
+func createLegacyExecutionRecord(ctx context.Context, storageProvider storage.StorageProvider, wfExec *types.WorkflowExecution) error {
 	exec := &types.Execution{
 		ExecutionID:       wfExec.ExecutionID,
 		RunID:             wfExec.WorkflowID,
@@ -117,7 +117,9 @@ func createLegacyExecutionRecord(ctx context.Context, storageProvider storage.St
 			Err(err).
 			Str("execution_id", wfExec.ExecutionID).
 			Msg("failed to create execution record for legacy call")
+		return err
 	}
+	return nil
 }
 
 // finalizeLegacyExecutionRecord copies the outcome of a legacy call onto its
@@ -317,7 +319,10 @@ func ExecuteReasonerHandler(storageProvider storage.StorageProvider) gin.Handler
 		}
 		workflowExecution.InputData = inputJSON
 		workflowExecution.InputSize = len(inputJSON)
-		createLegacyExecutionRecord(ctx, storageProvider, workflowExecution)
+		if err := createLegacyExecutionRecord(ctx, storageProvider, workflowExecution); err != nil {
+			c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create execution record"})
+			return
+		}
 
 		agentURL, endpointErr := checkedAgentURL(targetNode, &parsedTarget{TargetName: reasonerName, TargetType: "reasoner"})
 		if endpointErr != nil {
@@ -719,7 +724,10 @@ func ExecuteSkillHandler(storageProvider storage.StorageProvider) gin.HandlerFun
 		}
 		workflowExecution.InputData = inputJSON
 		workflowExecution.InputSize = len(inputJSON)
-		createLegacyExecutionRecord(ctx, storageProvider, workflowExecution)
+		if err := createLegacyExecutionRecord(ctx, storageProvider, workflowExecution); err != nil {
+			c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create execution record"})
+			return
+		}
 
 		agentURL, endpointErr := checkedAgentURL(targetNode, &parsedTarget{TargetName: skillName, TargetType: "skill"})
 		if endpointErr != nil {

control-plane/internal/handlers/reasoners_test.go

@@ -9,6 +9,7 @@ import (
 	"net/http/httptest"
 	"path/filepath"
 	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -135,8 +136,8 @@ func TestExecuteReasonerHandler_NodeLookupAndAvailabilityErrors(t *testing.T) {
 		{
 			name: "node not found",
 			store: &reasonerHandlerStorage{
-				exec:                 newTestExecutionStorage(nil),
-				getAgentErr:          errors.New("missing"),
+				exec:        newTestExecutionStorage(nil),
+				getAgentErr: errors.New("missing"),
 			},
 			wantCode:   http.StatusNotFound,
 			wantErrMsg: "node 'node-404' not found",
@@ -547,43 +548,6 @@ func TestExecuteReasonerHandler_AgentStatusCallbackFindsExecution(t *testing.T)
 	require.Equal(t, "agent reported failure", *final.ErrorMessage)
 }
 
-type failingCreateRecordStorage struct {
-	*reasonerHandlerStorage
-}
-
-func (s *failingCreateRecordStorage) CreateExecutionRecord(ctx context.Context, execution *types.Execution) error {
-	return errors.New("insert failed")
-}
-
-// A failed executions insert is logged and the legacy call still goes through,
-// so the endpoint gains no new failure mode.
-func TestExecuteReasonerHandler_ExecutionRowInsertFailureDoesNotFailCall(t *testing.T) {
-	gin.SetMode(gin.TestMode)
-
-	store := newReasonerHandlerStorage(nil)
-	observed := make(chan legacyDispatchObservation, 1)
-	agentServer := legacyRowAgent(t, store, observed)
-	defer agentServer.Close()
-	store.agent = newReasonerAgent(agentServer.URL)
-
-	router := gin.New()
-	router.POST("/reasoners/:reasoner_id", ExecuteReasonerHandler(&failingCreateRecordStorage{store}))
-
-	req := httptest.NewRequest(http.MethodPost, "/reasoners/node-1.ping", strings.NewReader(`{}`))
-	req.Header.Set("Content-Type", "application/json")
-	resp := httptest.NewRecorder()
-	router.ServeHTTP(resp, req)
-	require.Equal(t, http.StatusOK, resp.Code)
-
-	obs := <-observed
-	require.Nil(t, obs.record)
-
-	wfExec, err := store.GetWorkflowExecution(context.Background(), obs.executionID)
-	require.NoError(t, err)
-	require.NotNil(t, wfExec)
-	require.Equal(t, string(types.ExecutionStatusSucceeded), string(wfExec.Status))
-}
-
 // newLegacyLocalStorage opens a real LocalStorage for the legacy lifecycle
 // tests, so the terminal-state guards and context handling of the real store
 // apply.
@@ -637,6 +601,78 @@ var legacyEndpoints = []legacyEndpoint{
 	},
 }
 
+type failedLegacyInsertStorage struct {
+	storage.StorageProvider
+	attempts int
+}
+
+func (s *failedLegacyInsertStorage) CreateExecutionRecord(context.Context, *types.Execution) error {
+	s.attempts++
+	return errors.New("injected execution insert failure: private database details")
+}
+
+func TestLegacyExecute_ExecutionInsertFailureStopsDispatch(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	for _, endpoint := range legacyEndpoints {
+		for _, deployment := range []string{"standard", "serverless"} {
+			t.Run(endpoint.name+"/"+deployment, func(t *testing.T) {
+				var dispatches atomic.Int32
+				agentServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+					dispatches.Add(1)
+					w.Header().Set("Content-Type", "application/json")
+					_, _ = w.Write([]byte(`{"ok":true}`))
+				}))
+				defer agentServer.Close()
+				agent := endpoint.agent(agentServer.URL)
+				agent.DeploymentType = deployment
+				base := newReasonerHandlerStorage(agent)
+				store := &failedLegacyInsertStorage{StorageProvider: base}
+				router := gin.New()
+				endpoint.mount(router, store)
+				req := httptest.NewRequest(http.MethodPost, endpoint.path, strings.NewReader(`{"input":{"q":1}}`))
+				req.Header.Set("Content-Type", "application/json")
+				resp := httptest.NewRecorder()
+				router.ServeHTTP(resp, req)
+
+				require.Equal(t, http.StatusInternalServerError, resp.Code, resp.Body.String())
+				require.JSONEq(t, `{"error":"failed to create execution record"}`, resp.Body.String())
+				require.Contains(t, resp.Header().Get("Content-Type"), "application/json")
+				require.Equal(t, 1, store.attempts)
+				require.Zero(t, dispatches.Load(), "failed persistence must prevent agent dispatch")
+				workflows, err := base.QueryWorkflowExecutions(context.Background(), types.WorkflowExecutionFilters{})
+				require.NoError(t, err)
+				require.Empty(t, workflows, "rejected work must not leave a running workflow")
+			})
+		}
+	}
+}
+
+func TestLegacyExecute_ServerlessCreatesExecutionRowBeforeDispatch(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	for _, endpoint := range legacyEndpoints {
+		t.Run(endpoint.name, func(t *testing.T) {
+			store := newReasonerHandlerStorage(nil)
+			observed := make(chan legacyDispatchObservation, 1)
+			agentServer := legacyRowAgent(t, store, observed)
+			defer agentServer.Close()
+			store.agent = endpoint.agent(agentServer.URL)
+			store.agent.DeploymentType = "serverless"
+			router := gin.New()
+			endpoint.mount(router, store)
+			req := httptest.NewRequest(http.MethodPost, endpoint.path, strings.NewReader(`{"input":{"q":1}}`))
+			req.Header.Set("Content-Type", "application/json")
+			resp := httptest.NewRecorder()
+			router.ServeHTTP(resp, req)
+			require.Equal(t, http.StatusOK, resp.Code, resp.Body.String())
+			obs := <-observed
+			require.NotEmpty(t, obs.executionID)
+			require.NotNil(t, obs.record, "execution must exist before serverless dispatch")
+			require.Equal(t, string(types.ExecutionStatusRunning), obs.record.Status)
+			require.JSONEq(t, `{"q":1}`, string(obs.record.InputPayload))
+		})
+	}
+}
+
 // asyncAckAgent answers like the Python and Go SDKs do for an async
 // execution: HTTP 202 with status "processing", while the work continues in
 // the background.

sdk/python/agentfield/agent.py

@@ -642,6 +642,57 @@ async def shutdown(self, timeout: int = 5):
             self._dispatcher_task = None
 
 
+def _is_transient_status_callback(status_code: int) -> bool:
+    """Return whether an execution status-callback HTTP status is retryable.
+
+    Mirrors the Go SDK's ``isTransientPollStatus`` (``sdk/go/agent/agent.go``),
+    which ``postExecutionStatus`` uses to decide whether to resend a callback.
+    408 (request timeout), 429 (too many requests), and any 5xx are transient.
+    Every other status is permanent: resending a 400, 401, 403, 404, or 409
+    cannot succeed, so the caller must fail fast.
+    """
+    if status_code in (408, 429):
+        return True
+    return status_code >= 500
+
+
+def _is_retryable_status_callback_error(exc: BaseException) -> bool:
+    """Return whether a status-callback exception is worth resending.
+
+    Go retries errors from ``http.Client.Do`` and returns immediately when
+    ``http.NewRequest`` fails. ``httpx.TransportError`` is that Do bucket:
+    connect/read/write/pool timeouts, network errors, and protocol errors.
+    ``httpx.TimeoutException`` is a ``TransportError``. The sync fallback in
+    ``AgentFieldClient._async_request`` raises the requests equivalents
+    (``ConnectionError`` and ``Timeout``), plus ``ChunkedEncodingError`` and
+    ``ContentDecodingError`` from eagerly reading a broken response body.
+    Bare ``TimeoutError`` and ``ConnectionError`` cover the same failures if
+    they surface unwrapped.
+
+    Request-construction and programming errors are permanent. That includes
+    ``httpx.InvalidURL``, ``requests.exceptions.InvalidURL``, and ``TypeError``.
+    """
+    if isinstance(exc, (TimeoutError, ConnectionError)):
+        return True
+
+    import httpx
+
+    if isinstance(exc, httpx.TransportError):
+        return True
+
+    import requests
+
+    return isinstance(
+        exc,
+        (
+            requests.exceptions.ConnectionError,
+            requests.exceptions.Timeout,
+            requests.exceptions.ChunkedEncodingError,
+            requests.exceptions.ContentDecodingError,
+        ),
+    )
+
+
 class Agent(FastAPI):
     """
     AgentField Agent - FastAPI subclass for creating AI agent nodes.
@@ -2957,7 +3008,11 @@ async def _post_execution_status(
         safe_payload = jsonable_encoder(payload)
         shutting_down = getattr(self, "_shutdown_cancelling", False)
         attempts = 1 if shutting_down else max_retries
+        # Match the Go SDK: 30s per attempt, five attempts, backoff of
+        # 1s, 2s, 4s, 8s (``time.Second << attempt``). Shutdown still sends
+        # a single 5s attempt so a stuck callback cannot hold process exit.
         request_timeout = 5.0 if shutting_down else 30.0
+        permanent_status: Optional[int] = None
         for attempt in range(attempts):
             try:
                 response = await self.client._async_request(
@@ -2973,10 +3028,26 @@ async def _post_execution_status(
                             f"Sent async status update for {execution_id} (attempt {attempt + 1})"
                         )
                     return
+                if not _is_transient_status_callback(response.status_code):
+                    # A client error such as 404 (unknown execution) or 409
+                    # (terminal status conflict) will not change on a resend,
+                    # so fail fast instead of backing off. The single error
+                    # log is below; a warning here would report the same failure twice.
+                    permanent_status = response.status_code
+                    break
                 log_warn(
                     f"Async status update failed with {response.status_code} for execution {execution_id}"
                 )
-            except Exception as exc:  # pragma: no cover - network errors
+            except Exception as exc:
+                if not _is_retryable_status_callback_error(exc):
+                    # Bad URLs and programming errors (TypeError) will not
+                    # succeed on a resend. Log once and return so the
+                    # exhaustion message below does not fire as well.
+                    log_error(
+                        f"Async status update for {execution_id} failed: "
+                        f"{exc}; not retrying"
+                    )
+                    return
                 log_warn(
                     f"Async status update attempt {attempt + 1} failed for {execution_id}: {exc}"
                 )
@@ -2987,8 +3058,15 @@ async def _post_execution_status(
                 f"Abandoning async status for {execution_id} during shutdown "
                 f"after {request_timeout:g}s attempt"
             )
+        elif permanent_status is not None:
+            log_error(
+                f"Async status update for {execution_id} failed with "
+                f"{permanent_status}; not retrying"
+            )
         else:
-            log_error(f"Failed to deliver async status for {execution_id} after retries")
+            log_error(
+                f"Failed to deliver async status for {execution_id} after retries"
+            )
 
     def _build_execution_callback_url(self, execution_id: str) -> Optional[str]:
         if not self.agentfield_server or not execution_id: