observability: phase 6.1 PR-C.3 — prepare_sync + fixture 010

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)

src/openarmature/graph/observer.py

@@ -60,6 +60,25 @@ async def log_observer(event: NodeEvent) -> None:
     The event parameter is positional-only (`event, /`) so structural
     conformance doesn't pin you to that name — any of `event`, `_event`,
     `e`, etc. matches.
+
+    Optional ``prepare_sync`` extension
+    -----------------------------------
+    An observer MAY additionally define a synchronous method::
+
+        def prepare_sync(self, event: NodeEvent, /) -> None: ...
+
+    that the engine calls IN THE ENGINE TASK, BEFORE queueing the
+    event for the async ``__call__``. This exists for observers that
+    need to set up state — e.g., open a span and stash a handle in
+    a ContextVar — that the engine itself must read synchronously
+    before running the node body (otherwise logs emitted on the
+    first line of the body wouldn't see the right span).
+
+    ``prepare_sync`` is **opt-in via ``hasattr``** — no subclass or
+    Protocol method required. Observers that don't define it skip
+    the synchronous prep entirely; observers that do define it run
+    only for ``"started"``-phase events, errors warned not propagated
+    (same isolation contract as the async path per spec §6).
     """
 
     async def __call__(self, event: NodeEvent, /) -> None: ...
@@ -344,12 +363,46 @@ def take_step(self) -> int:
 def _dispatch(context: _InvocationContext, event: NodeEvent) -> None:
     """Enqueue a node event for the delivery worker.
 
+    For ``"started"``-phase events, also call any subscribed observer's
+    optional ``prepare_sync(event)`` synchronously — in the engine task,
+    BEFORE queueing — so observers that need to publish per-event state
+    the engine itself reads in the same engine-task scope (e.g., the
+    OTel observer setting ``current_active_observer_span`` for the
+    engine to attach into the OTel context) can do so before the node
+    body runs.
+
+    Phase-gated forwarding: ``prepare_sync`` only fires when ``"started"``
+    is in the subscribed observer's ``phases`` set, mirroring how the
+    async ``deliver_loop`` filters dispatch. A user who explicitly
+    subscribes only to ``{"completed"}`` doesn't get the synchronous
+    prep — the wrapper acts as a uniform phase shield across both
+    sync prep and async dispatch.
+
+    Errors from ``prepare_sync`` follow the same isolation contract as
+    the async path per spec §6: don't propagate, don't break siblings,
+    don't block the queueing or subsequent events. Reported via
+    ``warnings.warn``.
+
     No-op when no observers exist for this depth — avoids paying the queue
     overhead for graphs that don't observe anything.
     """
     observers = context.full_observers()
     if not observers:
         return
+    if event.phase == "started":
+        for subscribed in observers:
+            if "started" not in subscribed.phases:
+                continue
+            prepare_sync = getattr(subscribed.observer, "prepare_sync", None)
+            if prepare_sync is None:
+                continue
+            try:
+                prepare_sync(event)
+            except Exception as e:
+                warnings.warn(
+                    f"observer prepare_sync raised {type(e).__name__}: {e}",
+                    stacklevel=2,
+                )
     context.queue.put_nowait(_QueuedItem(event=event, observers=observers))
 
 

src/openarmature/observability/correlation.py

@@ -290,8 +290,68 @@ def _reset_attempt_index(token: Token[int]) -> None:
     _attempt_index_var.reset(token)
 
 
+# ---------------------------------------------------------------------------
+# Active observer span — for engine-side OTel context attach inside
+# ``innermost``. Populated synchronously by an observer's ``prepare_sync``
+# hook BEFORE the engine queues the started event; read by ``innermost``
+# AFTER ``_dispatch_started`` returns to attach the span into the OTel
+# context for the duration of the node-body chain.
+#
+# Inverted directionality vs. the engine→observer ContextVars above:
+# this one flows observer→engine. The producer (an opt-in observer's
+# ``prepare_sync``) and the consumer (``innermost``) both run in the
+# engine task, so the same-task ContextVar contract holds — last-writer-
+# wins is fine in practice (charter §6 says "one OTelObserver per
+# private provider," so multi-OTelObserver attach is rare).
+#
+# Typed as ``object | None`` rather than ``Span | None`` so the base
+# package stays free of an OpenTelemetry import. The OTel observer
+# writes ``Span`` instances; the engine treats the value opaquely and
+# delegates the actual attach to a try-imported OTel helper.
+# ---------------------------------------------------------------------------
+
+
+_active_observer_span_var: ContextVar[object | None] = ContextVar(
+    "openarmature.active_observer_span", default=None
+)
+
+
+def current_active_observer_span() -> object | None:
+    """Return the active observer-side span for the current node body,
+    or ``None`` if no observer published one (no opt-in observer with
+    ``prepare_sync`` is attached, or this is being called outside a
+    node-body scope).
+
+    Engine-readable handle to the span an opt-in observer's
+    ``prepare_sync`` created synchronously during dispatch. The engine's
+    ``innermost`` reads this AFTER ``_dispatch_started`` returns and
+    attaches the span into the OTel context (via a try-imported OTel
+    helper) so that any logs emitted FROM INSIDE the node body — even
+    on the first line, before any ``await`` — pick up the span's
+    trace_id/span_id via OTel's ``LoggingHandler``.
+
+    Backend coupling note: typed as ``object | None`` so this primitive
+    works in installs without the ``[otel]`` extras. OTel observers
+    write OpenTelemetry ``Span`` instances; the engine treats the
+    value opaquely.
+    """
+    return _active_observer_span_var.get()
+
+
+def _set_active_observer_span(value: object | None) -> Token[object | None]:
+    """Set the active observer span. Internal — observers' ``prepare_sync``
+    implementations call this synchronously before returning so the
+    engine's ``innermost`` reads the right value when it attaches."""
+    return _active_observer_span_var.set(value)
+
+
+def _reset_active_observer_span(token: Token[object | None]) -> None:
+    _active_observer_span_var.reset(token)
+
+
 __all__ = [
     # Public surface — readable from anywhere within an invocation.
+    "current_active_observer_span",
     "current_active_observers",
     "current_attempt_index",
     "current_correlation_id",
@@ -304,13 +364,15 @@ def _reset_attempt_index(token: Token[int]) -> None:
     # pyright's strict ``reportUnusedFunction`` flagging them as
     # dead. Underscore-prefixed; not part of the user-facing API.
     "_reset_active_dispatch",
+    "_reset_active_observer_span",
     "_reset_active_observers",
     "_reset_attempt_index",
     "_reset_correlation_id",
     "_reset_fan_out_index",
     "_reset_invocation_id",
     "_reset_namespace_prefix",
     "_set_active_dispatch",
+    "_set_active_observer_span",
     "_set_active_observers",
     "_set_attempt_index",
     "_set_correlation_id",