Skip to content

Commit 3dd5af2

Browse files
Implement 0065 failure-isolation cause fidelity
FailureIsolationMiddleware now resolves the FailureIsolatedEvent's caught_exception through node_exception carrier wrappers to the nearest categorized originating cause, instead of reporting the masking node_exception at non-node placements (instance / branch / parent-node middleware). Both category and message come from the resolved cause for coherence, and the resolution agrees with what the retry classifier acts on. Node-level placement is unchanged.
1 parent f275136 commit 3dd5af2

3 files changed

Lines changed: 166 additions & 12 deletions

File tree

CHANGELOG.md

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). The
1414
### Changed
1515

1616
- **`RetryMiddleware` now takes a `RetryConfig` record** instead of individual constructor kwargs (proposal 0050 prep). The four retry settings (`max_attempts` / `classifier` / `backoff` / `on_retry`, each optional) move onto a frozen `RetryConfig`; construct as `RetryMiddleware(RetryConfig(max_attempts=...))`, while bare `RetryMiddleware()` still applies the defaults. This is a breaking change to the `RetryMiddleware` constructor. The record is the same shape the upcoming call-level `complete(retry=...)` parameter will accept, so one retry config serves both the per-node and per-call layers. `None` fields resolve to the canonical defaults (`default_classifier` / `exponential_jitter_backoff`) at use, preserving the prior behavior.
17+
- **Failure-isolation events report the originating cause's category at non-node placements** (proposal 0065, pipeline-utilities §6.3). When `FailureIsolationMiddleware` runs as instance middleware (§9.7), branch middleware (§11.7), or parent-node middleware on a fan-out / parallel-branches node, the graph engine has already wrapped the originating error as a `node_exception` carrier before the middleware catches it. `FailureIsolatedEvent.caught_exception.category` now resolves through that carrier (and any nested carriers) to the nearest categorized originating cause and reports its category instead of the masking `node_exception`, so the reported category agrees with what the §6.1 retry classifier acted on. For example, an instance whose retries exhaust on `provider_unavailable` now surfaces `provider_unavailable` rather than `node_exception`. The `message` tracks the resolved cause for category/message coherence. Node-level placement was already faithful and is unchanged, and catch/degrade behavior is unchanged at every site (only the event's reported cause changes). The wrapped-instance/branch lineage SHOULD (`fan_out_index` / `branch_name`) is deferred to a follow-up, since it needs the engine to surface per-instance identity to the wrapping-site middleware.
1718

1819
## [0.13.0] — 2026-06-09
1920

src/openarmature/graph/middleware/failure_isolation.py

Lines changed: 45 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,38 @@
5555
DegradedUpdate = Mapping[str, Any] | Callable[[Any], Mapping[str, Any]]
5656

5757

58+
def _resolve_cause(exc: Exception) -> BaseException:
59+
# Cause fidelity (proposal 0065 / §6.3, plus the python "nearest
60+
# categorized" refinement). Walk the ``__cause__`` chain to the most
61+
# actionable cause, skipping graph-engine §4 ``node_exception`` carrier
62+
# wrappers (``NodeException`` and subtypes such as
63+
# ``ParallelBranchesBranchFailed``) the engine applies at a non-node
64+
# placement (§9.7 instance, §11.7 branch, §9.6 / §11.6 parent-node
65+
# middleware). Returns the FIRST non-carrier exception that carries a
66+
# string ``category`` — so a deliberately re-categorized surface error
67+
# wins, while an uncategorized surface error resolves to the categorized
68+
# cause beneath it (the same chain §6.1's default classifier consults
69+
# for retryability, so the reported category agrees with what retry
70+
# acted on). When nothing in the chain carries a category, returns the
71+
# originating non-carrier raise (its own message, null category).
72+
# Node-level placement has no carrier, so ``exc`` itself is the
73+
# originating raise. The local import keeps ``errors`` off the
74+
# middleware module-load path, matching the deferred ``events`` import
75+
# in ``_emit_event``.
76+
from openarmature.graph.errors import NodeException
77+
78+
origin: BaseException | None = None
79+
current: BaseException | None = exc
80+
while current is not None:
81+
if not isinstance(current, NodeException):
82+
if origin is None:
83+
origin = current
84+
if isinstance(getattr(current, "category", None), str):
85+
return current
86+
current = current.__cause__
87+
return origin if origin is not None else exc
88+
89+
5890
class FailureIsolationMiddleware:
5991
"""Catch exceptions escaping the inner chain; return a degraded
6092
partial update.
@@ -145,17 +177,16 @@ def _emit_event(self, state: Any, exc: Exception, degraded: Mapping[str, Any]) -
145177
# path and defers it until the first catch.
146178
from openarmature.graph.events import CaughtException, FailureIsolatedEvent
147179

148-
# A categorized exception (e.g. an llm-provider error) carries a
149-
# string ``category``. When the engine has wrapped the original
150-
# in a graph-engine error before it reached the middleware, the
151-
# category rides on ``__cause__`` — walk it the same way the
152-
# default retry classifier does so the caught failure's category
153-
# survives the wrapping. A bare exception yields ``None``.
154-
category = getattr(exc, "category", None)
155-
if not isinstance(category, str):
156-
cause = getattr(exc, "__cause__", None)
157-
cause_category = getattr(cause, "category", None) if cause is not None else None
158-
category = cause_category if isinstance(cause_category, str) else None
180+
# Cause fidelity (proposal 0065 / §6.3). ``_resolve_cause`` walks
181+
# past graph-engine ``node_exception`` carrier wrappers to the
182+
# nearest categorized originating cause (see its comment); the
183+
# reported ``category`` and ``message`` both come from it so they
184+
# describe one exception — NOT the masking ``node_exception``. A
185+
# bare / uncategorized cause yields a null category. Node-level
186+
# placement has no carrier, so this is the caught exception itself.
187+
cause = _resolve_cause(exc)
188+
cause_category = getattr(cause, "category", None)
189+
category = cause_category if isinstance(cause_category, str) else None
159190
# ``attempt_index`` here is deliberately the NODE-level baseline,
160191
# not a per-attempt wire index: failure isolation is a node-level
161192
# concern ("the node, across its retries, was isolated"). When
@@ -176,7 +207,9 @@ def _emit_event(self, state: Any, exc: Exception, degraded: Mapping[str, Any]) -
176207
branch_name=current_branch_name(),
177208
pre_state=state,
178209
post_state=degraded,
179-
caught_exception=CaughtException(category=category, message=str(exc)),
210+
# ``message`` tracks the resolved cause (§6.3 SHOULD) so
211+
# the reported category and message describe one exception.
212+
caught_exception=CaughtException(category=category, message=str(cause)),
180213
)
181214
)
182215

tests/unit/test_failure_isolation_middleware.py

Lines changed: 120 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@
2828
append,
2929
deterministic_backoff,
3030
)
31+
from openarmature.graph.errors import NodeException, ParallelBranchesBranchFailed
3132
from openarmature.graph.middleware import NextCall
3233
from openarmature.observability.correlation import (
3334
_reset_active_dispatch,
@@ -41,6 +42,10 @@ class _TransientError(Exception):
4142
category = "provider_rate_limit"
4243

4344

45+
class _NonTransientError(Exception):
46+
category = "provider_invalid_request"
47+
48+
4449
def _raises(exc: BaseException) -> NextCall:
4550
"""A ``next`` callable that raises ``exc`` when invoked."""
4651

@@ -210,6 +215,121 @@ async def test_bare_exception_has_null_category() -> None:
210215
assert events[0].caught_exception.message == "plain"
211216

212217

218+
# ---------------------------------------------------------------------------
219+
# Cause fidelity at carrier-wrapper sites (proposal 0065)
220+
# ---------------------------------------------------------------------------
221+
222+
223+
async def test_node_exception_carrier_resolves_to_originating_category() -> None:
224+
# At a non-node placement the engine wraps the originating error as a
225+
# node_exception carrier before the middleware catches it; the event
226+
# reports the originating category, NOT the masking node_exception.
227+
events: list[Any] = []
228+
disp_token = _set_active_dispatch(lambda e: events.append(e))
229+
carrier = NodeException(node_name="work", cause=_TransientError("rate limited"), recoverable_state={})
230+
try:
231+
mw = FailureIsolationMiddleware(degraded_update={"result": []}, event_name="iso")
232+
out = await mw("s", _raises(carrier))
233+
finally:
234+
_reset_active_dispatch(disp_token)
235+
236+
assert out == {"result": []}
237+
assert isinstance(events[0], FailureIsolatedEvent)
238+
assert events[0].caught_exception.category == "provider_rate_limit"
239+
assert events[0].caught_exception.message == "rate limited"
240+
241+
242+
async def test_nested_carriers_resolve_to_originating_category() -> None:
243+
# Nested subgraph boundaries stack node_exception carriers; resolution
244+
# walks all of them to the originating cause.
245+
events: list[Any] = []
246+
disp_token = _set_active_dispatch(lambda e: events.append(e))
247+
inner = NodeException(node_name="inner", cause=_TransientError("rate limited"), recoverable_state={})
248+
outer = NodeException(node_name="outer", cause=inner, recoverable_state={})
249+
try:
250+
mw = FailureIsolationMiddleware(degraded_update={"result": []}, event_name="iso")
251+
await mw("s", _raises(outer))
252+
finally:
253+
_reset_active_dispatch(disp_token)
254+
255+
assert events[0].caught_exception.category == "provider_rate_limit"
256+
257+
258+
async def test_branch_carrier_subtype_resolves_to_originating_category() -> None:
259+
# The §11.7 branch site catches a ParallelBranchesBranchFailed (a
260+
# NodeException subtype); resolution still reaches the originating cause.
261+
events: list[Any] = []
262+
disp_token = _set_active_dispatch(lambda e: events.append(e))
263+
carrier = ParallelBranchesBranchFailed(
264+
node_name="dispatcher",
265+
cause=_TransientError("rate limited"),
266+
recoverable_state={},
267+
branch_name="only",
268+
)
269+
try:
270+
mw = FailureIsolationMiddleware(degraded_update={"result": []}, event_name="iso")
271+
await mw("s", _raises(carrier))
272+
finally:
273+
_reset_active_dispatch(disp_token)
274+
275+
assert events[0].caught_exception.category == "provider_rate_limit"
276+
277+
278+
async def test_carrier_over_uncategorized_cause_is_null() -> None:
279+
# Resolving through the carrier reaches a cause with no category, so the
280+
# reported category is null (the existing bare-exception rule).
281+
events: list[Any] = []
282+
disp_token = _set_active_dispatch(lambda e: events.append(e))
283+
carrier = NodeException(node_name="work", cause=ValueError("boom"), recoverable_state={})
284+
try:
285+
mw = FailureIsolationMiddleware(degraded_update={"result": []}, event_name="iso")
286+
await mw("s", _raises(carrier))
287+
finally:
288+
_reset_active_dispatch(disp_token)
289+
290+
assert events[0].caught_exception.category is None
291+
assert events[0].caught_exception.message == "boom"
292+
293+
294+
async def test_uncategorized_surface_resolves_to_categorized_cause() -> None:
295+
# A node that wraps a categorized provider error in an uncategorized
296+
# domain error: the event surfaces the underlying provider category
297+
# (agreeing with what §6.1's classifier retries on), and the message
298+
# tracks that same cause for coherence.
299+
events: list[Any] = []
300+
disp_token = _set_active_dispatch(lambda e: events.append(e))
301+
surface = ValueError("wrapped")
302+
surface.__cause__ = _TransientError("rate limited")
303+
carrier = NodeException(node_name="work", cause=surface, recoverable_state={})
304+
try:
305+
mw = FailureIsolationMiddleware(degraded_update={"result": []}, event_name="iso")
306+
await mw("s", _raises(carrier))
307+
finally:
308+
_reset_active_dispatch(disp_token)
309+
310+
assert events[0].caught_exception.category == "provider_rate_limit"
311+
assert events[0].caught_exception.message == "rate limited"
312+
313+
314+
async def test_categorized_surface_wins_over_deeper_cause() -> None:
315+
# A node that deliberately re-categorizes (raises a categorized error
316+
# FROM a categorized cause): the nearest category wins, so the node's
317+
# re-categorization is respected rather than the deeper cause.
318+
events: list[Any] = []
319+
disp_token = _set_active_dispatch(lambda e: events.append(e))
320+
surface = _NonTransientError("misconfigured")
321+
surface.__cause__ = _TransientError("rate limited")
322+
carrier = NodeException(node_name="work", cause=surface, recoverable_state={})
323+
try:
324+
mw = FailureIsolationMiddleware(degraded_update={"result": []}, event_name="iso")
325+
await mw("s", _raises(carrier))
326+
finally:
327+
_reset_active_dispatch(disp_token)
328+
329+
assert events[0].caught_exception.category == "provider_invalid_request"
330+
assert events[0].caught_exception.message == "misconfigured"
331+
332+
213333
async def test_no_event_outside_invocation() -> None:
214334
# current_dispatch() is None outside an invocation; the degrade still
215335
# happens, no event is emitted, and nothing raises.

0 commit comments

Comments
 (0)