Implement proposal 0044 (parallel-branches dispatch span)

CHANGELOG.md

@@ -18,6 +18,14 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). The
 
   The `Observer.__call__` signature widens to `NodeEvent | MetadataAugmentationEvent | InvocationStartedEvent | InvocationCompletedEvent`. The new `ObserverEvent` type alias (re-exported from `openarmature.graph`) gives observer authors a one-name handle on the union; existing observers that ignore non-`NodeEvent` variants early-return after an `isinstance(event, NodeEvent)` check.
 - **`LangfuseTrace.input` / `LangfuseTrace.output` dataclass fields** on the in-memory recorder, populated by the new observer paths.
+- **Parallel-branches OTel dispatch span synthesis** (proposal 0044, observability §5.7). Mirroring the fan-out per-instance dispatch synthesis (proposal 0013), the OTel observer now synthesizes a per-branch dispatch span between the parallel-branches NODE span and each branch's inner-node spans. New `ParallelBranchesEventConfig` payload on `NodeEvent` (`branch_names`, `branch_count`, `error_policy`, `parent_node_name`); engine populates it on the parallel-branches NODE's `started` / `completed` events. New OTel span attributes:
+  - `openarmature.parallel_branches.branch_count` + `openarmature.parallel_branches.error_policy` on the parallel-branches NODE span.
+  - `openarmature.node.branch_name` + `openarmature.parallel_branches.parent_node_name` on each per-branch dispatch span.
+  - `openarmature.node.branch_name` on every inner-node span beneath a per-branch dispatch span.
+
+### Changed (breaking, pre-1.0)
+
+- **OTel span attribute `openarmature.branch_name` is renamed to `openarmature.node.branch_name`** to align with the spec §5.7 attribute namespace. Prior python releases emitted `openarmature.branch_name` as a workaround because the spec hadn't defined an OTel attribute carrying `branch_name` yet; proposal 0044 (v0.36.0) formalizes the namespace. **Downstream dashboards, queries, or alerts filtering on the old attribute name MUST update.** Pre-1.0 break; the prior name was python-implementation-only and was never spec-normative.
 
 ### Changed
 
@@ -32,7 +40,7 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). The
 
 ### Notes
 
-- **Pinned spec version bumped from v0.31.0 to v0.35.0.** Absorbs proposals 0042 (reserved-key extension), 0043 (Langfuse trace.input/output sourcing), and the textual additions in v0.32.0 (Gemini wire-format mapping, 0038, not yet implemented) and v0.33.0 (sessions capability, 0020, not yet implemented).
+- **Pinned spec version bumped from v0.31.0 to v0.36.0.** Absorbs proposals 0042 (reserved-key extension), 0043 (Langfuse trace.input/output sourcing), 0044 (parallel-branches OTel dispatch span), and the textual additions in v0.32.0 (Gemini wire-format mapping, 0038, not yet implemented) and v0.33.0 (sessions capability, 0020, not yet implemented).
 - `LangfuseSDKAdapter` now applies `trace.input` / `trace.output` to the live Langfuse Trace. Input lands on the first real observation under the trace via `set_trace_io`; output uses a synthetic short-lived `openarmature.trace_io` observation as the carrier. The InMemoryLangfuseClient used by tests applies the fields directly.
 - Conformance fixture `observability/conformance/037-langfuse-trace-input-output` activated for all five cases (default stub / `disable_state_payload=False` / hooks non-null / hooks null-fallthrough / resume re-fire). The langfuse harness grew per-case `checkpointer: in_memory` wiring, a compact `flaky:` test seam, and a two-phase resume-flow assertion path.
 - The Langfuse v4 SDK marks `set_current_trace_io` / `Span.set_trace_io` deprecated ("removal in a future major version"). Empirical verification against Langfuse Cloud (v4.7.1, 2026-05-29) confirms it remains the **only** path that populates the Traces list view's headline `Input` / `Output` columns; `propagate_attributes(metadata=...)` does not substitute for it in the current UI. We will revisit when Langfuse publishes a concrete migration guide for v5.

conformance.toml

@@ -32,7 +32,7 @@
 
 [manifest]
 implementation = "openarmature-python"
-spec_pin = "v0.35.0"
+spec_pin = "v0.36.0"
 
 # Status values:
 #   implemented   — shipped behavior matches the proposal's contract
@@ -210,3 +210,8 @@ since = "0.11.0"
 [proposals."0043"]
 status = "implemented"
 since = "0.11.0"
+
+# Spec v0.36.0 (proposal 0044).
+[proposals."0044"]
+status = "implemented"
+since = "0.11.0"

pyproject.toml

@@ -58,7 +58,7 @@ Specification = "https://github.com/LunarCommand/openarmature-spec"
 openarmature = "openarmature.cli:main"
 
 [tool.openarmature]
-spec_version = "0.35.0"
+spec_version = "0.36.0"
 
 [dependency-groups]
 dev = [

src/openarmature/AGENTS.md

@@ -1,6 +1,6 @@
 # OpenArmature — Agent documentation
 
-*This is the agent guide bundled with the openarmature Python package, version 0.10.0 (spec v0.35.0). For the full docs site see [openarmature.ai](https://openarmature.ai). For the canonical spec text see [openarmature.org/capabilities](https://openarmature.org/capabilities/). For project-specific conventions for the code you're editing, see the host project's `AGENTS.md` or `CLAUDE.md`.*
+*This is the agent guide bundled with the openarmature Python package, version 0.10.0 (spec v0.36.0). For the full docs site see [openarmature.ai](https://openarmature.ai). For the canonical spec text see [openarmature.org/capabilities](https://openarmature.org/capabilities/). For project-specific conventions for the code you're editing, see the host project's `AGENTS.md` or `CLAUDE.md`.*
 
 ## TL;DR
 
@@ -10,7 +10,7 @@ OpenArmature is a workflow framework for LLM pipelines and tool-calling agents 
 
 ## Capability contracts
 
-_Sourced from openarmature-spec v0.35.0. Each entry below reproduces §1 (Purpose) and §2 (Concepts) of the capability's `spec.md`. For the full spec text (execution model, error semantics, determinism, observer hooks, etc.) see the linked docs site._
+_Sourced from openarmature-spec v0.36.0. Each entry below reproduces §1 (Purpose) and §2 (Concepts) of the capability's `spec.md`. For the full spec text (execution model, error semantics, determinism, observer hooks, etc.) see the linked docs site._
 
 ### Capability: `graph-engine`
 

src/openarmature/__init__.py

@@ -25,4 +25,4 @@
 """
 
 __version__ = "0.10.0"
-__spec_version__ = "0.35.0"
+__spec_version__ = "0.36.0"

src/openarmature/graph/compiled.py

@@ -100,6 +100,7 @@
     InvocationCompletedEvent,
     InvocationStartedEvent,
     NodeEvent,
+    ParallelBranchesEventConfig,
 )
 from .middleware import ChainCall, Middleware, compose_chain
 from .nodes import Node
@@ -1985,6 +1986,29 @@ async def _step_parallel_branches_node(
         attempt_counter: list[int] = [0]
         deferred_info: list[tuple[int, StateT, StateT] | None] = [None]
 
+        # Per proposal 0044 (observability §5.7, v0.36.0): the
+        # resolved parallel-branches configuration is static at
+        # compile time (no count / concurrency resolvers like fan-out
+        # has), so build the event config once here and ship it on
+        # both started + completed events. ``branch_names`` mirrors
+        # the dispatch order ParallelBranchesNode uses internally
+        # (insertion order of the ``branches`` dict per pipeline-
+        # utilities §11.1).
+        #
+        # Python dicts preserve insertion order (PEP 468; guaranteed
+        # since 3.7), and YAML / direct-dict-literal ``branches``
+        # construction at the call site preserves the source order
+        # through into the dict's keys().  Spec §11.1 ties branch
+        # declaration order to dispatch order, so this tuple IS the
+        # declaration order observers should see.
+        branch_names: tuple[str, ...] = tuple(node.branches.keys())
+        parallel_branches_event_config = ParallelBranchesEventConfig(
+            branch_names=branch_names,
+            branch_count=len(branch_names),
+            error_policy=node.error_policy,
+            parent_node_name=current,
+        )
+
         async def innermost(s: Any) -> Mapping[str, Any]:
             attempt_counter[0] += 1
             # Read from ContextVar — see ``_step_function_node``'s
@@ -1998,6 +2022,7 @@ async def innermost(s: Any) -> Mapping[str, Any]:
                 step,
                 s,
                 attempt_index=attempt_index,
+                parallel_branches_config=parallel_branches_event_config,
             )
             otel_token = _attach_active_observer_span()
             try:
@@ -2012,6 +2037,7 @@ async def innermost(s: Any) -> Mapping[str, Any]:
                         s,
                         error=e,
                         attempt_index=attempt_index,
+                        parallel_branches_config=parallel_branches_event_config,
                     )
                     raise
                 except Exception as e:
@@ -2024,6 +2050,7 @@ async def innermost(s: Any) -> Mapping[str, Any]:
                         s,
                         error=wrapped,
                         attempt_index=attempt_index,
+                        parallel_branches_config=parallel_branches_event_config,
                     )
                     raise wrapped from e
             finally:
@@ -2041,6 +2068,7 @@ async def innermost(s: Any) -> Mapping[str, Any]:
                     s,
                     error=e,
                     attempt_index=attempt_index,
+                    parallel_branches_config=parallel_branches_event_config,
                 )
                 raise
 
@@ -2094,6 +2122,7 @@ def finalize_completed(edge_error: RuntimeGraphError | None) -> None:
                     final_pre_state,
                     post_state=final_merged,
                     attempt_index=final_attempt_index,
+                    parallel_branches_config=parallel_branches_event_config,
                 )
             else:
                 self._dispatch_completed(
@@ -2104,6 +2133,7 @@ def finalize_completed(edge_error: RuntimeGraphError | None) -> None:
                     final_pre_state,
                     error=edge_error,
                     attempt_index=final_attempt_index,
+                    parallel_branches_config=parallel_branches_event_config,
                 )
 
         return _StepResult(state=merged_outer, finalize_completed=finalize_completed)
@@ -2118,6 +2148,7 @@ def _dispatch_started(
         *,
         attempt_index: int = 0,
         fan_out_config: FanOutEventConfig | None = None,
+        parallel_branches_config: ParallelBranchesEventConfig | None = None,
     ) -> None:
         # Per graph-engine §6 + pipeline-utilities §11: read the
         # active branch_name (set by ParallelBranchesNode inside
@@ -2140,6 +2171,7 @@ def _dispatch_started(
                 attempt_index=attempt_index,
                 fan_out_index=context.fan_out_index,
                 fan_out_config=fan_out_config,
+                parallel_branches_config=parallel_branches_config,
                 branch_name=current_branch_name(),
                 subgraph_identities=context.subgraph_identities,
                 caller_invocation_metadata=current_invocation_metadata(),
@@ -2158,6 +2190,7 @@ def _dispatch_completed(
         error: RuntimeGraphError | None = None,
         attempt_index: int = 0,
         fan_out_config: FanOutEventConfig | None = None,
+        parallel_branches_config: ParallelBranchesEventConfig | None = None,
     ) -> None:
         from openarmature.observability.correlation import current_branch_name  # noqa: PLC0415
 
@@ -2175,6 +2208,7 @@ def _dispatch_completed(
                 attempt_index=attempt_index,
                 fan_out_index=context.fan_out_index,
                 fan_out_config=fan_out_config,
+                parallel_branches_config=parallel_branches_config,
                 branch_name=current_branch_name(),
                 subgraph_identities=context.subgraph_identities,
                 caller_invocation_metadata=current_invocation_metadata(),

src/openarmature/graph/events.py

@@ -77,6 +77,52 @@ class FanOutEventConfig:
     parent_node_name: str
 
 
+# Spec: realizes observability §5.7 parallel-branches attributes via
+# the event-payload mechanism added by proposal 0044 (v0.36.0).
+# Backend observers cache ``parent_node_name`` off the parallel-
+# branches node's started event and apply it on every per-branch
+# dispatch span they synthesize (observability §5.7 mandates
+# ``openarmature.parallel_branches.parent_node_name`` on per-branch
+# dispatch spans). Mirrors :class:`FanOutEventConfig`'s shape.
+@dataclass(frozen=True)
+class ParallelBranchesEventConfig:
+    """Resolved parallel-branches configuration carried on a parallel-
+    branches node's own events.
+
+    Parallel-branches node events carry the resolved configuration so
+    backend observers can attribute the parallel-branches node span
+    (``branch_count`` / ``error_policy``) and synthesize per-branch
+    dispatch spans with the right ``branch_names`` + ``parent_node_name``.
+
+    Populated ONLY on ``started`` and ``completed`` events for a
+    parallel-branches node itself (partition by node type, not event
+    category; INCLUDES retried attempts of a parallel-branches node
+    when retry middleware wraps it). All other events leave
+    ``NodeEvent.parallel_branches_config`` null.
+
+    Field shapes:
+
+    - ``branch_names``: non-empty ordered tuple of strings. The branch
+      identifiers in declaration / dispatch order, as configured on
+      the parallel-branches node (pipeline-utilities §11.1).
+    - ``branch_count``: positive int. Equals ``len(branch_names)``.
+      Surfaced explicitly so observers don't have to derive it.
+    - ``error_policy``: one of ``"fail_fast"`` or ``"collect"`` (per
+      pipeline-utilities §11.5).
+    - ``parent_node_name``: the parallel-branches node's name in the
+      parent graph. Carried here for caching by backend observers
+      when attributing per-branch dispatch spans.
+
+    All four fields MUST be present when ``parallel_branches_config``
+    is populated.
+    """
+
+    branch_names: tuple[str, ...]
+    branch_count: int
+    error_policy: str
+    parent_node_name: str
+
+
 # Spec: realizes graph-engine §6 NodeEvent (started/completed pair
 # model from proposal 0005, v0.6.0). The ``checkpoint_saved`` phase
 # is the Python shape for §10.8 save events (§10.8 SHOULDs an event
@@ -190,6 +236,17 @@ class NodeEvent:
     attempt_index: int = 0
     fan_out_index: int | None = None
     fan_out_config: FanOutEventConfig | None = None
+    # Per observability §5.7 / proposal 0044 (v0.36.0): resolved
+    # parallel-branches configuration carried on the parallel-branches
+    # NODE's own events (mirroring ``fan_out_config`` on a fan-out
+    # NODE's events). Populated on both ``started`` and ``completed``
+    # events for a parallel-branches NODE (including retried
+    # attempts); absent on every other event. Carries the §5.7
+    # branch_count + error_policy + parent_node_name surface so
+    # backend observers can attribute the parallel-branches NODE span
+    # and synthesize per-branch dispatch spans without re-reading the
+    # graph's static config.
+    parallel_branches_config: ParallelBranchesEventConfig | None = None
     # Per pipeline-utilities §11 / graph-engine §6 (proposal 0011):
     # optional non-empty string populated only on events from nodes
     # that execute inside a parallel-branches branch. The
@@ -358,4 +415,5 @@ class InvocationCompletedEvent:
     "InvocationStartedEvent",
     "MetadataAugmentationEvent",
     "NodeEvent",
+    "ParallelBranchesEventConfig",
 ]