From 28e80b4a663550917f4f48b374beba6072f6e807 Mon Sep 17 00:00:00 2001 From: chris-colinsky Date: Sat, 9 May 2026 13:17:56 -0700 Subject: [PATCH] =?UTF-8?q?graph:=20phase=206.1=20PR-C.1=20=E2=80=94=20pro?= =?UTF-8?q?posal=200012=20+=20v0.9.0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implement spec graph-engine v0.9.0 (proposal 0012): the engine's ``completed`` observer event for the just-completed node now dispatches AFTER edge evaluation completes, not between merge and edge eval. Edge-resolution failures (``routing_error``, ``edge_exception``) populate the ``error`` field of the preceding node's ``completed`` event, sharing the started/completed pair rather than producing a separate event pair (per §3 step 3 + §6 revisions in proposal 0012). The existing observer ``_handle_completed`` ERROR-mapping path picks up the new error categories automatically; no observer change required. Mechanism: introduce ``_StepResult(state, finalize_completed)`` returned by the per-step dispatchers. Failure-path dispatches (``node_exception`` / ``reducer_error`` / ``state_validation_error``) stay inline inside ``innermost`` — those errors short-circuit before edge eval can run, preserving §3's "before the failure propagates to the caller" MUST. On success, ``innermost`` stores ``(attempt_index, pre_state, merged)`` for the FINAL successful attempt; the outer scope builds a ``finalize_completed`` closure that ``_invoke`` calls AFTER edge eval, passing either ``None`` (success → dispatch completed with ``post_state``) or the edge error (dispatch completed with ``error`` populated). ``_step_subgraph_node`` is unchanged in behavior — the wrapper is transparent per fixture 013. Returns ``_StepResult(state, _no_op_finalize)`` so edge errors after a transparent wrapper propagate silently per §4 without an observer event pair (proposal 0012's "share the preceding node's pair" MUST is conditional on the pair existing). Same for middleware that short-circuits without invoking ``next``. Inline docstring at the wrapper's _StepResult return anchors the contract for future readers. Pin sites bumped to v0.9.0 (three-place sync per CLAUDE.md): ``openarmature-spec`` submodule pointer to ``78fe9b4``, ``pyproject.toml`` ``tool.openarmature.spec_version``, ``src/openarmature/__init__.py`` ``__spec_version__``, ``tests/test_smoke.py`` drift-guard literal. The ``OTelObserver.spec_version`` attribute updates automatically via ``_read_spec_version`` (per PR-A). Tests: - New unit tests in ``tests/unit/test_runtime_errors.py`` verify the new event shape: exactly one started + one completed for the preceding node, completed has ``error`` populated and ``post_state`` absent, downstream node never fires events. - ``tests/conformance/test_conformance.py`` gains a ``_run_fixture_020`` driver for the new graph-engine fixture ``020-observer-edge-error-events`` (two sub-cases: routing_error and edge_exception). The fixture's edge-condition ``callable:`` directives don't match the adapter DSL; the driver translates the semantics directly. Cases-shape support added to the generic harness for future cases-shape graph-engine fixtures. - ``tests/conformance/test_observability.py`` removes ``004-otel-routing-error-attribution`` from ``_DEFERRED_FIXTURES`` (was deferred through Phase 6.1 PR-C pending this proposal) and adds a ``_run_fixture_004`` driver. Verifies preceding node's span ends ERROR with ``status_description == "routing_error"`` + recorded exception event + ``openarmature.error.category`` attribute, no separate edge-function span, invocation span propagates ERROR. - ``tests/conformance/harness/expectations.py``: ``GraphEngineExpected`` gains an ``invariants`` field so fixture 020's ``invariants:`` block parses cleanly. 398 tests pass (was 392; net +6 from new fixture drivers + unit tests + fixture-020 sub-cases that previously errored). 4 skipped (was 5; fixture 004 now driven). Pyright clean. PR-C.2 (fan-out per-instance dispatch + fixture 006) and PR-C.3 (observer ``prepare_sync`` + fixture 010) sit independently behind their respective architectural pieces; land in either order after this PR ships. Phase 6.1 closes when all three merge. --- openarmature-spec | 2 +- pyproject.toml | 2 +- src/openarmature/__init__.py | 2 +- src/openarmature/graph/compiled.py | 266 ++++++++++++++++++---- tests/conformance/harness/expectations.py | 3 + tests/conformance/test_conformance.py | 145 +++++++++++- tests/conformance/test_observability.py | 86 ++++++- tests/test_smoke.py | 2 +- tests/unit/test_runtime_errors.py | 111 +++++++++ 9 files changed, 561 insertions(+), 58 deletions(-) diff --git a/openarmature-spec b/openarmature-spec index a910b15..78fe9b4 160000 --- a/openarmature-spec +++ b/openarmature-spec @@ -1 +1 @@ -Subproject commit a910b15367226c1086722765221d729154a07a8b +Subproject commit 78fe9b4b0f53deafb89716ad0ffe6151d175db4a diff --git a/pyproject.toml b/pyproject.toml index 04f4639..16591da 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -36,7 +36,7 @@ Repository = "https://github.com/LunarCommand/openarmature-python" Specification = "https://github.com/LunarCommand/openarmature-spec" [tool.openarmature] -spec_version = "0.8.2" +spec_version = "0.9.0" [dependency-groups] dev = [ diff --git a/src/openarmature/__init__.py b/src/openarmature/__init__.py index c8796a7..74cf11d 100644 --- a/src/openarmature/__init__.py +++ b/src/openarmature/__init__.py @@ -1,4 +1,4 @@ """OpenArmature — workflow framework for LLM pipelines and tool-calling agents.""" __version__ = "0.4.0rc0" -__spec_version__ = "0.8.2" +__spec_version__ = "0.9.0" diff --git a/src/openarmature/graph/compiled.py b/src/openarmature/graph/compiled.py index 73a023d..f222eb6 100644 --- a/src/openarmature/graph/compiled.py +++ b/src/openarmature/graph/compiled.py @@ -24,7 +24,7 @@ import asyncio import time import uuid -from collections.abc import Iterable, Mapping +from collections.abc import Callable, Iterable, Mapping from dataclasses import dataclass, field from typing import Any, cast @@ -130,6 +130,55 @@ def _merge_partial[StateT: State]( ) from e +@dataclass(frozen=True) +class _StepResult[StateT: State]: + """Return shape of the per-step dispatchers + (``_step_function_node`` / ``_step_subgraph_node`` / + ``_step_fan_out_node``) under the proposal-0012 v0.9.0 swap. + + Spec graph-engine §3 step 3 (revised) requires the + ``completed`` event for the just-completed node to fire AFTER + edge evaluation completes — so that edge-resolution failures + (``routing_error``, ``edge_exception``) land on the preceding + node's completed event with ``error`` populated, sharing the + started/completed pair rather than producing a separate event + pair (§6 revised). + + The step dispatchers can't call ``_dispatch_completed`` for + the success path themselves anymore, because the outcome + isn't knowable until edge eval (which lives in ``_invoke``) + runs. Failure-path dispatches (``node_exception`` / + ``reducer_error`` / ``state_validation_error``) still fire + inline inside ``innermost`` — those errors short-circuit + before edge eval can run, and the step function raises out. + + For the success path, the step dispatcher returns the + finalized state plus a closure ``finalize_completed`` that + ``_invoke`` calls AFTER edge eval, passing either ``None`` + (edge eval succeeded → dispatch completed with + ``post_state``) or the edge error (dispatch completed with + ``error`` populated). + + For ``_step_subgraph_node``, the wrapper is transparent per + fixture 013 (no started/completed pair); ``finalize_completed`` + is a no-op closure so edge errors after a subgraph wrapper + propagate silently per proposal 0012's "preceding unit's + pair" framing applied to a unit that never had one. Same for + middleware that short-circuits without invoking ``next``. + """ + + state: StateT + finalize_completed: Callable[[RuntimeGraphError | None], None] + + +def _no_op_finalize(_edge_error: RuntimeGraphError | None) -> None: + """Default ``finalize_completed`` for cases where the step + didn't dispatch a started/completed pair — subgraph wrappers + (transparent per fixture 013) and middleware that short- + circuits without invoking ``next``. Edge errors propagate + silently per proposal 0012 + fixture 013.""" + + @dataclass(frozen=True) class CompiledGraph[StateT: State]: """An immutable, executable graph produced by `GraphBuilder.compile()`. @@ -455,18 +504,19 @@ async def _invoke( if current_namespace in context.resume_skip_set: # Advance edge selection from loaded state. edge = self.edges[current] + skip_target: str | EndSentinel if isinstance(edge, StaticEdge): - target: str | EndSentinel = edge.target + skip_target = edge.target else: try: - target = edge.fn(state) + skip_target = edge.fn(state) except Exception as e: raise EdgeException(source_node=current, cause=e, recoverable_state=state) from e - if target is END: + if skip_target is END: return state - if not isinstance(target, str) or target not in self.nodes: - raise RoutingError(source_node=current, returned=target, recoverable_state=state) - current = target + if not isinstance(skip_target, str) or skip_target not in self.nodes: + raise RoutingError(source_node=current, returned=skip_target, recoverable_state=state) + current = skip_target continue if isinstance(node, FanOutNode): @@ -476,7 +526,7 @@ async def _invoke( # as one parent dispatch (per §9.6) — instance-level # concurrency lives inside the FanOutNode itself. fn_node = cast("FanOutNode[StateT, State]", node) - state = await self._step_fan_out_node(fn_node, current, state, context) + step_result = await self._step_fan_out_node(fn_node, current, state, context) elif isinstance(node, SubgraphNode): # Subgraph wrappers are transparent to the observer protocol # (per fixture 013): no event is dispatched for the wrapper @@ -497,26 +547,55 @@ async def _invoke( # and pass the parent's chain — the inner state class # lives on the subgraph's own CompiledGraph. sub = cast("SubgraphNode[StateT, State]", node) - state = await self._step_subgraph_node(sub, current, state, context) + step_result = await self._step_subgraph_node(sub, current, state, context) else: - state = await self._step_function_node(node, current, state, context) - + step_result = await self._step_function_node(node, current, state, context) + state = step_result.state + + # Per spec graph-engine §3 step 3 (revised in proposal + # 0012 / v0.9.0): the engine MUST dispatch the + # ``completed`` event AFTER edge evaluation completes. + # Edge-resolution failures (``routing_error`` / + # ``edge_exception``) populate the ``error`` field of + # the just-completed node's ``completed`` event, + # sharing the started/completed pair rather than + # producing a separate one (§6 revised). The step + # function deferred its success-case dispatch via + # ``finalize_completed``; we call it below with the + # edge outcome. edge = self.edges[current] + edge_error: RuntimeGraphError | None = None + target: str | EndSentinel | None = None if isinstance(edge, StaticEdge): - target: str | EndSentinel = edge.target + target = edge.target else: try: target = edge.fn(state) except Exception as e: - raise EdgeException(source_node=current, cause=e, recoverable_state=state) from e + edge_error = EdgeException(source_node=current, cause=e, recoverable_state=state) + if edge_error is None: + # Validate the conditional edge's return — undeclared + # target is a ``routing_error``. + if target is not END and not (isinstance(target, str) and target in self.nodes): + edge_error = RoutingError(source_node=current, returned=target, recoverable_state=state) + + # Dispatch the deferred completed event with the edge + # outcome. For function and fan-out nodes this is the + # success/failure dispatch the proposal pinned to + # post-edge-eval timing. For subgraph wrappers (no + # event pair) this is a no-op closure per + # ``_step_subgraph_node``'s `_no_op_finalize` — + # silent propagation per proposal 0012 + fixture 013. + step_result.finalize_completed(edge_error) + if edge_error is not None: + raise edge_error if target is END: return state - - if not isinstance(target, str) or target not in self.nodes: - raise RoutingError(source_node=current, returned=target, recoverable_state=state) - - current = target + # Non-END targets are validated above; mypy/pyright + # don't narrow through the ``edge_error`` path, so + # cast for the assignment. + current = cast("str", target) async def _step_function_node( self, @@ -524,7 +603,7 @@ async def _step_function_node( current: str, state: StateT, context: _InvocationContext, - ) -> StateT: + ) -> _StepResult[StateT]: """Run one function-node step through the middleware chain. Per pipeline-utilities §3, the runtime chain composes: @@ -538,11 +617,22 @@ async def _step_function_node( multiple started/completed event pairs from the engine, each tagged with an incrementing ``attempt_index`` (graph-engine §6). - The chain is built fresh per dispatch so each step has its own - attempt counter. Subgraph isolation per pipeline-utilities §4 is - achieved by NOT including the parent's per-graph middleware when - the subgraph's own ``_step_function_node`` runs — each - CompiledGraph carries its own ``middleware`` tuple. + Per proposal-0012 v0.9.0: the success-case ``completed`` event + for the FINAL successful attempt fires AFTER edge eval, not + inside ``innermost``. Failure-case dispatches + (``node_exception`` / ``reducer_error`` / + ``state_validation_error``) stay inline in ``innermost`` — + those errors short-circuit before edge eval can run, so the + spec's "before the failure propagates" MUST is preserved by + the inline dispatch. + + Returns a :class:`_StepResult` carrying the merged state + + a ``finalize_completed`` closure that ``_invoke`` invokes + after edge eval, passing either ``None`` (edge succeeded) or + the edge error (``RoutingError`` / ``EdgeException``). The + closure dispatches the deferred completed event with the + right shape: ``post_state=merged`` on success, ``error`` + populated on edge-resolution failure. """ step = context.take_step() namespace = context.namespace_prefix + (current,) @@ -553,6 +643,15 @@ async def _step_function_node( # the final successful attempt_index in the checkpoint save. attempt_counter: list[int] = [0] + # Cell holding the FINAL successful attempt's + # (attempt_index, pre_state, merged) — populated by + # ``innermost`` on each successful invocation, overwritten + # if retry middleware re-enters. Stays ``None`` if the chain + # never reached a successful attempt (e.g., middleware + # short-circuited without invoking ``next``, or every + # attempt failed and the chain raised). + deferred_info: list[tuple[int, StateT, StateT] | None] = [None] + async def innermost(s: Any) -> Mapping[str, Any]: # Per pipeline-utilities §5 + graph-engine §6: per-attempt # events use the wrapped §4 error type (NodeException etc.) @@ -601,15 +700,10 @@ async def innermost(s: Any) -> Mapping[str, Any]: ) raise - self._dispatch_completed( - context, - current, - namespace, - step, - s, - post_state=merged, - attempt_index=attempt_index, - ) + # Defer the success-case completed dispatch to + # ``finalize_completed`` per proposal-0012; just + # record the info for the outer scope. + deferred_info[0] = (attempt_index, cast("StateT", s), cast("StateT", merged)) # Return the partial (not the merged state) so middleware sees # the partial-update shape per pipeline-utilities §2. The # engine's canonical merge against the original state happens @@ -673,7 +767,41 @@ async def innermost(s: Any) -> Mapping[str, Any]: attempt_index=max(0, attempt_counter[0] - 1), post_state=merged_outer, ) - return merged_outer + + # Build the deferred-dispatch closure for the success-case + # completed event. ``_invoke`` calls this after edge eval. + info = deferred_info[0] + if info is None: + # Middleware short-circuited without invoking ``next`` — + # no started/completed pair fired. Edge errors after this + # node propagate silently per proposal-0012 + fixture-013 + # framing (preceding unit emitted no pair to share). + return _StepResult(state=merged_outer, finalize_completed=_no_op_finalize) + final_attempt_index, final_pre_state, final_merged = info + + def finalize_completed(edge_error: RuntimeGraphError | None) -> None: + if edge_error is None: + self._dispatch_completed( + context, + current, + namespace, + step, + final_pre_state, + post_state=final_merged, + attempt_index=final_attempt_index, + ) + else: + self._dispatch_completed( + context, + current, + namespace, + step, + final_pre_state, + error=edge_error, + attempt_index=final_attempt_index, + ) + + return _StepResult(state=merged_outer, finalize_completed=finalize_completed) async def _step_subgraph_node( self, @@ -681,7 +809,7 @@ async def _step_subgraph_node( current: str, state: StateT, context: _InvocationContext, - ) -> StateT: + ) -> _StepResult[StateT]: """Run one subgraph-as-node step through the parent's middleware chain. Per pipeline-utilities §4: the parent's per-graph middleware plus @@ -693,6 +821,16 @@ async def _step_subgraph_node( No started/completed events fire for the wrapper itself; the events come from the subgraph's internal node executions (per fixture 013). + + Per proposal-0012 v0.9.0 + spec coordination: edge errors + AFTER a transparent subgraph wrapper propagate to the caller + as ``RuntimeGraphError`` per §4 WITHOUT an associated + completed event — the wrapper has no started/completed pair + to share, and proposal 0012's "preceding node's pair" MUST + is vacuous (not violated) when the preceding unit emitted + no pair. The :class:`_StepResult` returned here uses + :func:`_no_op_finalize` so the outer ``_invoke`` call to + ``finalize_completed(edge_error)`` is a no-op. """ async def innermost(s: Any) -> Mapping[str, Any]: @@ -737,7 +875,8 @@ async def innermost(s: Any) -> Mapping[str, Any]: _reset_namespace_prefix(namespace_token) _reset_active_dispatch(dispatch_token) _reset_active_observers(observers_token) - return _merge_partial(state, final_partial, self.reducers, current) + merged = _merge_partial(state, final_partial, self.reducers, current) + return _StepResult(state=merged, finalize_completed=_no_op_finalize) async def _step_fan_out_node( self, @@ -745,7 +884,7 @@ async def _step_fan_out_node( current: str, state: StateT, context: _InvocationContext, - ) -> StateT: + ) -> _StepResult[StateT]: """Run one fan-out-as-node step through the parent's middleware chain. Per pipeline-utilities §9.6: the parent's per-graph + per-node @@ -757,6 +896,12 @@ async def _step_fan_out_node( ``fan_out_index`` populated. Raw exceptions escaping the chain become NodeException per §4. + + Per proposal-0012 v0.9.0: the fan-out's success-case + completed event fires AFTER edge eval (mirrors + ``_step_function_node``). Failure-path dispatches stay + inline; the success-case is deferred via the returned + :class:`_StepResult`. """ step = context.take_step() namespace = context.namespace_prefix + (current,) @@ -768,6 +913,11 @@ async def _step_fan_out_node( # hardcoded 0. attempt_counter: list[int] = [0] + # Cell holding the FINAL successful attempt's + # (attempt_index, pre_state, merged); see same comment in + # ``_step_function_node``. + deferred_info: list[tuple[int, StateT, StateT] | None] = [None] + async def innermost(s: Any) -> Mapping[str, Any]: attempt_index = attempt_counter[0] attempt_counter[0] += 1 @@ -802,15 +952,9 @@ async def innermost(s: Any) -> Mapping[str, Any]: ) raise - self._dispatch_completed( - context, - current, - namespace, - step, - s, - post_state=merged, - attempt_index=attempt_index, - ) + # Defer the success-case completed dispatch per + # proposal-0012; record the info for the outer scope. + deferred_info[0] = (attempt_index, cast("StateT", s), cast("StateT", merged)) return partial finally: _reset_attempt_index(attempt_token) @@ -861,7 +1005,35 @@ async def innermost(s: Any) -> Mapping[str, Any]: attempt_index=max(0, attempt_counter[0] - 1), post_state=merged_outer, ) - return merged_outer + + info = deferred_info[0] + if info is None: + return _StepResult(state=merged_outer, finalize_completed=_no_op_finalize) + final_attempt_index, final_pre_state, final_merged = info + + def finalize_completed(edge_error: RuntimeGraphError | None) -> None: + if edge_error is None: + self._dispatch_completed( + context, + current, + namespace, + step, + final_pre_state, + post_state=final_merged, + attempt_index=final_attempt_index, + ) + else: + self._dispatch_completed( + context, + current, + namespace, + step, + final_pre_state, + error=edge_error, + attempt_index=final_attempt_index, + ) + + return _StepResult(state=merged_outer, finalize_completed=finalize_completed) @staticmethod def _dispatch_started( diff --git a/tests/conformance/harness/expectations.py b/tests/conformance/harness/expectations.py index 6f03853..640bbd7 100644 --- a/tests/conformance/harness/expectations.py +++ b/tests/conformance/harness/expectations.py @@ -48,6 +48,9 @@ class GraphEngineExpected(_ForbidExtras): observer_events: Any = None delivery_order: list[dict[str, Any]] | None = None observer_event_invariants: dict[str, Any] | None = None + # 020 — proposal-0012 fixture: assertions about edge-resolution + # failure event shapes. Permissive dict until Phase 1. + invariants: dict[str, Any] | None = None # 015 — invoke() returns normally; obs_raiser's exceptions surface to # warnings rather than propagate. no_propagated_error: bool | None = None diff --git a/tests/conformance/test_conformance.py b/tests/conformance/test_conformance.py index a78b533..904cb54 100644 --- a/tests/conformance/test_conformance.py +++ b/tests/conformance/test_conformance.py @@ -7,6 +7,7 @@ from __future__ import annotations +from collections.abc import Mapping from pathlib import Path from typing import Any, cast @@ -14,12 +15,18 @@ import yaml from openarmature.graph import ( + END, CompileError, + EdgeException, + EndSentinel, + GraphBuilder, NodeException, RoutingError, RuntimeGraphError, + State, SubscribedObserver, ) +from openarmature.graph.events import NodeEvent from openarmature.graph.observer import Observer from .adapter import ( @@ -71,7 +78,7 @@ def _fixture_id(path: Path) -> str: ) -def _unsupported_directive(spec: dict[str, Any]) -> str | None: +def _unsupported_directive(spec: Mapping[str, Any]) -> str | None: """Return the first node directive the legacy adapter can't translate, or None if every node uses one of the directives it handles. Walks both the top-level graph and an optional inner ``subgraph`` block.""" @@ -162,11 +169,34 @@ def _compile_subgraphs_map( async def test_runtime_fixture(fixture_path: Path) -> None: spec = _load(fixture_path) + # ``cases:`` form (e.g., 020-observer-edge-error-events): each entry + # is a self-contained per-case spec. Iterate; treat each case as a + # standalone fixture body. Wrapping AssertionError with the case + # name keeps failure messages locatable. + # Fixture 020 uses edge-condition `callable:` directives + # (`state_field_read`, `edge_raises`) that are unique to this fixture + # and don't fit the generic adapter DSL. Custom driver below. + if fixture_path.stem == "020-observer-edge-error-events": + await _run_fixture_020(spec) + return + + if "cases" in spec: + for case in cast("list[dict[str, Any]]", spec["cases"]): + try: + await _run_runtime_case(case, fixture_path.stem) + except AssertionError as e: + raise AssertionError(f"case {case.get('name')!r}: {e}") from e + return + + await _run_runtime_case(spec, fixture_path.stem) + + +async def _run_runtime_case(spec: Mapping[str, Any], fixture_id: str) -> None: # Skip fixtures whose nodes use directives the legacy adapter doesn't # translate (fan_out, flaky variants, calls_llm, etc.). Each directive # is gated to the phase that lands its runtime support. if (hit := _unsupported_directive(spec)) is not None: - pytest.skip(f"{fixture_path.stem}: unsupported node directive {hit}") + pytest.skip(f"{fixture_id}: unsupported node directive {hit}") # Subgraph fixtures (006, 011, 013) declare an inner subgraph via the # singular `subgraph:` key. Fixture 019 introduces the plural `subgraphs:` @@ -292,6 +322,117 @@ async def test_runtime_fixture(fixture_path: Path) -> None: ) +# --------------------------------------------------------------------------- +# Fixture 020 — observer edge-error events (proposal 0012 / spec v0.9.0) +# +# Two sub-cases verifying §3 step 3 (revised) + §6 (revised): edge-resolution +# failures (routing_error, edge_exception) land on the preceding node's +# completed event with `error` populated, sharing the started/completed pair +# rather than producing a separate event pair. +# +# Custom driver (rather than the generic harness) because fixture 020's +# edge-condition `callable:` directives (`state_field_read`, `edge_raises`) +# don't match the adapter DSL's `if_field/equals/then/else` shape — these +# directives are spec-defined just for this fixture's two sub-cases. The +# semantic intent is captured directly here. +# --------------------------------------------------------------------------- + + +async def _run_fixture_020(spec: Mapping[str, Any]) -> None: + cases = cast("list[dict[str, Any]]", spec["cases"]) + for case in cases: + case_name = cast("str", case["name"]) + try: + await _run_fixture_020_case(case) + except AssertionError as e: + raise AssertionError(f"case {case_name!r}: {e}") from e + + +async def _run_fixture_020_case(case: Mapping[str, Any]) -> None: + """Build a two-node graph (a → b) with the case's edge-condition + directive translated to a Python edge function, then assert the + proposal-0012 contract on the captured observer events.""" + + class FixtureState(State): + x: int = 0 + + received: list[NodeEvent] = [] + + async def observer(event: NodeEvent) -> None: + received.append(event) + + async def node_a(_state: Any) -> dict[str, Any]: + return {"x": 1} + + async def node_b(_state: Any) -> dict[str, Any]: + return {"x": 2} + + cond = cast("dict[str, Any]", case["edges"][0]["condition"]) + callable_name = cast("str", cond["callable"]) + if callable_name == "state_field_read": + # Per spec proposal 0012 fixture 020: the edge function returns + # the value from initial_state's named field. The fixture's case + # initial_state populates that field with a string that is NOT a + # declared node name in the graph, so the engine raises + # ``RoutingError``. We reproduce the semantic via a closure + # over the initial_state value. + initial_state_dict = cast("dict[str, Any]", case.get("initial_state", {})) + field_name = cast("str", cond["field"]) + target_value = cast("str", initial_state_dict[field_name]) + + def edge_fn(_state: Any) -> str | EndSentinel: + return target_value + elif callable_name == "edge_raises": + message = cast("str", cond.get("message", "edge raised")) + + def edge_fn(_state: Any) -> str | EndSentinel: + raise RuntimeError(message) + else: + raise AssertionError(f"fixture 020: unknown condition callable {callable_name!r}") + + g = ( + GraphBuilder(FixtureState) + .add_node("a", node_a) + .add_node("b", node_b) + .add_conditional_edge("a", edge_fn) + .add_edge("b", END) + .set_entry("a") + .compile() + ) + g.attach_observer(observer) + + expected_error = cast("dict[str, Any]", case["expected_error"]) + expected_category = cast("str", expected_error["category"]) + + with pytest.raises(RuntimeGraphError) as excinfo: + await g.invoke(FixtureState()) + await g.drain() + + err = excinfo.value + assert err.category == expected_category + if expected_category == "routing_error": + assert isinstance(err, RoutingError) + elif expected_category == "edge_exception": + assert isinstance(err, EdgeException) + + # Per the revised §6 contract: edge-resolution failures share the + # preceding node's started/completed pair. Node a fires exactly one + # started + one completed event; node b never fires. + a_events = [e for e in received if e.node_name == "a"] + b_events = [e for e in received if e.node_name == "b"] + assert len(a_events) == 2, f"expected 2 events for node a; got {len(a_events)}" + assert b_events == [], f"node b MUST never fire events on edge-resolution failure; got {b_events}" + + started, completed = a_events + assert started.phase == "started" + assert completed.phase == "completed" + assert completed.post_state is None, ( + "completed event MUST have post_state absent when edge resolution fails" + ) + assert completed.error is not None + assert completed.error.category == expected_category + + # --------------------------------------------------------------------------- # 007 compile-errors: one parametrized case per entry in the `cases:` table. # --------------------------------------------------------------------------- diff --git a/tests/conformance/test_observability.py b/tests/conformance/test_observability.py index 8fd5d3e..abebc3d 100644 --- a/tests/conformance/test_observability.py +++ b/tests/conformance/test_observability.py @@ -70,6 +70,7 @@ "001-otel-basic-trace", "002-otel-subgraph-hierarchy", "003-otel-error-status", + "004-otel-routing-error-attribution", "005-otel-llm-provider-span-nested", "007-otel-retry-attempt-spans", "008-otel-detached-trace-mode", @@ -80,11 +81,6 @@ _DEFERRED_FIXTURES: dict[str, str] = { - "004-otel-routing-error-attribution": ( - "Awaiting proposal 0012 — routing-error attribution requires the §3/§6 " - "ordering swap (completed dispatch after edge eval) so RoutingError lands " - "on the preceding node's completed event. Lands in PR-C.1 once v0.9.0 ships." - ), "006-otel-fan-out-instance-attribution": ( "Needs non-detached fan-out per-instance dispatch span synthesis + " "FanOutConfig metadata surfacing per spec §5.4 (current observer opens one " @@ -139,6 +135,8 @@ async def test_observability_fixture(fixture_path: Path) -> None: await _run_fixture_002(spec) elif fixture_id == "003-otel-error-status": await _run_fixture_003(spec) + elif fixture_id == "004-otel-routing-error-attribution": + await _run_fixture_004(spec) elif fixture_id == "005-otel-llm-provider-span-nested": await _run_fixture_005(spec) elif fixture_id == "007-otel-retry-attempt-spans": @@ -374,6 +372,84 @@ async def _run_fixture_003(spec: Mapping[str, Any]) -> None: ) +# --------------------------------------------------------------------------- +# Fixture 004 — routing-error attribution (proposal 0012 / spec v0.9.0) +# --------------------------------------------------------------------------- + + +async def _run_fixture_004(spec: Mapping[str, Any]) -> None: + """Spec §4.2 + spec v0.9.0 / proposal 0012: routing errors land on + the preceding node's ``completed`` event with ``error`` populated + (sharing the started/completed pair rather than producing a + separate one). The OTel observer's existing + ``_handle_completed`` ERROR-mapping path picks this up + automatically — no observer-side change needed for the swap. + + Driver verifies: the ``pick`` node's span ends ERROR with + ``status_description == "routing_error"``, an ``exception`` + event recorded, and the ``openarmature.error.category`` + attribute. No span for the edge function (no ``edge_spans``) + per §4.2's "edge logic folded into the preceding node span" + framing.""" + from opentelemetry.trace import StatusCode + + from openarmature.graph import RuntimeGraphError + + observer, exporter = _build_observer() + trace_log: list[str] = [] + built = build_graph(spec, trace=trace_log) + compiled = built.builder.compile() + compiled.attach_observer(observer) + initial_state = built.initial_state(spec.get("initial_state", {})) + with pytest.raises(RuntimeGraphError) as excinfo: + await compiled.invoke(initial_state) + assert excinfo.value.category == "routing_error" + await compiled.drain() + observer.shutdown() + spans = exporter.get_finished_spans() + + by_name = {s.name: s for s in spans} + + pick = by_name.get("pick") + assert pick is not None + assert pick.status.status_code == StatusCode.ERROR, ( + f"preceding node 'pick' span MUST be ERROR; got {pick.status.status_code}" + ) + assert pick.status.description == "routing_error", ( + f"preceding node 'pick' span status_description MUST be 'routing_error'; " + f"got {pick.status.description!r}" + ) + pick_attrs = dict(pick.attributes or {}) + assert pick_attrs.get("openarmature.error.category") == "routing_error" + # Exception event recorded on the span via record_exception. + exception_events = [e for e in pick.events if e.name == "exception"] + event_names = [e.name for e in pick.events] + assert len(exception_events) >= 1, ( + f"'pick' MUST have at least one 'exception' event recorded; got {event_names}" + ) + + # Per fixture 004's "no_edge_spans: true" — the edge function + # itself does not produce a separate span; the routing error is + # folded into the preceding node's span. + edge_span_names = {"edge", "openarmature.edge", "edge_function"} + edge_spans = [s for s in spans if s.name in edge_span_names] + assert edge_spans == [], ( + f"there MUST be no separate edge-function spans per §4.2; got {[s.name for s in edge_spans]}" + ) + + # Unreachable nodes never fire spans (they were unreached). + for unreachable in ("unreachable_a", "unreachable_b"): + assert unreachable not in by_name, f"{unreachable!r} MUST not produce a span — never reached" + + # Invocation span ends ERROR per the §4.2 invocation-status + # propagation contract (PR-C review fix). + inv = by_name.get("openarmature.invocation") + assert inv is not None + assert inv.status.status_code == StatusCode.ERROR, ( + f"invocation span MUST end ERROR when a child errors; got {inv.status.status_code}" + ) + + # --------------------------------------------------------------------------- # Fixture 007 — retry attempt spans # --------------------------------------------------------------------------- diff --git a/tests/test_smoke.py b/tests/test_smoke.py index 1329099..cd55887 100644 --- a/tests/test_smoke.py +++ b/tests/test_smoke.py @@ -3,4 +3,4 @@ def test_package_versions() -> None: assert openarmature.__version__ == "0.4.0rc0" - assert openarmature.__spec_version__ == "0.8.2" + assert openarmature.__spec_version__ == "0.9.0" diff --git a/tests/unit/test_runtime_errors.py b/tests/unit/test_runtime_errors.py index e780687..0b7f213 100644 --- a/tests/unit/test_runtime_errors.py +++ b/tests/unit/test_runtime_errors.py @@ -149,3 +149,114 @@ def project_out( assert err.node_name == "sub" assert isinstance(err.__cause__, RuntimeError) assert str(err.__cause__) == "project_in boom" + + +# --------------------------------------------------------------------------- +# Spec graph-engine v0.9.0 (proposal 0012): edge-resolution failures land on +# the preceding node's `completed` event with `error` populated, sharing the +# started/completed pair rather than producing a separate event pair. +# --------------------------------------------------------------------------- + + +async def test_routing_error_lands_on_preceding_node_completed_event() -> None: + """Per §3 step 3 (revised) + §6 (revised): a `routing_error` from a + conditional edge that returns an undeclared target lands on the + preceding node's `completed` event with `error` populated, NOT in a + separate event pair. The downstream node never fires events.""" + from openarmature.graph import RoutingError + from openarmature.graph.events import NodeEvent + + received: list[NodeEvent] = [] + + async def observer(event: NodeEvent) -> None: + received.append(event) + + async def node_a(_state: Any) -> dict[str, Any]: + return {"score": 1} + + async def node_b(_state: Any) -> dict[str, Any]: + return {"score": 99} + + def routing_to_nowhere(_state: Any) -> str | EndSentinel: + # 'nonexistent_node' is not declared — engine raises RoutingError. + return "nonexistent_node" + + g = ( + GraphBuilder(S) + .add_node("a", node_a) + .add_node("b", node_b) + .add_conditional_edge("a", routing_to_nowhere) + .add_edge("b", END) + .set_entry("a") + .compile() + ) + g.attach_observer(observer) + + with pytest.raises(RoutingError): + await g.invoke(S()) + await g.drain() + + # Exactly one started + one completed pair for node a; no events for b. + a_events = [e for e in received if e.node_name == "a"] + b_events = [e for e in received if e.node_name == "b"] + assert len(a_events) == 2, f"expected 2 events for node a (started + completed); got {len(a_events)}" + assert b_events == [], f"node b MUST never fire events on routing error; got {b_events}" + + started, completed = a_events + assert started.phase == "started" + assert completed.phase == "completed" + # Completed event carries the routing error, not a success post_state. + assert completed.post_state is None, ( + "completed event MUST have post_state absent when edge resolution fails" + ) + assert completed.error is not None + assert completed.error.category == "routing_error" + + +async def test_edge_exception_lands_on_preceding_node_completed_event() -> None: + """Per §3 step 3 (revised) + §6 (revised): an `edge_exception` from a + conditional edge function raising lands on the preceding node's + `completed` event with `error` populated, NOT in a separate event + pair. The downstream node never fires events.""" + from openarmature.graph.events import NodeEvent + + received: list[NodeEvent] = [] + + async def observer(event: NodeEvent) -> None: + received.append(event) + + async def node_a(_state: Any) -> dict[str, Any]: + return {"score": 1} + + async def node_b(_state: Any) -> dict[str, Any]: + return {"score": 99} + + def raising_edge(_state: Any) -> str | EndSentinel: + raise RuntimeError("edge boom") + + g = ( + GraphBuilder(S) + .add_node("a", node_a) + .add_node("b", node_b) + .add_conditional_edge("a", raising_edge) + .add_edge("b", END) + .set_entry("a") + .compile() + ) + g.attach_observer(observer) + + with pytest.raises(EdgeException): + await g.invoke(S()) + await g.drain() + + a_events = [e for e in received if e.node_name == "a"] + b_events = [e for e in received if e.node_name == "b"] + assert len(a_events) == 2 + assert b_events == [] + + started, completed = a_events + assert started.phase == "started" + assert completed.phase == "completed" + assert completed.post_state is None + assert completed.error is not None + assert completed.error.category == "edge_exception"