diff --git a/.github/codeql/codeql-config.yml b/.github/codeql/codeql-config.yml index 6747e29..909e9d4 100644 --- a/.github/codeql/codeql-config.yml +++ b/.github/codeql/codeql-config.yml @@ -48,3 +48,23 @@ query-filters: # so dropping the CodeQL rule doesn't lose signal. - exclude: id: py/unused-import + # ``py/unsafe-cyclic-import`` flags the textual import shape + # without honoring ``if TYPE_CHECKING:`` gates. The two cases + # in this codebase are mutual-typing pairs: + # + # - ``graph/compiled.py`` and ``graph/fan_out.py`` reference + # each other's types only via TYPE_CHECKING blocks + # (``compiled`` imports ``FanOutNode`` for type annotations; + # ``fan_out`` imports ``CompiledGraph`` for the + # ``FanOutConfig.subgraph`` field annotation). Both sides use + # ``from __future__ import annotations`` so annotations are + # strings; no runtime cycle exists. + # + # The canonical Python workaround for typed-module pairs is + # exactly this TYPE_CHECKING gating. Removing either side + # breaks pyright's type resolution for generics across the + # boundary. Pyright's ``reportImportCycles`` already covers the + # genuine runtime-cycle cases at type-check time, so dropping + # the CodeQL rule doesn't lose signal. + - exclude: + id: py/unsafe-cyclic-import diff --git a/openarmature-spec b/openarmature-spec index 78fe9b4..ff86945 160000 --- a/openarmature-spec +++ b/openarmature-spec @@ -1 +1 @@ -Subproject commit 78fe9b4b0f53deafb89716ad0ffe6151d175db4a +Subproject commit ff86945747c9d4767e5ededad62bab1c9c4e244a diff --git a/pyproject.toml b/pyproject.toml index 16591da..ef9c6cb 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.9.0" +spec_version = "0.10.0" [dependency-groups] dev = [ diff --git a/src/openarmature/__init__.py b/src/openarmature/__init__.py index 74cf11d..b0e8dc4 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.9.0" +__spec_version__ = "0.10.0" diff --git a/src/openarmature/graph/compiled.py b/src/openarmature/graph/compiled.py index f222eb6..c830e16 100644 --- a/src/openarmature/graph/compiled.py +++ b/src/openarmature/graph/compiled.py @@ -21,12 +21,26 @@ `cast(MyState, ...)` at the call site. """ +from __future__ import annotations + import asyncio import time import uuid from collections.abc import Callable, Iterable, Mapping from dataclasses import dataclass, field -from typing import Any, cast +from typing import TYPE_CHECKING, Any, cast + +if TYPE_CHECKING: + # ``FanOutNode`` lives in ``.fan_out`` which has a TYPE_CHECKING + # back-reference to ``CompiledGraph`` here. Importing at module + # top would create a textual cycle CodeQL's + # ``py/cyclic-import`` rule flags (no runtime issue — + # ``fan_out``'s ``compiled`` import is itself TYPE_CHECKING-gated + # — but the static analyzer doesn't see that). Type annotations + # use the string form via ``from __future__ import annotations``; + # runtime use (the ``isinstance`` check in ``_invoke``) imports + # lazily inside the function. + from .fan_out import FanOutNode from pydantic import ValidationError @@ -67,8 +81,7 @@ RuntimeGraphError, StateValidationError, ) -from .events import NodeEvent -from .fan_out import FanOutNode +from .events import FanOutEventConfig, NodeEvent from .middleware import ChainCall, Middleware, compose_chain from .nodes import Node from .observer import ( @@ -519,6 +532,12 @@ async def _invoke( current = skip_target continue + # Lazy import: keeps the textual cycle off the module + # graph (``fan_out`` has a TYPE_CHECKING back-reference + # to this module). Function-scope import is cheap once + # cached; this branch fires once per fan-out step. + from .fan_out import FanOutNode # noqa: PLC0415 + if isinstance(node, FanOutNode): # Fan-out nodes are recognized as a distinct node type # per pipeline-utilities §9. Dispatched through @@ -913,6 +932,100 @@ async def _step_fan_out_node( # hardcoded 0. attempt_counter: list[int] = [0] + # Resolve the fan-out config eagerly so the resolved values + # ride on every fan-out node event (per spec proposal 0013, + # v0.10.0: ``fan_out_config`` is populated on fan-out node + # events including retried attempts). For ``items_field`` + # mode the count is ``len(parent_state.)``; for + # ``count`` mode it's ``_resolve_count``. ``_resolve_concurrency`` + # is pure regardless. Repeating these inside + # ``FanOutNode.run_with_context`` is cheap and matches the + # values surfaced here. + # Lazy import: function-scope to avoid a module-top + # textual cycle CodeQL flags. ``fan_out`` has a + # TYPE_CHECKING back-reference to this module, so the + # static-analyzer view of an importable cycle goes away + # when the engine doesn't reach into ``fan_out`` at module + # load time. Fires once per fan-out step. + from .fan_out import _resolve_concurrency, _resolve_count # noqa: PLC0415 + + # Resolver failures (callable count/concurrency raising, + # ``getattr`` on a malformed state, etc.) used to land inside + # ``innermost``'s ``except Exception → NodeException`` block + # below and produce a started/completed event pair via the + # surrounding dispatches. Hoisting resolution out of + # ``run_with_context`` for the eager ``FanOutEventConfig`` + # build moved them past that scope, so re-establish the + # contract here: surface a started/completed pair with + # ``fan_out_config=None`` (we never built one) and raise as + # ``NodeException``. + try: + if node.config.items_field is not None: + items_attr: Any = getattr(state, node.config.items_field, []) + if not isinstance(items_attr, list): + raise NodeException( + node_name=current, + cause=TypeError(f"items_field {node.config.items_field!r} is not a list at runtime"), + recoverable_state=state, + ) + item_count = len(cast("list[Any]", items_attr)) + else: + item_count = _resolve_count(current, node.config, state) + concurrency_resolved: int | None = _resolve_concurrency(current, node.config, state) + fan_out_event_config = FanOutEventConfig( + item_count=item_count, + concurrency=concurrency_resolved, + error_policy=node.config.error_policy, + parent_node_name=current, + ) + except NodeException as resolution_error: + self._dispatch_started( + context, + current, + namespace, + step, + state, + attempt_index=0, + fan_out_config=None, + ) + self._dispatch_completed( + context, + current, + namespace, + step, + state, + error=resolution_error, + attempt_index=0, + fan_out_config=None, + ) + raise + except Exception as resolution_error: + wrapped = NodeException( + node_name=current, + cause=resolution_error, + recoverable_state=state, + ) + self._dispatch_started( + context, + current, + namespace, + step, + state, + attempt_index=0, + fan_out_config=None, + ) + self._dispatch_completed( + context, + current, + namespace, + step, + state, + error=wrapped, + attempt_index=0, + fan_out_config=None, + ) + raise wrapped from resolution_error + # Cell holding the FINAL successful attempt's # (attempt_index, pre_state, merged); see same comment in # ``_step_function_node``. @@ -923,12 +1036,32 @@ async def innermost(s: Any) -> Mapping[str, Any]: attempt_counter[0] += 1 attempt_token = _set_attempt_index(attempt_index) try: - self._dispatch_started(context, current, namespace, step, s, attempt_index=attempt_index) + self._dispatch_started( + context, + current, + namespace, + step, + s, + attempt_index=attempt_index, + fan_out_config=fan_out_event_config, + ) try: - partial = await node.run_with_context(s, context) + partial = await node.run_with_context( + s, + context, + pre_resolved_count=item_count, + pre_resolved_concurrency=(concurrency_resolved,), + ) except RuntimeGraphError as e: self._dispatch_completed( - context, current, namespace, step, s, error=e, attempt_index=attempt_index + context, + current, + namespace, + step, + s, + error=e, + attempt_index=attempt_index, + fan_out_config=fan_out_event_config, ) raise except Exception as e: @@ -941,6 +1074,7 @@ async def innermost(s: Any) -> Mapping[str, Any]: s, error=wrapped, attempt_index=attempt_index, + fan_out_config=fan_out_event_config, ) raise wrapped from e @@ -948,7 +1082,14 @@ async def innermost(s: Any) -> Mapping[str, Any]: merged = _merge_partial(s, partial, self.reducers, current) except (ReducerError, StateValidationError) as e: self._dispatch_completed( - context, current, namespace, step, s, error=e, attempt_index=attempt_index + context, + current, + namespace, + step, + s, + error=e, + attempt_index=attempt_index, + fan_out_config=fan_out_event_config, ) raise @@ -1021,6 +1162,7 @@ def finalize_completed(edge_error: RuntimeGraphError | None) -> None: final_pre_state, post_state=final_merged, attempt_index=final_attempt_index, + fan_out_config=fan_out_event_config, ) else: self._dispatch_completed( @@ -1031,6 +1173,7 @@ def finalize_completed(edge_error: RuntimeGraphError | None) -> None: final_pre_state, error=edge_error, attempt_index=final_attempt_index, + fan_out_config=fan_out_event_config, ) return _StepResult(state=merged_outer, finalize_completed=finalize_completed) @@ -1044,6 +1187,7 @@ def _dispatch_started( pre_state: State, *, attempt_index: int = 0, + fan_out_config: FanOutEventConfig | None = None, ) -> None: _dispatch( context, @@ -1058,6 +1202,7 @@ def _dispatch_started( parent_states=context.parent_states_prefix, attempt_index=attempt_index, fan_out_index=context.fan_out_index, + fan_out_config=fan_out_config, ), ) @@ -1072,6 +1217,7 @@ def _dispatch_completed( post_state: State | None = None, error: RuntimeGraphError | None = None, attempt_index: int = 0, + fan_out_config: FanOutEventConfig | None = None, ) -> None: _dispatch( context, @@ -1086,6 +1232,7 @@ def _dispatch_completed( parent_states=context.parent_states_prefix, attempt_index=attempt_index, fan_out_index=context.fan_out_index, + fan_out_config=fan_out_config, ), ) diff --git a/src/openarmature/graph/events.py b/src/openarmature/graph/events.py index 205fa1d..67595c4 100644 --- a/src/openarmature/graph/events.py +++ b/src/openarmature/graph/events.py @@ -16,6 +16,52 @@ from .state import State +@dataclass(frozen=True) +class FanOutEventConfig: + """Spec §6 + §5.4 (per spec proposal 0013, v0.10.0): + fan-out node events carry the resolved configuration so backend + observers can attribute the fan-out node span (``item_count`` / + ``concurrency`` / ``error_policy``) and synthesize per-instance + spans with the right ``parent_node_name``. + + Populated ONLY on ``started`` and ``completed`` events for a + fan-out node itself (partition by node type, not event category — + INCLUDES retried attempts of a fan-out node when retry middleware + wraps it). All other events leave ``NodeEvent.fan_out_config`` + null. + + Field shapes: + + - ``item_count`` — non-negative int. The resolved instance count + per pipeline-utilities §9 (matches ``count_field`` value when + configured; matches ``len(items_field)`` in items_field mode). + - ``concurrency`` — positive int OR ``None`` (unbounded). Per + pipeline-utilities §9.2: zero or negative is rejected at config + resolution time as ``fan_out_invalid_concurrency``. Backend + mappings translate ``None`` to a sentinel at the attribute layer + (e.g., ``openarmature.fan_out.concurrency = 0`` per + observability §5.4) — that translation is observer-internal, + not engine-internal. + - ``error_policy`` — one of ``"fail_fast"`` or ``"collect"`` per + pipeline-utilities §9.4. + - ``parent_node_name`` — the fan-out node's name in the parent + graph. Carried here for caching by backend observers when + attributing per-instance spans (§5.4 mandates + ``openarmature.fan_out.parent_node_name`` on per-instance spans; + the engine surfaces the name once on the fan-out node's started + event, the observer caches and applies on every per-instance + span it synthesizes). + + All four fields MUST be present when ``fan_out_config`` is + populated. Only ``concurrency`` is nullable. + """ + + item_count: int + concurrency: int | None + error_policy: str + parent_node_name: str + + @dataclass(frozen=True) class NodeEvent: """A single node-boundary event delivered to observers. @@ -55,6 +101,9 @@ class NodeEvent: retries. `0` for nodes not wrapped by retry middleware. - `fan_out_index` is the 0-based index of this fan-out instance among its siblings. `None` for nodes not inside a fan-out. + - `fan_out_config` carries resolved fan-out configuration on events + from a fan-out NODE itself (per spec proposal 0013, v0.10.0). See + :class:`FanOutEventConfig`. ``None`` on every other event. Invariants: - On `started` events, `post_state` and `error` MUST both be None. @@ -72,6 +121,7 @@ class NodeEvent: parent_states: tuple[State, ...] attempt_index: int = 0 fan_out_index: int | None = None + fan_out_config: FanOutEventConfig | None = None -__all__ = ["NodeEvent"] +__all__ = ["FanOutEventConfig", "NodeEvent"] diff --git a/src/openarmature/graph/fan_out.py b/src/openarmature/graph/fan_out.py index d97e278..97790c4 100644 --- a/src/openarmature/graph/fan_out.py +++ b/src/openarmature/graph/fan_out.py @@ -107,6 +107,9 @@ async def run_with_context( self, state: ParentT, context: _InvocationContext, + *, + pre_resolved_count: int | None = None, + pre_resolved_concurrency: tuple[int | None] | None = None, ) -> Mapping[str, Any]: """Execute the fan-out and return the merged partial update. @@ -114,9 +117,18 @@ async def run_with_context( build per-instance states, run concurrently with the configured error policy, fan-in collected/extra fields, write count_field and errors_field if configured. + + ``pre_resolved_count`` / ``pre_resolved_concurrency`` are the + proposal-0013 v0.10.0 hooks: when the engine has already + resolved the config eagerly to populate + ``NodeEvent.fan_out_config`` for the fan-out node's events, + it passes the resolved values in so callable resolvers + aren't invoked twice. ``pre_resolved_concurrency`` is wrapped + in a 1-tuple to disambiguate "caller passed ``None`` + (unbounded)" from "caller didn't pass anything." """ cfg = self.config - instance_states = _build_instance_states(self.name, cfg, state) + instance_states = _build_instance_states(self.name, cfg, state, pre_resolved_count=pre_resolved_count) instance_count = len(instance_states) if instance_count == 0: @@ -127,7 +139,10 @@ async def run_with_context( # reducer absorbs as a no-op). return _empty_fan_out_partial(cfg) - max_concurrency = _resolve_concurrency(self.name, cfg, state) + if pre_resolved_concurrency is not None: + max_concurrency = pre_resolved_concurrency[0] + else: + max_concurrency = _resolve_concurrency(self.name, cfg, state) # Per-instance task: build the instance_middleware chain, run # the subgraph against it, and return the per-instance partial @@ -187,6 +202,8 @@ def _build_instance_states( node_name: str, cfg: FanOutConfig, parent_state: Any, + *, + pre_resolved_count: int | None = None, ) -> list[Any]: """Project parent state to per-instance subgraph states. @@ -194,6 +211,13 @@ def _build_instance_states( - items_field mode: one instance per item, item_field gets the item - count mode: ``count`` instances, item_field absent - both modes: inputs map parent fields onto subgraph state fields + + ``pre_resolved_count`` (proposal-0013 hook): if the engine has + already resolved ``cfg.count`` to populate + ``NodeEvent.fan_out_config.item_count``, the resolved value is + passed in here so the callable resolver isn't invoked twice. + Only consulted in ``count`` mode; ``items_field`` mode reads + from the state field unchanged. """ sub_state_cls = cfg.subgraph.state_cls parent_dump = parent_state.model_dump() @@ -226,7 +250,10 @@ def _build_instance_states( return instances # count mode - resolved_count = _resolve_count(node_name, cfg, parent_state) + if pre_resolved_count is not None: + resolved_count = pre_resolved_count + else: + resolved_count = _resolve_count(node_name, cfg, parent_state) instances = [] for _idx in range(resolved_count): init = {} diff --git a/src/openarmature/observability/otel/observer.py b/src/openarmature/observability/otel/observer.py index 5960a2f..1d65fa0 100644 --- a/src/openarmature/observability/otel/observer.py +++ b/src/openarmature/observability/otel/observer.py @@ -131,6 +131,25 @@ class _InvState: subgraph_spans: dict[tuple[str, ...], _OpenSpan] = field(default_factory=dict[tuple[str, ...], _OpenSpan]) detached_roots: dict[tuple[str, ...], _OpenSpan] = field(default_factory=dict[tuple[str, ...], _OpenSpan]) fan_out_instance_root_prefixes: set[tuple[str, ...]] = field(default_factory=set[tuple[str, ...]]) + # Per spec §5.4 + proposal 0013 (v0.10.0): non-detached fan-outs + # synthesize per-instance dispatch spans nested between the fan-out + # node span and the inner-node spans. Keyed by ``prefix + + # (str(fan_out_index),)`` mirroring the detached path's keying in + # ``detached_roots``. Lives in the parent trace (not a fresh + # trace_id, unlike detached). Closed when the fan-out node's + # ``completed`` event fires. + fan_out_instance_spans: dict[tuple[str, ...], _OpenSpan] = field( + default_factory=dict[tuple[str, ...], _OpenSpan] + ) + # ``parent_node_name`` cache (per spec proposal 0013 correction #4): + # the fan-out node's name surfaces on its own ``started`` event via + # ``NodeEvent.fan_out_config.parent_node_name``. The observer caches + # it keyed by the fan-out node's namespace prefix so when + # ``_open_fan_out_instance_dispatch_span`` synthesizes a per-instance + # dispatch span, it can attach + # ``openarmature.fan_out.parent_node_name`` even though the inner + # event itself doesn't carry ``fan_out_config``. + fan_out_parent_node_name: dict[tuple[str, ...], str] = field(default_factory=dict[tuple[str, ...], str]) @dataclass @@ -253,6 +272,15 @@ def _handle_started(self, event: NodeEvent) -> None: if invocation_id not in self._invocation_span: self._open_invocation_span(invocation_id, correlation_id, event) + # Per spec proposal 0013 (v0.10.0): cache the fan-out node's + # ``parent_node_name`` from its ``started`` event so + # ``_open_fan_out_instance_dispatch_span`` can attach + # ``openarmature.fan_out.parent_node_name`` to per-instance + # spans. Inner events from inside the fan-out's instances + # don't carry ``fan_out_config`` themselves; the cache bridges. + if event.fan_out_config is not None and event.fan_out_index is None: + inv_state.fan_out_parent_node_name[event.namespace] = event.fan_out_config.parent_node_name + # Synthesize subgraph dispatch spans for any ancestor namespace # prefix that doesn't have one yet (per observability §4.5). # Also closes subgraph spans we've left. @@ -286,6 +314,16 @@ def _handle_completed(self, event: NodeEvent) -> None: for key in list(inv_state.detached_roots.keys()): if len(key) > len(event.namespace) and key[: len(event.namespace)] == event.namespace: self._close_detached_root(inv_state, key) + # Per spec proposal 0013 (v0.10.0): if this is the fan-out + # node's own completion (non-detached path), close all + # per-instance dispatch spans synthesized for this fan-out. + # Children-before-parents close ordering: per-instance + # dispatch spans before the fan-out node's own span. + if event.fan_out_index is None and event.fan_out_config is not None: + for key in list(inv_state.fan_out_instance_spans.keys()): + if len(key) > len(event.namespace) and key[: len(event.namespace)] == event.namespace: + self._close_fan_out_instance_dispatch_span(inv_state, key) + inv_state.fan_out_parent_node_name.pop(event.namespace, None) key = self._key_for(event) open_span = inv_state.open_spans.pop(key, None) if open_span is None: @@ -498,6 +536,16 @@ def _resolve_parent_context( root = inv_state.detached_roots.get(instance_key) if root is not None: return set_span_in_context(root.span) + # 1a'. Non-detached per-instance dispatch span (proposal + # 0013, v0.10.0). Same keying as the detached path + # but lives in the parent trace under the fan-out + # node's span. Inner-node events from inside a + # non-detached fan-out instance parent under THIS + # per-instance dispatch span (not the shared fan-out + # node span at ``namespace[:1]``). + instance_dispatch = inv_state.fan_out_instance_spans.get(instance_key) + if instance_dispatch is not None: + return set_span_in_context(instance_dispatch.span) # 1b. Detached subgraph root at any matching prefix wins # (highest precedence — events inside a detached subtree # always parent under the detached root, never bleed up). @@ -568,6 +616,21 @@ def _sync_subgraph_spans( continue if prefix in inv_state.detached_roots: continue + # Per spec proposal 0013 (v0.10.0): for non-detached + # fan-out instances, the per-instance dispatch span lives + # at ``prefix + (str(fan_out_index),)`` in + # ``fan_out_instance_spans``. If the per-instance dispatch + # span for THIS event's instance is already open, the + # fan-out node span at ``prefix`` is already open too + # (synthesized as part of the fan-out node's started + # event), so we don't open a new subgraph span at + # ``prefix``. + if ( + depth == 1 + and event.fan_out_index is not None + and (prefix + (str(event.fan_out_index),)) in inv_state.fan_out_instance_spans + ): + continue # If this prefix's first segment is configured as a # detached subgraph, mint a fresh trace. if depth == 1 and prefix[0] in self.detached_subgraphs: @@ -579,6 +642,22 @@ def _sync_subgraph_spans( if depth == 1 and event.fan_out_index is not None and prefix[0] in self.detached_fan_outs: self._open_detached_fan_out_instance_root(inv_state, correlation_id, prefix, event) continue + # Per spec §5.4 + proposal 0013: non-detached fan-out + # instances get a synthetic per-instance dispatch span + # under the fan-out node span. Triggered by an event + # from inside a fan-out instance (event.fan_out_index + # populated) at depth 1 (the fan-out node's namespace + # prefix), where the parent_node_name cache has an + # entry — i.e., the fan-out node's started event has + # been seen. + if ( + depth == 1 + and event.fan_out_index is not None + and prefix[0] not in self.detached_fan_outs + and prefix in inv_state.fan_out_parent_node_name + ): + self._open_fan_out_instance_dispatch_span(inv_state, correlation_id, prefix, event) + continue self._open_subgraph_span(inv_state, invocation_id, correlation_id, prefix) def _open_subgraph_span( @@ -747,6 +826,63 @@ def _open_detached_fan_out_instance_root( inv_state.detached_roots[instance_key] = _OpenSpan(span=instance_root) inv_state.fan_out_instance_root_prefixes.add(instance_key) + def _open_fan_out_instance_dispatch_span( + self, + inv_state: _InvState, + correlation_id: str | None, + prefix: tuple[str, ...], + event: NodeEvent, + ) -> None: + """Per-instance dispatch span for a non-detached fan-out + (per spec §5.4 + proposal 0013, v0.10.0). Mirror of + ``_open_detached_fan_out_instance_root`` but lives in the + parent trace (no fresh trace_id). + + Parents under the fan-out node span at ``prefix``. Span name + is the fan-out node's name; attributes are + ``openarmature.fan_out.parent_node_name`` (looked up from the + cache populated when the fan-out node's ``started`` event + landed), ``openarmature.node.fan_out_index`` (from + ``event.fan_out_index``), and the correlation_id if set. + + Stored in ``inv_state.fan_out_instance_spans`` keyed by + ``prefix + (str(fan_out_index),)``. Closed when the fan-out + node's own ``completed`` event fires (children-before-parents + ordering). + """ + # Find the fan-out node's open span (latest attempt under + # retry) to use as parent. + fan_out_open = self._find_fan_out_node_span(inv_state, prefix) + parent_ctx: object + if fan_out_open is not None: + parent_ctx = set_span_in_context(fan_out_open.span) + else: + parent_ctx = otel_context.Context() + + parent_node_name = inv_state.fan_out_parent_node_name.get(prefix, prefix[-1]) + attrs: dict[str, Any] = { + "openarmature.node.name": prefix[-1], + "openarmature.fan_out.parent_node_name": parent_node_name, + "openarmature.node.fan_out_index": event.fan_out_index, + } + if correlation_id is not None: + attrs["openarmature.correlation_id"] = correlation_id + instance_span = self._tracer.start_span( + name=prefix[-1], + context=cast("Any", parent_ctx), + kind=SpanKind.INTERNAL, + attributes=attrs, + ) + instance_key = prefix + (str(event.fan_out_index),) + inv_state.fan_out_instance_spans[instance_key] = _OpenSpan(span=instance_span) + + def _close_fan_out_instance_dispatch_span(self, inv_state: _InvState, key: tuple[str, ...]) -> None: + open_span = inv_state.fan_out_instance_spans.pop(key, None) + if open_span is None: + return + open_span.span.set_status(Status(StatusCode.OK)) + open_span.span.end() + def _close_detached_root(self, inv_state: _InvState, prefix: tuple[str, ...]) -> None: inv_state.fan_out_instance_root_prefixes.discard(prefix) open_span = inv_state.detached_roots.pop(prefix, None) @@ -789,6 +925,18 @@ def _node_attrs(self, event: NodeEvent, correlation_id: str | None) -> dict[str, attrs["openarmature.node.fan_out_index"] = event.fan_out_index if correlation_id is not None: attrs["openarmature.correlation_id"] = correlation_id + # Per spec §5.4 + proposal 0013 (v0.10.0): fan-out node spans + # carry item_count / concurrency / error_policy. ``concurrency`` + # is ``int | None`` on the event payload (spec §9.2 canonical + # type); the §5.4 attribute is a bare int with ``0`` as the + # "unbounded" sentinel (OTel attribute primitives can't carry + # null cleanly), so this is the OTel-attribute-layer + # translation. + if event.fan_out_config is not None: + cfg = event.fan_out_config + attrs["openarmature.fan_out.item_count"] = cfg.item_count + attrs["openarmature.fan_out.concurrency"] = 0 if cfg.concurrency is None else cfg.concurrency + attrs["openarmature.fan_out.error_policy"] = cfg.error_policy return attrs # ------------------------------------------------------------------ @@ -830,13 +978,34 @@ def _drain_inv_state(self, inv_state: _InvState) -> None: """Close any still-open spans in a per-invocation state container in child→parent order. LLM spans (deepest leaves) → leaf node spans (sorted deepest-first by namespace) → - detached roots → subgraph dispatch spans. Matches the - ordering used in ``shutdown``.""" + non-detached fan-out per-instance dispatch spans → detached + roots → subgraph dispatch spans. Matches the ordering used in + ``shutdown``.""" for call_id in list(inv_state.open_llm_spans.keys()): open_span = inv_state.open_llm_spans.pop(call_id, None) if open_span is not None: self._drain_open_span(open_span) - for key in sorted(inv_state.open_spans.keys(), key=lambda k: -len(k[0])): + # Inner-node spans (depth >= 2) drain first — these include + # the inner-node bodies inside fan-out instances, which are + # children of the per-instance dispatch spans. + for key in sorted( + (k for k in inv_state.open_spans if len(k[0]) >= 2), + key=lambda k: -len(k[0]), + ): + open_span = inv_state.open_spans.pop(key, None) + if open_span is not None: + self._drain_open_span(open_span) + # Per-instance dispatch spans (children of the fan-out NODE + # span) drain BEFORE depth-1 entries of ``open_spans`` — + # otherwise the fan-out NODE span (depth 1) would end + # before its per-instance children, violating + # children-before-parents close ordering. + for key in sorted(inv_state.fan_out_instance_spans.keys(), key=lambda k: -len(k)): + self._close_fan_out_instance_dispatch_span(inv_state, key) + # Remaining depth-1 entries in ``open_spans`` drain last — + # these are the fan-out NODE span itself + any sibling + # depth-1 leaf nodes still open. + for key in list(inv_state.open_spans.keys()): open_span = inv_state.open_spans.pop(key, None) if open_span is not None: self._drain_open_span(open_span) diff --git a/tests/conformance/adapter.py b/tests/conformance/adapter.py index 322d066..b51b31d 100644 --- a/tests/conformance/adapter.py +++ b/tests/conformance/adapter.py @@ -370,9 +370,17 @@ async def run_with_context( self, state: State, context: _InvocationContext, + *, + pre_resolved_count: int | None = None, + pre_resolved_concurrency: tuple[int | None] | None = None, ) -> Mapping[str, Any]: self.trace_list.append(self.name) - return await super().run_with_context(state, context) + return await super().run_with_context( + state, + context, + pre_resolved_count=pre_resolved_count, + pre_resolved_concurrency=pre_resolved_concurrency, + ) @dataclass(frozen=True) diff --git a/tests/conformance/test_observability.py b/tests/conformance/test_observability.py index abebc3d..5b20f54 100644 --- a/tests/conformance/test_observability.py +++ b/tests/conformance/test_observability.py @@ -72,6 +72,7 @@ "003-otel-error-status", "004-otel-routing-error-attribution", "005-otel-llm-provider-span-nested", + "006-otel-fan-out-instance-attribution", "007-otel-retry-attempt-spans", "008-otel-detached-trace-mode", "009-otel-correlation-id-cross-cutting", @@ -81,11 +82,6 @@ _DEFERRED_FIXTURES: dict[str, str] = { - "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 " - "shared synthetic span at the namespace prefix, not per-instance). Lands in PR-C.2." - ), "010-otel-log-correlation": ( "Needs synchronous observer prep hook (prepare_sync) so the engine task can " "attach the observer's span to OTel context for the duration of node-body " @@ -139,6 +135,8 @@ async def test_observability_fixture(fixture_path: Path) -> None: await _run_fixture_004(spec) elif fixture_id == "005-otel-llm-provider-span-nested": await _run_fixture_005(spec) + elif fixture_id == "006-otel-fan-out-instance-attribution": + await _run_fixture_006(spec) elif fixture_id == "007-otel-retry-attempt-spans": await _run_fixture_007(spec) elif fixture_id == "008-otel-detached-trace-mode": @@ -455,6 +453,103 @@ async def _run_fixture_004(spec: Mapping[str, Any]) -> None: # --------------------------------------------------------------------------- +async def _run_fixture_006(spec: Mapping[str, Any]) -> None: + """Spec §5.4 + proposal 0013 (v0.10.0): non-detached fan-out + instances synthesize per-instance dispatch spans nested between + the fan-out node span and the inner-node spans. The fan-out node + span carries ``item_count`` / ``concurrency`` / ``error_policy`` + from ``NodeEvent.fan_out_config``; per-instance spans carry + ``fan_out_index`` and ``parent_node_name``.""" + # Subgraphs declared at the spec level (outside ``cases:``) — the + # ``worker`` subgraph used by every case in this fixture lives + # there. Compile once and reuse across cases. + _patch_unsupported_directives(spec) + subgraphs = _compile_subgraphs(spec) + cases = cast("list[dict[str, Any]]", spec["cases"]) + for case in cases: + case_name = cast("str", case["name"]) + try: + await _run_fixture_006_case(case, subgraphs) + except AssertionError as e: + raise AssertionError(f"case {case_name!r}: {e}") from e + + +async def _run_fixture_006_case(case: Mapping[str, Any], subgraphs: Mapping[str, Any]) -> None: + _patch_unsupported_directives(case) + + observer, exporter = _build_observer() + trace_log: list[str] = [] + built = build_graph(case, subgraphs=dict(subgraphs), trace=trace_log) + compiled = built.builder.compile() + compiled.attach_observer(observer) + initial_state = built.initial_state(case.get("initial_state", {})) + await compiled.invoke(initial_state) + await compiled.drain() + observer.shutdown() + spans = exporter.get_finished_spans() + + # Span-tree shape per fixture: + # invocation + # └─ process (fan-out NODE) — item_count=3, concurrency=2, error_policy="collect" + # ├─ process (instance, fan_out_index=0) — parent_node_name="process" + # │ └─ compute + # ├─ process (instance, fan_out_index=1) — parent_node_name="process" + # │ └─ compute + # └─ process (instance, fan_out_index=2) — parent_node_name="process" + # └─ compute + process_spans = [s for s in spans if s.name == "process"] + # Expect 4: 1 fan-out node + 3 per-instance dispatch spans. + assert len(process_spans) == 4, ( + f"expected 4 'process' spans (1 fan-out node + 3 per-instance dispatch); got {len(process_spans)}" + ) + # Fan-out node span carries the three §5.4 attributes. + fan_out_node_spans = [ + s + for s in process_spans + if dict(s.attributes or {}).get("openarmature.fan_out.item_count") is not None + ] + assert len(fan_out_node_spans) == 1, ( + f"expected exactly 1 fan-out NODE span (with item_count attribute); got {len(fan_out_node_spans)}" + ) + fan_out_node_span = fan_out_node_spans[0] + fan_out_attrs = dict(fan_out_node_span.attributes or {}) + assert fan_out_attrs.get("openarmature.fan_out.item_count") == 3 + assert fan_out_attrs.get("openarmature.fan_out.concurrency") == 2 + assert fan_out_attrs.get("openarmature.fan_out.error_policy") == "collect" + + # Per-instance dispatch spans: 3 of them, each with + # fan_out_index 0..2 and parent_node_name "process". + per_instance_spans = [s for s in process_spans if s != fan_out_node_span] + assert len(per_instance_spans) == 3 + fan_out_indices: set[int] = set() + for s in per_instance_spans: + attrs = dict(s.attributes or {}) + idx = attrs.get("openarmature.node.fan_out_index") + assert isinstance(idx, int), f"per-instance span MUST carry fan_out_index; got attrs={attrs}" + fan_out_indices.add(idx) + assert attrs.get("openarmature.fan_out.parent_node_name") == "process", ( + f"per-instance span MUST carry parent_node_name='process'; got {attrs}" + ) + # Each per-instance dispatch span parents under the fan-out + # node span (proposal 0013 + §5.4 nesting). + assert s.parent is not None and s.parent.span_id == fan_out_node_span.context.span_id, ( + "per-instance dispatch span MUST parent under the fan-out node span" + ) + assert fan_out_indices == {0, 1, 2}, ( + f"per-instance fan_out_index range MUST be 0..2; got {sorted(fan_out_indices)}" + ) + + # Each per-instance dispatch span has a 'compute' child (the + # inner-node work). + per_instance_ids = {s.context.span_id for s in per_instance_spans} + compute_spans = [s for s in spans if s.name == "compute"] + assert len(compute_spans) == 3, f"expected 3 compute spans; got {len(compute_spans)}" + for cs in compute_spans: + assert cs.parent is not None and cs.parent.span_id in per_instance_ids, ( + "compute span MUST parent under a per-instance dispatch span" + ) + + async def _run_fixture_007(spec: Mapping[str, Any]) -> None: """Two sub-cases: diff --git a/tests/test_smoke.py b/tests/test_smoke.py index cd55887..aa01689 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.9.0" + assert openarmature.__spec_version__ == "0.10.0" diff --git a/tests/unit/test_observability_otel.py b/tests/unit/test_observability_otel.py index 94ad62e..20e1474 100644 --- a/tests/unit/test_observability_otel.py +++ b/tests/unit/test_observability_otel.py @@ -24,6 +24,7 @@ import logging import pytest +from pydantic import Field # Skip the entire module if otel extras aren't installed. pytest.importorskip("opentelemetry.sdk.trace") @@ -271,6 +272,9 @@ async def test_disable_llm_spans_skips_llm_provider_span() -> None: span_processor=SimpleSpanProcessor(exporter), disable_llm_spans=True, ) + # ``step=-1`` mirrors the synthetic value ``OpenAIProvider._llm_event`` + # mints (openai.py:643) — LLM-provider events aren't tied to graph step + # sequencing. started = NodeEvent( node_name="openarmature.llm.complete", namespace=("openarmature.llm.complete",), @@ -556,8 +560,8 @@ async def test_concurrent_fan_out_no_lifo_violation() -> None: import warnings class _ParentState(State): - items: list[int] = [] - results: list[int] = [] + items: list[int] = Field(default_factory=list[int]) + results: list[int] = Field(default_factory=list[int]) class _ChildState(State): item: int = 0 diff --git a/tests/unit/test_runtime_errors.py b/tests/unit/test_runtime_errors.py index 0b7f213..acd0c8b 100644 --- a/tests/unit/test_runtime_errors.py +++ b/tests/unit/test_runtime_errors.py @@ -105,7 +105,7 @@ async def test_subgraph_projection_error_wrapped_as_node_exception() -> None: NodeException tagged with the subgraph wrapper's name so callers see a uniform error contract.""" - from openarmature.graph import NodeException, ProjectionStrategy + from openarmature.graph import NodeException class Inner(State): x: int = 0 @@ -115,8 +115,9 @@ async def _inner_node(_s: Inner) -> dict[str, Any]: inner_g = GraphBuilder(Inner).add_node("i", _inner_node).add_edge("i", END).set_entry("i").compile() - # Parameter names match the ProjectionStrategy Protocol exactly so - # pyright's strict structural conformance check passes. + # Parameter names match the ProjectionStrategy Protocol exactly so the + # call-site ``projection=BoomProjection()`` below passes pyright's strict + # structural conformance check. class BoomProjection: def project_in(self, parent_state: S, subgraph_state_cls: type[Inner]) -> Inner: raise RuntimeError("project_in boom") @@ -129,8 +130,6 @@ def project_out( ) -> dict[str, Any]: return {} - _: ProjectionStrategy[S, Inner] = BoomProjection() - g = ( GraphBuilder(S) .add_subgraph_node("sub", inner_g, projection=BoomProjection())