graph: phase 2 polish — review followups (#15)

Late-breaking feedback from the merged Phase 2 PR. All cosmetic /
defensive; no behavior change to landed conformance fixtures.

- retry.py: rename `state` to `_state` in `default_classifier` and
  drop the `del state` workaround. Underscore-prefix is the canonical
  Python convention for intentionally unused parameters; both ruff
  and pyright recognize it without the awkward `del`.

- compiled.py: replace the `attempt_counter = [0]` mutable-list
  closure trick in `_step_function_node` with `nonlocal`. Same
  semantics, more idiomatic post-Python-3.0 style.

- middleware/_core.py: add a "Performance note" paragraph to
  compose_chain explaining the per-dispatch build pattern and
  flagging the fan-out optimization opportunity (cache the chain at
  compile time, inject only the per-dispatch attempt counter via a
  thin wrapper). Premature without numbers; worth measuring when
  Phase 3's fan-out runtime lands.

- test_middleware.py: two new unit tests pinning load-bearing
  contracts that retry/timing exercise only implicitly:
  - test_middleware_can_call_next_repeatedly — pins §2's reentrant-
    next contract by calling next 5 times in a loop.
  - test_timing_callback_failure_replaces_original_exception —
    pins current behavior when a TimingMiddleware on_complete
    callback raises on the failure path: the callback's exception
    propagates, the original is preserved on __context__ via
    Python's standard exception chaining. Documents the contract
    so a future change toward "preserve original via explicit
    raise-from" doesn't silently regress.

Total tests: 248 passing (+2), 1 skipped.

Author: chris-colinsky

src/openarmature/graph/compiled.py

@@ -330,7 +330,7 @@ async def _step_function_node(
         # The innermost layer dispatches the per-attempt event pair around a
         # single call to ``node.run``. Each call to this inner increments
         # the attempt counter; middleware composes around it.
-        attempt_counter = [0]
+        attempt_counter = 0
 
         async def innermost(s: Any) -> Mapping[str, Any]:
             # Per pipeline-utilities §5 + graph-engine §6: per-attempt
@@ -340,8 +340,9 @@ async def innermost(s: Any) -> Mapping[str, Any]:
             # the original `category` attribute (timing's
             # exception_category, retry's classifier). The engine wraps
             # any exception that escapes the chain, OUTSIDE this layer.
-            attempt_index = attempt_counter[0]
-            attempt_counter[0] = attempt_index + 1
+            nonlocal attempt_counter
+            attempt_index = attempt_counter
+            attempt_counter += 1
 
             self._dispatch_started(context, current, namespace, step, s, attempt_index=attempt_index)
 

src/openarmature/graph/middleware/_core.py

@@ -104,6 +104,17 @@ def compose_chain(
     Calling it once = one full chain traversal = at LEAST one call to
     ``innermost`` (more if a middleware calls ``next`` repeatedly, e.g.
     retry).
+
+    Performance note: this is called fresh per dispatch from
+    ``CompiledGraph._step_function_node``, producing one closure layer
+    per middleware on every node step. For typical workloads
+    (single-digit middleware × hundreds of node activations) this is
+    negligible. Under heavy fan-out (Phase 3+), e.g. 10K instances × 5
+    inner nodes × 3 middlewares = 150K closure constructions per
+    invocation; worth measuring with realistic workloads when the
+    fan-out runtime lands. The optimization shape (cache the chain at
+    compile time, inject only the per-dispatch attempt counter via a
+    thin wrapper) is straightforward but premature without numbers.
     """
     chain: ChainCall = innermost
     for mw in reversed(middlewares):

src/openarmature/graph/middleware/retry.py

@@ -36,7 +36,7 @@
 )
 
 
-def default_classifier(exc: Exception, state: Any) -> bool:
+def default_classifier(exc: Exception, _state: Any) -> bool:
     """Spec §6.1 default classifier — purely category-based, ignores state.
 
     Returns True if either the exception itself or its ``__cause__``
@@ -45,10 +45,12 @@ def default_classifier(exc: Exception, state: Any) -> bool:
     ``NodeException`` wrapping an llm-provider transient — per the spec:
     "a `node_exception` whose `__cause__` is a transient category MUST
     be classified as transient."
+
+    The ``_state`` parameter is ignored by the default; the leading
+    underscore is the canonical Python convention for "intentionally
+    unused" while keeping the signature stable for user-supplied
+    state-aware classifiers.
     """
-    # Suppress the unused-arg warning while keeping the signature stable
-    # for user-supplied state-aware classifiers.
-    del state
     direct = getattr(exc, "category", None)
     if isinstance(direct, str) and direct in TRANSIENT_CATEGORIES:
         return True

tests/unit/test_middleware.py

@@ -294,6 +294,68 @@ async def innermost(_state: Any) -> Mapping[str, Any]:
     assert abs(rec.duration_ms - 10.0) < 0.01
 
 
+# ===== Reentrant next: middleware can call next more than twice =====
+
+
+async def test_middleware_can_call_next_repeatedly() -> None:
+    """Per spec §2: a middleware MAY call ``next`` more than once. Retry
+    exercises this with N=2-3 attempts; this test pins the contract
+    independently by calling ``next`` 5 times in a loop and asserting the
+    inner runs exactly that many times."""
+    inner_calls = [0]
+
+    async def call_five_times(state: Any, next_: NextCall) -> Mapping[str, Any]:
+        last: Mapping[str, Any] = {}
+        for _ in range(5):
+            last = await next_(state)
+        return last
+
+    async def innermost(_state: Any) -> Mapping[str, Any]:
+        inner_calls[0] += 1
+        return {"trace": [f"call-{inner_calls[0]}"]}
+
+    chain = compose_chain([call_five_times], innermost)
+    result = await chain(TraceState())
+
+    assert inner_calls[0] == 5
+    # Final result is the partial returned from the LAST call to next.
+    assert result == {"trace": ["call-5"]}
+
+
+# ===== Timing callback failure on the failure path masks the original =====
+
+
+async def test_timing_callback_failure_replaces_original_exception() -> None:
+    """Pins the current behavior: when a node raises and TimingMiddleware's
+    ``on_complete`` ALSO raises in the failure path, the callback's
+    exception propagates instead of the original. Python's standard
+    exception chaining preserves the original on ``__context__``, but the
+    active exception observers see is the callback's.
+
+    Spec §6.2 says callbacks SHOULD be fast and infallible — this test
+    documents what happens if a user violates that, so a future change
+    that wants to preserve the original (e.g., via explicit ``raise exc
+    from cb_exc``) doesn't silently regress this contract.
+    """
+
+    async def bad_callback(_record: TimingRecord) -> None:
+        raise ValueError("callback bug")
+
+    async def innermost(_state: Any) -> Mapping[str, Any]:
+        raise _CategorizedFatal()
+
+    timing = TimingMiddleware(node_name="alpha", on_complete=bad_callback)
+    chain = compose_chain([timing], innermost)
+
+    with pytest.raises(ValueError, match="callback bug") as excinfo:
+        await chain(TraceState())
+
+    # The original node exception is preserved on __context__, the
+    # standard Python exception-chaining link for "exception raised
+    # while handling another."
+    assert isinstance(excinfo.value.__context__, _CategorizedFatal)
+
+
 # ===== 8. Subgraph isolation =====