graph: phase 2 PR review fixes

Three real bugs surfaced by review:

- compiled.py: _step_subgraph_node now wraps raw exceptions escaping
  the parent's middleware chain as NodeException with recoverable_state,
  matching _step_function_node. Previously a middleware-raised exception
  around a SubgraphNode dispatch escaped uncaught.

- test_pipeline_utilities.py: the determinism re-run path
  (run_count > 1) now passes the clock stub when re-translating
  middleware. Was dropping it, so a future fixture combining
  determinism with TimingMiddleware would silently use real
  time.monotonic on subsequent runs.

- test_pipeline_utilities.py: the subgraph middleware translation
  path now also passes the clock stub. Same omission as the
  determinism path — affects any future fixture combining a subgraph
  with TimingMiddleware.

No current fixture exercises the latter two combinations; fix is
defensive so the harness faithfully translates fixture intent.

Author: chris-colinsky

src/openarmature/graph/compiled.py

@@ -443,7 +443,17 @@ async def innermost(s: Any) -> Mapping[str, Any]:
             innermost,
         )
 
-        final_partial = await chain(state)
+        try:
+            final_partial = await chain(state)
+        except RuntimeGraphError:
+            raise
+        except Exception as e:
+            # Same wrap as _step_function_node: a raw exception escaping
+            # the parent's middleware chain (e.g., a middleware bug or a
+            # projection error) becomes NodeException tagged with the
+            # SubgraphNode's wrapper name so §4 recoverable_state is
+            # preserved.
+            raise NodeException(node_name=current, cause=e, recoverable_state=state) from e
         return _merge_partial(state, final_partial, self.reducers, current)
 
     @staticmethod

tests/conformance/test_pipeline_utilities.py

@@ -331,7 +331,7 @@ async def _run_one(spec: Mapping[str, Any], monkeypatch: pytest.MonkeyPatch) ->
     if "subgraph" in spec:
         sub_spec = spec["subgraph"]
         sub_sinks = sinks  # same sinks; subgraph middleware shares the harness's lists
-        sub_graph_mw, sub_node_mw = _translate_middleware_block(sub_spec.get("middleware"), sub_sinks)
+        sub_graph_mw, sub_node_mw = _translate_middleware_block(sub_spec.get("middleware"), sub_sinks, clock)
         sub_built = build_graph(
             sub_spec,
             model_name=f"{sub_spec['name'].title()}State",
@@ -382,7 +382,7 @@ async def _run_one(spec: Mapping[str, Any], monkeypatch: pytest.MonkeyPatch) ->
         run_graph_mw, run_node_mw = (
             (graph_mw, node_mw)
             if run_count == 1
-            else _translate_middleware_block(spec.get("middleware"), run_sinks)
+            else _translate_middleware_block(spec.get("middleware"), run_sinks, clock)
         )
         run_built = build_graph(
             spec,