Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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 = [
Expand Down
2 changes: 1 addition & 1 deletion src/openarmature/__init__.py
Original file line number Diff line number Diff line change
@@ -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"
266 changes: 219 additions & 47 deletions src/openarmature/graph/compiled.py

Large diffs are not rendered by default.

3 changes: 3 additions & 0 deletions tests/conformance/harness/expectations.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
145 changes: 143 additions & 2 deletions tests/conformance/test_conformance.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand Down Expand Up @@ -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."""
Expand Down Expand Up @@ -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:`
Expand Down Expand Up @@ -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.
# ---------------------------------------------------------------------------
Expand Down
86 changes: 81 additions & 5 deletions tests/conformance/test_observability.py
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand All @@ -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 "
Expand Down Expand Up @@ -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":
Expand Down Expand Up @@ -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
# ---------------------------------------------------------------------------
Expand Down
2 changes: 1 addition & 1 deletion tests/test_smoke.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Loading
Loading