diff --git a/control-plane/internal/handlers/reasoners.go b/control-plane/internal/handlers/reasoners.go index 2a159d2d8..0a1e205fa 100644 --- a/control-plane/internal/handlers/reasoners.go +++ b/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 { diff --git a/control-plane/internal/handlers/reasoners_test.go b/control-plane/internal/handlers/reasoners_test.go index 05d63861d..23f08d36f 100644 --- a/control-plane/internal/handlers/reasoners_test.go +++ b/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. diff --git a/sdk/python/agentfield/agent.py b/sdk/python/agentfield/agent.py index 948e53067..bd47bb6a1 100644 --- a/sdk/python/agentfield/agent.py +++ b/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: diff --git a/sdk/python/tests/test_async_execution.py b/sdk/python/tests/test_async_execution.py index 5f87a35d6..2e3d875db 100644 --- a/sdk/python/tests/test_async_execution.py +++ b/sdk/python/tests/test_async_execution.py @@ -1,12 +1,20 @@ import asyncio +import json +import threading import time +from http.server import BaseHTTPRequestHandler, HTTPServer import httpx import pytest +import requests -from agentfield.agent import Agent +from agentfield.agent import ( + Agent, + _is_retryable_status_callback_error, + _is_transient_status_callback, +) from agentfield.agent_pause import PauseClock -from agentfield.client import AgentFieldClient, ApprovalResult +from agentfield.client import AgentFieldClient, AgentFieldClientError, ApprovalResult @pytest.mark.asyncio @@ -69,7 +77,7 @@ def __init__(self, status_code: int): async def fake_request(self, method, url, **kwargs): attempts["count"] += 1 if attempts["count"] < 3: - raise RuntimeError("transient error") + raise httpx.ConnectError("connection refused") return DummyResponse(200) monkeypatch.setattr(AgentFieldClient, "_async_request", fake_request) @@ -92,6 +100,474 @@ async def fake_sleep(delay): assert sleeps == [1, 2] +def test_is_transient_status_callback_matches_go_policy(): + """Same contract as Go ``isTransientPollStatus``: 408, 429, and 5xx only.""" + for code in (408, 429, 500, 502, 503, 504, 599): + assert _is_transient_status_callback(code), code + for code in (100, 204, 301, 302, 400, 401, 403, 404, 409, 418, 422, 499): + assert not _is_transient_status_callback(code), code + + +def test_is_retryable_status_callback_error_matches_transport_failures(): + """Do-equivalent failures retry; request-construction and bugs do not.""" + retryable = ( + httpx.ConnectTimeout("timed out"), + httpx.ReadTimeout("timed out"), + httpx.WriteTimeout("timed out"), + httpx.PoolTimeout("timed out"), + httpx.ConnectError("connection refused"), + httpx.RemoteProtocolError("malformed response"), + TimeoutError("timed out"), + ConnectionResetError("reset"), + requests.exceptions.Timeout("timed out"), + requests.exceptions.ConnectionError("connection refused"), + requests.exceptions.ConnectTimeout("timed out"), + requests.exceptions.ChunkedEncodingError("incomplete response"), + requests.exceptions.ContentDecodingError("invalid gzip response"), + ) + for exc in retryable: + assert _is_retryable_status_callback_error(exc), type(exc) + + permanent = ( + httpx.InvalidURL("http://[::1"), + TypeError("bad payload"), + RuntimeError("bug"), + ValueError("bug"), + requests.exceptions.InvalidURL("http://[::1"), + requests.exceptions.MissingSchema("control-plane"), + asyncio.CancelledError(), + ) + for exc in permanent: + assert not _is_retryable_status_callback_error(exc), type(exc) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("status_code", [400, 401, 403, 404, 409]) +async def test_post_execution_status_does_not_retry_permanent_client_errors( + monkeypatch, status_code +): + agent = Agent( + node_id="test-agent", agentfield_server="http://control", auto_register=False + ) + attempts = {"count": 0} + timeouts = [] + + class DummyResponse: + def __init__(self, code: int): + self.status_code = code + + async def fake_request(self, method, url, **kwargs): + attempts["count"] += 1 + timeouts.append(kwargs.get("timeout")) + return DummyResponse(status_code) + + monkeypatch.setattr(AgentFieldClient, "_async_request", fake_request) + + sleeps = [] + + async def fake_sleep(delay): + sleeps.append(delay) + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + + warns = [] + errors = [] + + def capture_warn(message, **kwargs): + warns.append(message) + + def capture_error(message, **kwargs): + errors.append(message) + + monkeypatch.setattr("agentfield.agent.log_warn", capture_warn) + monkeypatch.setattr("agentfield.agent.log_error", capture_error) + + await agent._post_execution_status( + "http://control/api/v1/executions/exec-1/status", + {"status": "succeeded"}, + "exec-1", + ) + + assert attempts["count"] == 1 + assert sleeps == [] + assert timeouts == [30.0] + assert warns == [] + assert len(errors) == 1 + assert str(status_code) in errors[0] + assert "not retrying" in errors[0] + + +@pytest.mark.integration +@pytest.mark.asyncio +@pytest.mark.parametrize("first_response", ["truncated", "invalid-gzip", 401, 404]) +async def test_post_execution_status_requests_fallback(monkeypatch, first_response): + """Exercise eager response reads through the real requests fallback.""" + received = [] + + class CallbackHandler(BaseHTTPRequestHandler): + def do_POST(self): + payload = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) + received.append((self.path, payload)) + first_attempt = len(received) == 1 + status = 200 + body = b"{}" + if first_attempt: + status = first_response if isinstance(first_response, int) else 503 + if first_response == "truncated": + body = b"partial" + elif first_response == "invalid-gzip": + body = b"not gzip data" + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Connection", "close") + if first_attempt and first_response == "invalid-gzip": + self.send_header("Content-Encoding", "gzip") + content_length = len(body) + if first_attempt and first_response == "truncated": + content_length += 100 + self.send_header("Content-Length", str(content_length)) + self.end_headers() + self.wfile.write(body) + self.wfile.flush() + self.close_connection = True + + def log_message(self, format, *args): + pass + + agent = Agent( + node_id="test-agent", agentfield_server="http://control", auto_register=False + ) + + async def unavailable_async_client(self): + raise AgentFieldClientError("httpx unavailable") + + monkeypatch.setattr( + AgentFieldClient, "get_async_http_client", unavailable_async_client + ) + # Keep the production session/adapter setup while isolating shared state and + # avoiding environment proxy settings for this loopback-only server. + monkeypatch.setattr(AgentFieldClient, "_shared_sync_session", None) + session = AgentFieldClient._get_sync_session() + session.trust_env = False + sleeps = [] + errors = [] + + async def fake_sleep(delay): + sleeps.append(delay) + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + monkeypatch.setattr( + "agentfield.agent.log_error", lambda message: errors.append(message) + ) + path = "/api/v1/executions/exec-1/status" + payload = {"status": "succeeded", "result": {"value": 7}} + server = HTTPServer(("127.0.0.1", 0), CallbackHandler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + await agent._post_execution_status( + f"http://127.0.0.1:{server.server_port}{path}", payload, "exec-1" + ) + if isinstance(first_response, int): + assert received == [(path, payload)] + assert sleeps == [] + assert len(errors) == 1 + assert str(first_response) in errors[0] + assert "not retrying" in errors[0] + else: + assert received == [(path, payload), (path, payload)] + assert sleeps == [1] + assert errors == [] + finally: + session.close() + server.shutdown() + server.server_close() + thread.join(timeout=5) + assert not thread.is_alive() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("status_code", [408, 429, 502]) +async def test_post_execution_status_retries_transient_http_statuses( + monkeypatch, status_code +): + agent = Agent( + node_id="test-agent", agentfield_server="http://control", auto_register=False + ) + attempts = {"count": 0} + + class DummyResponse: + def __init__(self, code: int): + self.status_code = code + + async def fake_request(self, method, url, **kwargs): + attempts["count"] += 1 + if attempts["count"] == 1: + return DummyResponse(status_code) + return DummyResponse(204) + + monkeypatch.setattr(AgentFieldClient, "_async_request", fake_request) + + sleeps = [] + + async def fake_sleep(delay): + sleeps.append(delay) + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + + await agent._post_execution_status( + "http://control/api/v1/executions/exec-1/status", + {"status": "succeeded"}, + "exec-1", + ) + + assert attempts["count"] == 2 + assert sleeps == [1] + + +@pytest.mark.asyncio +async def test_post_execution_status_exhausts_transient_retries(monkeypatch): + agent = Agent( + node_id="test-agent", agentfield_server="http://control", auto_register=False + ) + attempts = {"count": 0} + + class DummyResponse: + def __init__(self, code: int): + self.status_code = code + + async def fake_request(self, method, url, **kwargs): + attempts["count"] += 1 + assert kwargs.get("timeout") == 30.0 + return DummyResponse(503) + + monkeypatch.setattr(AgentFieldClient, "_async_request", fake_request) + + sleeps = [] + + async def fake_sleep(delay): + sleeps.append(delay) + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + + await agent._post_execution_status( + "http://control/api/v1/executions/exec-1/status", + {"status": "running"}, + "exec-1", + ) + + assert attempts["count"] == 5 + assert sleeps == [1, 2, 4, 8] + + +@pytest.mark.asyncio +async def test_post_execution_status_stops_after_permanent_status(monkeypatch): + """A later 409 must not keep the backoff going after an earlier 502.""" + agent = Agent( + node_id="test-agent", agentfield_server="http://control", auto_register=False + ) + codes = [502, 409] + attempts = {"count": 0} + + class DummyResponse: + def __init__(self, code: int): + self.status_code = code + + async def fake_request(self, method, url, **kwargs): + attempts["count"] += 1 + return DummyResponse(codes.pop(0)) + + monkeypatch.setattr(AgentFieldClient, "_async_request", fake_request) + + sleeps = [] + + async def fake_sleep(delay): + sleeps.append(delay) + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + + await agent._post_execution_status( + "http://control/api/v1/executions/exec-1/status", + {"status": "succeeded"}, + "exec-1", + ) + + assert attempts["count"] == 2 + assert sleeps == [1] + assert codes == [] + + +@pytest.mark.asyncio +async def test_post_execution_status_shutdown_still_single_attempt(monkeypatch): + agent = Agent( + node_id="test-agent", agentfield_server="http://control", auto_register=False + ) + agent._shutdown_cancelling = True + attempts = {"count": 0} + timeouts = [] + + class DummyResponse: + def __init__(self, code: int): + self.status_code = code + + async def fake_request(self, method, url, **kwargs): + attempts["count"] += 1 + timeouts.append(kwargs.get("timeout")) + return DummyResponse(503) + + monkeypatch.setattr(AgentFieldClient, "_async_request", fake_request) + + sleeps = [] + + async def fake_sleep(delay): + sleeps.append(delay) + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + + await agent._post_execution_status( + "http://control/api/v1/executions/exec-1/status", + {"status": "cancelled"}, + "exec-1", + ) + + assert attempts["count"] == 1 + assert sleeps == [] + assert timeouts == [5.0] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "exc", + [ + httpx.ConnectTimeout("timed out"), + httpx.ReadTimeout("timed out"), + httpx.ConnectError("connection refused"), + requests.exceptions.Timeout("timed out"), + requests.exceptions.ConnectionError("connection refused"), + ], +) +async def test_post_execution_status_retries_transport_errors(monkeypatch, exc): + agent = Agent( + node_id="test-agent", agentfield_server="http://control", auto_register=False + ) + attempts = {"count": 0} + + class DummyResponse: + def __init__(self, status_code: int): + self.status_code = status_code + + async def fake_request(self, method, url, **kwargs): + attempts["count"] += 1 + if attempts["count"] == 1: + raise exc + return DummyResponse(204) + + monkeypatch.setattr(AgentFieldClient, "_async_request", fake_request) + + sleeps = [] + + async def fake_sleep(delay): + sleeps.append(delay) + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + + await agent._post_execution_status( + "http://control/api/v1/executions/exec-1/status", + {"status": "running"}, + "exec-1", + ) + + assert attempts["count"] == 2 + assert sleeps == [1] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "exc", + [ + httpx.InvalidURL("http://[::1"), + TypeError("bad payload"), + RuntimeError("bug"), + requests.exceptions.InvalidURL("http://[::1"), + requests.exceptions.MissingSchema("control-plane"), + ], +) +async def test_post_execution_status_does_not_retry_permanent_exceptions( + monkeypatch, exc +): + agent = Agent( + node_id="test-agent", agentfield_server="http://control", auto_register=False + ) + attempts = {"count": 0} + warns = [] + errors = [] + + async def fake_request(self, method, url, **kwargs): + attempts["count"] += 1 + raise exc + + monkeypatch.setattr(AgentFieldClient, "_async_request", fake_request) + + sleeps = [] + + async def fake_sleep(delay): + sleeps.append(delay) + + def capture_warn(message, **kwargs): + warns.append(message) + + def capture_error(message, **kwargs): + errors.append(message) + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + monkeypatch.setattr("agentfield.agent.log_warn", capture_warn) + monkeypatch.setattr("agentfield.agent.log_error", capture_error) + + await agent._post_execution_status( + "http://control/api/v1/executions/exec-1/status", + {"status": "succeeded"}, + "exec-1", + ) + + assert attempts["count"] == 1 + assert sleeps == [] + assert warns == [] + assert len(errors) == 1 + assert "not retrying" in errors[0] + + +@pytest.mark.asyncio +async def test_post_execution_status_propagates_cancellation(monkeypatch): + """Cancellation must leave the retry loop immediately, with no backoff.""" + agent = Agent( + node_id="test-agent", agentfield_server="http://control", auto_register=False + ) + attempts = {"count": 0} + + async def fake_request(self, method, url, **kwargs): + attempts["count"] += 1 + raise asyncio.CancelledError() + + monkeypatch.setattr(AgentFieldClient, "_async_request", fake_request) + + sleeps = [] + + async def fake_sleep(delay): + sleeps.append(delay) + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + + with pytest.raises(asyncio.CancelledError): + await agent._post_execution_status( + "http://control/api/v1/executions/exec-1/status", + {"status": "succeeded"}, + "exec-1", + ) + + assert attempts["count"] == 1 + assert sleeps == [] + + @pytest.mark.asyncio async def test_pause_does_not_consume_active_timeout_budget(monkeypatch): """A reasoner paused in ``app.pause()`` for longer than the wall-clock