From ed54ceb2291e5c8a73dbd23a61c9c7b76316c799 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 16 Oct 2026 14:20:18 +0000 Subject: [PATCH 1/4] fix(sdk/python): stop retrying execution status callbacks on non-retryable 4xx Align _post_execution_status with the Go SDK's postExecutionStatus: transport errors, 408, 429, and 5xx retry up to five times with 1+2+4+8s backoff and a 30s per-attempt timeout; every other HTTP status returns immediately. 404 is permanent now that #1086 persists the execution row. Refs #1088 Co-authored-by: Steven Huang --- sdk/python/agentfield/agent.py | 33 +++- sdk/python/tests/test_async_execution.py | 199 ++++++++++++++++++++++- 2 files changed, 230 insertions(+), 2 deletions(-) diff --git a/sdk/python/agentfield/agent.py b/sdk/python/agentfield/agent.py index 948e53067..b617e9c85 100644 --- a/sdk/python/agentfield/agent.py +++ b/sdk/python/agentfield/agent.py @@ -642,6 +642,20 @@ 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 + + class Agent(FastAPI): """ AgentField Agent - FastAPI subclass for creating AI agent nodes. @@ -2957,7 +2971,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( @@ -2976,6 +2994,12 @@ async def _post_execution_status( log_warn( f"Async status update failed with {response.status_code} for execution {execution_id}" ) + 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. + permanent_status = response.status_code + break except Exception as exc: # pragma: no cover - network errors log_warn( f"Async status update attempt {attempt + 1} failed for {execution_id}: {exc}" @@ -2987,8 +3011,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..acff33a81 100644 --- a/sdk/python/tests/test_async_execution.py +++ b/sdk/python/tests/test_async_execution.py @@ -4,7 +4,7 @@ import httpx import pytest -from agentfield.agent import Agent +from agentfield.agent import Agent, _is_transient_status_callback from agentfield.agent_pause import PauseClock from agentfield.client import AgentFieldClient, ApprovalResult @@ -92,6 +92,203 @@ 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 + + +@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) + + 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] + + +@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 async def test_pause_does_not_consume_active_timeout_budget(monkeypatch): """A reasoner paused in ``app.pause()`` for longer than the wall-clock From bc8149d8189ea35e4f990e040132ecd1bf3f78e7 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 16 Oct 2026 14:47:47 +0000 Subject: [PATCH 2/4] fix(sdk/python): retry status callbacks only on transport failures Match Go postExecutionStatus: resend httpx.TransportError (including timeouts), the requests connection and timeout errors from the sync fallback, and HTTP 408, 429, and 5xx. Fail fast on request-construction and programming errors such as InvalidURL and TypeError. CancelledError still leaves the loop immediately. A permanent HTTP status is logged once, and the stale no-cover pragma on the handler is gone. Refs #1088 Co-authored-by: Steven Huang --- sdk/python/agentfield/agent.py | 53 ++++++- sdk/python/tests/test_async_execution.py | 189 ++++++++++++++++++++++- 2 files changed, 235 insertions(+), 7 deletions(-) diff --git a/sdk/python/agentfield/agent.py b/sdk/python/agentfield/agent.py index b617e9c85..41cba8190 100644 --- a/sdk/python/agentfield/agent.py +++ b/sdk/python/agentfield/agent.py @@ -656,6 +656,39 @@ def _is_transient_status_callback(status_code: int) -> bool: 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``). 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, + ), + ) + + class Agent(FastAPI): """ AgentField Agent - FastAPI subclass for creating AI agent nodes. @@ -2991,16 +3024,26 @@ async def _post_execution_status( f"Sent async status update for {execution_id} (attempt {attempt + 1})" ) return - log_warn( - f"Async status update failed with {response.status_code} for execution {execution_id}" - ) 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. + # 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 - except Exception as exc: # pragma: no cover - network errors + log_warn( + f"Async status update failed with {response.status_code} for execution {execution_id}" + ) + 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}" ) diff --git a/sdk/python/tests/test_async_execution.py b/sdk/python/tests/test_async_execution.py index acff33a81..be561dc80 100644 --- a/sdk/python/tests/test_async_execution.py +++ b/sdk/python/tests/test_async_execution.py @@ -3,8 +3,13 @@ import httpx import pytest +import requests -from agentfield.agent import Agent, _is_transient_status_callback +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 @@ -69,7 +74,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) @@ -100,6 +105,37 @@ def test_is_transient_status_callback_matches_go_policy(): 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"), + ) + 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( @@ -129,6 +165,18 @@ async def fake_sleep(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"}, @@ -138,6 +186,10 @@ async def fake_sleep(delay): 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.asyncio @@ -289,6 +341,139 @@ async def fake_sleep(delay): 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 From 0d12834478461e76cbedec8699b18cb3b4e1a460 Mon Sep 17 00:00:00 2001 From: remote-controlled-man <134844631+remote-controlled-man@users.noreply.github.com> Date: Tue, 20 Oct 2026 10:31:36 +0800 Subject: [PATCH 3/4] fix(sdk/python): retry interrupted callback response bodies --- sdk/python/agentfield/agent.py | 8 +- sdk/python/tests/test_async_execution.py | 96 +++++++++++++++++++++++- 2 files changed, 101 insertions(+), 3 deletions(-) diff --git a/sdk/python/agentfield/agent.py b/sdk/python/agentfield/agent.py index 41cba8190..bd47bb6a1 100644 --- a/sdk/python/agentfield/agent.py +++ b/sdk/python/agentfield/agent.py @@ -664,8 +664,10 @@ def _is_retryable_status_callback_error(exc: BaseException) -> bool: 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``). Bare ``TimeoutError`` and - ``ConnectionError`` cover the same failures if they surface unwrapped. + (``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``. @@ -685,6 +687,8 @@ def _is_retryable_status_callback_error(exc: BaseException) -> bool: ( requests.exceptions.ConnectionError, requests.exceptions.Timeout, + requests.exceptions.ChunkedEncodingError, + requests.exceptions.ContentDecodingError, ), ) diff --git a/sdk/python/tests/test_async_execution.py b/sdk/python/tests/test_async_execution.py index be561dc80..2e3d875db 100644 --- a/sdk/python/tests/test_async_execution.py +++ b/sdk/python/tests/test_async_execution.py @@ -1,5 +1,8 @@ import asyncio +import json +import threading import time +from http.server import BaseHTTPRequestHandler, HTTPServer import httpx import pytest @@ -11,7 +14,7 @@ _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 @@ -119,6 +122,8 @@ def test_is_retryable_status_callback_error_matches_transport_failures(): 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) @@ -192,6 +197,95 @@ def capture_error(message, **kwargs): 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( From 8a4b07f862754fd607bc3e7b51aec7c67d2b6d3e Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Tue, 20 Oct 2026 09:00:37 -0400 Subject: [PATCH 4/4] fix(control-plane): fail legacy dispatch on execution insert errors Return a structured 500 before reasoner or skill dispatch when the execution row cannot be persisted, including serverless nodes. Previously the insert failure was only logged and the call was dispatched anyway, so the agent's status callback could get a 404 that this PR now treats as permanent while a later workflow event created the row, leaving it running with no result. Cover insert failures and keep the persist-before-dispatch success path. Co-Authored-By: Claude Opus 5.5 (1M context) --- control-plane/internal/handlers/reasoners.go | 14 ++- .../internal/handlers/reasoners_test.go | 114 ++++++++++++------ 2 files changed, 86 insertions(+), 42 deletions(-) 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.