feat(sdk): ReasonerFailed exception for reporting reasoner failure with result

sdk/python/agentfield/__init__.py

@@ -64,6 +64,7 @@
     AgentFieldClientError,
     ExecutionTimeoutError,
     MemoryAccessError,
+    ReasonerFailed,
     RegistrationError,
     ValidationError,
 )
@@ -158,6 +159,7 @@
     "AgentFieldClientError",
     "ExecutionTimeoutError",
     "MemoryAccessError",
+    "ReasonerFailed",
     "RegistrationError",
     "ValidationError",
     # Trigger / webhook plugin system

sdk/python/agentfield/agent.py

@@ -2573,6 +2573,14 @@ async def _watchdog() -> None:
                 "execution_id": execution_id,
                 "reasoner": reasoner_name,
             }
+            # A reasoner that ran, determined its own work failed, and wants its
+            # structured outcome preserved raises ReasonerFailed(result=...).
+            # Carry that result onto the failed-status payload so the control
+            # plane records status=failed WITHOUT discarding the rich result
+            # (it stores the result payload regardless of terminal status).
+            from .exceptions import ReasonerFailed
+            if isinstance(exc, ReasonerFailed) and exc.result is not None:
+                payload["result"] = jsonable_encoder(exc.result)
             log_error(f"Execution {execution_id} failed asynchronously: {exc}")
         finally:
             # If we landed here without the reasoner finishing (e.g. our own

sdk/python/agentfield/exceptions.py

@@ -33,6 +33,44 @@ class ExecutionFailedError(AgentFieldClientError):
     pass
 
 
+class ReasonerFailed(AgentFieldError):
+    """Raised *inside* a reasoner to report that the work ran but failed.
+
+    Returning a value from a reasoner — even one whose payload says
+    ``success: False`` — makes the async execution handler record the
+    execution as ``succeeded`` (it only distinguishes "returned" from
+    "raised", never inspecting the result). A build that completes zero
+    issues and merges nothing therefore surfaces as green, which is easy to
+    act on incorrectly.
+
+    Raise this when the reasoner has determined its own work failed but you
+    still want the structured ``result`` preserved on the execution record.
+    The handler posts ``status="failed"`` to the control plane while also
+    sending ``result`` (the control plane stores the result payload
+    regardless of status), so the rich outcome — debt, DAG state, any PR
+    that was opened — is not lost behind a bare error string.
+
+    Args:
+        message: Human-readable failure summary (becomes the execution error).
+        result: Optional structured result to preserve on the execution
+            record (e.g. ``BuildResult.model_dump()``). JSON-encoded by the
+            handler before it is posted.
+        error_details: Optional structured error metadata, mirrored onto the
+            status payload's ``error_details`` field.
+    """
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        result: object | None = None,
+        error_details: object | None = None,
+    ) -> None:
+        super().__init__(message)
+        self.result = result
+        self.error_details = error_details
+
+
 class ExecutionTimeoutError(AgentFieldError):
     """Execution timed out waiting for completion."""
 
@@ -80,6 +118,7 @@ class ValidationError(AgentFieldError):
     "AgentFieldError",
     "AgentFieldClientError",
     "ExecutionFailedError",
+    "ReasonerFailed",
     "ExecutionTimeoutError",
     "ExecutionCancelledError",
     "MemoryAccessError",

sdk/python/tests/test_async_execution.py

@@ -976,3 +976,108 @@ async def test_poll_active_executions_still_times_out_without_pause_clock():
     assert state.status == ExecutionStatus.TIMEOUT, (
         "no pause_clock attached: wallclock overdue check must still fire"
     )
+
+
+@pytest.mark.asyncio
+async def test_reasoner_failed_reports_failed_status_and_preserves_result(monkeypatch):
+    """A reasoner that raises ReasonerFailed must surface as status=failed
+    while still carrying its structured result (so the control plane, which
+    stores the result payload regardless of terminal status, keeps the rich
+    outcome rather than just a bare error string)."""
+    from agentfield.exceptions import ReasonerFailed
+
+    agent = Agent(
+        node_id="test-agent", agentfield_server="http://control", auto_register=False
+    )
+
+    @agent.reasoner()
+    async def build() -> dict:
+        await asyncio.sleep(0)
+        raise ReasonerFailed(
+            "Build failed: 0/3 issues completed, no branches merged",
+            result={"success": False, "completed_issues": 0, "merged_branches": 0},
+            error_details={"reason": "empty_build"},
+        )
+
+    recorded = []
+
+    class DummyResponse:
+        status_code = 200
+
+        def json(self):
+            return {}
+
+    async def fake_request(self, method, url, **kwargs):
+        recorded.append({"url": url, "json": kwargs.get("json")})
+        return DummyResponse()
+
+    monkeypatch.setattr(AgentFieldClient, "_async_request", fake_request)
+
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=agent), base_url="http://agent"
+    ) as client:
+        response = await client.post(
+            "/reasoners/build",
+            json={},
+            headers={"X-Execution-ID": "exec-fail-1"},
+        )
+
+    assert response.status_code == 202
+    await asyncio.sleep(0.1)
+
+    status_calls = [e for e in recorded if "/executions/" in e["url"]]
+    assert status_calls, "expected async status callback"
+    payload = status_calls[-1]["json"]
+    assert payload["status"] == "failed"
+    assert "0/3 issues" in payload["error"]
+    assert payload["error_details"] == {"reason": "empty_build"}
+    # Structured result preserved alongside the failed status.
+    assert payload["result"]["success"] is False
+    assert payload["result"]["completed_issues"] == 0
+
+
+@pytest.mark.asyncio
+async def test_plain_exception_failed_status_has_no_result(monkeypatch):
+    """Regression guard: a generic exception still maps to status=failed with
+    no result key — ReasonerFailed's result-preservation must not leak into
+    the ordinary failure path."""
+    agent = Agent(
+        node_id="test-agent", agentfield_server="http://control", auto_register=False
+    )
+
+    @agent.reasoner()
+    async def boom() -> dict:
+        await asyncio.sleep(0)
+        raise RuntimeError("kaboom")
+
+    recorded = []
+
+    class DummyResponse:
+        status_code = 200
+
+        def json(self):
+            return {}
+
+    async def fake_request(self, method, url, **kwargs):
+        recorded.append({"url": url, "json": kwargs.get("json")})
+        return DummyResponse()
+
+    monkeypatch.setattr(AgentFieldClient, "_async_request", fake_request)
+
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=agent), base_url="http://agent"
+    ) as client:
+        response = await client.post(
+            "/reasoners/boom",
+            json={},
+            headers={"X-Execution-ID": "exec-fail-2"},
+        )
+
+    assert response.status_code == 202
+    await asyncio.sleep(0.1)
+
+    status_calls = [e for e in recorded if "/executions/" in e["url"]]
+    payload = status_calls[-1]["json"]
+    assert payload["status"] == "failed"
+    assert payload["error"] == "kaboom"
+    assert "result" not in payload