graph: phase 6.1 PR-C.1 — proposal 0012 + v0.9.0

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.

Author: chris-colinsky

openarmature-spec

@@ -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 = [

pyproject.toml

@@ -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"

src/openarmature/__init__.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

src/openarmature/graph/compiled.py

@@ -7,19 +7,26 @@
 
 from __future__ import annotations
 
+from collections.abc import Mapping
 from pathlib import Path
 from typing import Any, cast
 
 import pytest
 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.
 # ---------------------------------------------------------------------------

tests/conformance/harness/expectations.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
 # ---------------------------------------------------------------------------

tests/conformance/test_conformance.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"