observability: phase 6.1 PR-C.2 — proposal 0013 + v0.10.0

.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

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

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"

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.<items_field>)``; 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,14 +1074,22 @@ 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
 
                 try:
                     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,
             ),
         )
 

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