Skip to content

Commit cc430e4

Browse files
Implement 0065 failure-isolation cause fidelity (#153)
* 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. * Guard _resolve_cause against cyclic cause chains Traverse only BaseException instances and track visited exceptions by id, so a non-exception __cause__ ends the walk and a self-referential chain terminates instead of hanging or raising in the degrade path.
1 parent f275136 commit cc430e4

3 files changed

Lines changed: 189 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: 50 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,43 @@
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+
seen: set[int] = set()
81+
# Traverse only BaseException instances (a non-exception ``__cause__``
82+
# ends the walk) and guard against a cyclic ``__cause__`` chain so a
83+
# malformed chain can't hang or crash the degrade path.
84+
while isinstance(current, BaseException) and id(current) not in seen:
85+
seen.add(id(current))
86+
if not isinstance(current, NodeException):
87+
if origin is None:
88+
origin = current
89+
if isinstance(getattr(current, "category", None), str):
90+
return current
91+
current = current.__cause__
92+
return origin if origin is not None else exc
93+
94+
5895
class FailureIsolationMiddleware:
5996
"""Catch exceptions escaping the inner chain; return a degraded
6097
partial update.
@@ -145,17 +182,16 @@ def _emit_event(self, state: Any, exc: Exception, degraded: Mapping[str, Any]) -
145182
# path and defers it until the first catch.
146183
from openarmature.graph.events import CaughtException, FailureIsolatedEvent
147184

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
185+
# Cause fidelity (proposal 0065 / §6.3). ``_resolve_cause`` walks
186+
# past graph-engine ``node_exception`` carrier wrappers to the
187+
# nearest categorized originating cause (see its comment); the
188+
# reported ``category`` and ``message`` both come from it so they
189+
# describe one exception — NOT the masking ``node_exception``. A
190+
# bare / uncategorized cause yields a null category. Node-level
191+
# placement has no carrier, so this is the caught exception itself.
192+
cause = _resolve_cause(exc)
193+
cause_category = getattr(cause, "category", None)
194+
category = cause_category if isinstance(cause_category, str) else None
159195
# ``attempt_index`` here is deliberately the NODE-level baseline,
160196
# not a per-attempt wire index: failure isolation is a node-level
161197
# concern ("the node, across its retries, was isolated"). When
@@ -176,7 +212,9 @@ def _emit_event(self, state: Any, exc: Exception, degraded: Mapping[str, Any]) -
176212
branch_name=current_branch_name(),
177213
pre_state=state,
178214
post_state=degraded,
179-
caught_exception=CaughtException(category=category, message=str(exc)),
215+
# ``message`` tracks the resolved cause (§6.3 SHOULD) so
216+
# the reported category and message describe one exception.
217+
caught_exception=CaughtException(category=category, message=str(cause)),
180218
)
181219
)
182220

tests/unit/test_failure_isolation_middleware.py

Lines changed: 138 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,139 @@ 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+
333+
async def test_cyclic_cause_chain_terminates() -> None:
334+
# Defensive: a self-referential __cause__ chain must not hang the
335+
# degrade path. Resolution terminates and the node still degrades.
336+
events: list[Any] = []
337+
disp_token = _set_active_dispatch(lambda e: events.append(e))
338+
a = NodeException(node_name="a", cause=ValueError("seed"), recoverable_state={})
339+
b = NodeException(node_name="b", cause=a, recoverable_state={})
340+
a.__cause__ = b # cycle: a -> b -> a
341+
try:
342+
mw = FailureIsolationMiddleware(degraded_update={"result": []}, event_name="iso")
343+
out = await mw("s", _raises(a))
344+
finally:
345+
_reset_active_dispatch(disp_token)
346+
347+
assert out == {"result": []}
348+
assert len(events) == 1
349+
350+
213351
async def test_no_event_outside_invocation() -> None:
214352
# current_dispatch() is None outside an invocation; the degrade still
215353
# happens, no event is emitted, and nothing raises.

0 commit comments

Comments
 (0)