prepare-sync: engine OTel attach around node bodies

Step 4 of PR-C.3.

Adds a try-imported OTel attach helper pair in compiled.py:
``_attach_active_observer_span`` reads
``current_active_observer_span`` (set synchronously by an
observer's ``prepare_sync`` before queueing) and splices the
span into the OTel context via
``opentelemetry.context.attach(set_span_in_context(span))``;
``_detach_active_observer_span`` pairs the detach in ``finally``.

Both ``_step_function_node``'s and ``_step_fan_out_node``'s
``innermost`` closures now attach right after
``_dispatch_started`` returns and detach in a ``finally`` around
the ``await node.run(...)`` / ``await node.run_with_context(...)``
call. That puts the attach scope around exactly the user-code
window — so logs emitted on the FIRST line of a node body, before
any ``await``, pick up the right ``trace_id``/``span_id`` via
OTel's ``LoggingHandler`` — and the detach fires before
``_dispatch_completed`` queues the completed event or the merge
runs.

The except branch binds the OTel names to ``None`` so pyright
narrows on ``if _otel_attach is None: ...`` rather than flagging
"possibly unbound." Engine stays no-OTel-dep at runtime: installs
without ``[otel]`` get a no-op attach/detach, the ContextVar
stays ``None``, and nothing changes. Drives the load-bearing
log-correlation cases landing in steps 5 and 6.

Author: chris-colinsky

src/openarmature/graph/compiled.py

@@ -70,6 +70,7 @@
     _set_fan_out_index,
     _set_invocation_id,
     _set_namespace_prefix,
+    current_active_observer_span,
 )
 
 from .edges import END, ConditionalEdge, EndSentinel, StaticEdge
@@ -99,6 +100,54 @@
 from .state import State
 from .subgraph import SubgraphNode
 
+# Try-import OpenTelemetry attach primitives so the engine can splice an
+# observer-published span into the OTel context for the duration of a
+# node body. The engine treats the span value opaquely (writes by an
+# observer's ``prepare_sync``, reads via ``current_active_observer_span``)
+# and only touches OTel when both: (a) the extras are installed, and
+# (b) an observer actually published a span. Installs without ``[otel]``
+# get a no-op attach/detach pair; the observer ContextVar stays
+# ``None`` and nothing changes.
+#
+# The names are bound to ``None`` in the except branch so pyright
+# narrows correctly at call sites (``if _otel_attach is None: ...``)
+# rather than flagging "possibly unbound."
+try:
+    from opentelemetry.context import attach as _otel_attach
+    from opentelemetry.context import detach as _otel_detach
+    from opentelemetry.trace.propagation import set_span_in_context as _otel_set_span_in_context
+except ImportError:  # pragma: no cover — exercised only in non-otel installs
+    _otel_attach = None  # type: ignore[assignment]
+    _otel_detach = None  # type: ignore[assignment]
+    _otel_set_span_in_context = None  # type: ignore[assignment]
+
+
+def _attach_active_observer_span() -> object | None:
+    """Read ``current_active_observer_span``; if an observer published
+    one and OTel is installed, attach the span into the OTel context
+    so that any logs emitted from the next user-code scope (a node
+    body) pick up the right ``trace_id``/``span_id`` via OTel's
+    ``LoggingHandler``.
+
+    Returns the OTel context token to hand back to
+    :func:`_detach_active_observer_span` in ``finally``, or ``None``
+    if no attach happened (no observer, no OTel, or both).
+    """
+    if _otel_attach is None or _otel_set_span_in_context is None:
+        return None
+    span = current_active_observer_span()
+    if span is None:
+        return None
+    return _otel_attach(_otel_set_span_in_context(cast("Any", span)))
+
+
+def _detach_active_observer_span(token: object | None) -> None:
+    """Pair to :func:`_attach_active_observer_span`. No-op when no
+    attach was performed (token is ``None``)."""
+    if token is None or _otel_detach is None:
+        return
+    _otel_detach(cast("Any", token))
+
 
 def _merge_partial[StateT: State](
     prior: StateT,
@@ -690,20 +739,30 @@ async def innermost(s: Any) -> Mapping[str, Any]:
             try:
                 self._dispatch_started(context, current, namespace, step, s, attempt_index=attempt_index)
 
+                # Splice the observer-published span (if any) into the
+                # OTel context so logs emitted from the FIRST line of
+                # the node body — before any ``await`` — pick up the
+                # right trace_id/span_id via OTel's LoggingHandler.
+                # Detach in ``finally`` so retries / merge / completed
+                # dispatch don't run with the span still active.
+                otel_token = _attach_active_observer_span()
                 try:
-                    partial = await node.run(s)
-                except Exception as e:
-                    wrapped = NodeException(node_name=current, cause=e, recoverable_state=s)
-                    self._dispatch_completed(
-                        context,
-                        current,
-                        namespace,
-                        step,
-                        s,
-                        error=wrapped,
-                        attempt_index=attempt_index,
-                    )
-                    raise
+                    try:
+                        partial = await node.run(s)
+                    except Exception as e:
+                        wrapped = NodeException(node_name=current, cause=e, recoverable_state=s)
+                        self._dispatch_completed(
+                            context,
+                            current,
+                            namespace,
+                            step,
+                            s,
+                            error=wrapped,
+                            attempt_index=attempt_index,
+                        )
+                        raise
+                finally:
+                    _detach_active_observer_span(otel_token)
 
                 try:
                     merged = _merge_partial(s, partial, self.reducers, current)
@@ -1045,38 +1104,51 @@ async def innermost(s: Any) -> Mapping[str, Any]:
                     attempt_index=attempt_index,
                     fan_out_config=fan_out_event_config,
                 )
+                # Same OTel attach pattern as ``_step_function_node``'s
+                # ``innermost`` — splice the observer-published span
+                # into the OTel context so logs emitted from inside
+                # the fan-out node's own scope (middleware bodies,
+                # the dispatch machinery) carry the right
+                # trace_id/span_id. Per-instance bodies get their own
+                # attach inside their ``_step_function_node``
+                # innermost when the recursive invocation hits leaf
+                # nodes.
+                otel_token = _attach_active_observer_span()
                 try:
-                    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,
-                        fan_out_config=fan_out_event_config,
-                    )
-                    raise
-                except Exception as e:
-                    wrapped = NodeException(node_name=current, cause=e, recoverable_state=s)
-                    self._dispatch_completed(
-                        context,
-                        current,
-                        namespace,
-                        step,
-                        s,
-                        error=wrapped,
-                        attempt_index=attempt_index,
-                        fan_out_config=fan_out_event_config,
-                    )
-                    raise wrapped from e
+                    try:
+                        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,
+                            fan_out_config=fan_out_event_config,
+                        )
+                        raise
+                    except Exception as e:
+                        wrapped = NodeException(node_name=current, cause=e, recoverable_state=s)
+                        self._dispatch_completed(
+                            context,
+                            current,
+                            namespace,
+                            step,
+                            s,
+                            error=wrapped,
+                            attempt_index=attempt_index,
+                            fan_out_config=fan_out_event_config,
+                        )
+                        raise wrapped from e
+                finally:
+                    _detach_active_observer_span(otel_token)
 
                 try:
                     merged = _merge_partial(s, partial, self.reducers, current)