Address CoPilot review feedback on Langfuse observer

Five real catches from the PR #81 review, all behavioral or
correctness fixes — no spec-touching changes.

1. Detached-subgraph link observation in main Trace was opened but
   never ended. Capture the handle returned by client.span(...) and
   call .end() immediately — the link observation is intentionally
   zero-duration metadata-only, mirroring the OTel observer's
   checkpoint-event synthetic-span pattern.

2. Synthetic subgraph dispatch / fan-out per-instance dispatch
   observations only close on namespace-cursor moves. A subgraph at
   the tail of an invocation never gets its close trigger fired.
   Added close_invocation(invocation_id) and shutdown() drain
   methods mirroring the OTel observer's lifecycle. close_invocation
   walks per-invocation state in child→parent order (LLM
   observations → leaf nodes sorted deepest-first → per-instance
   fan-out dispatches → subgraph dispatches), ending each.
   shutdown() iterates every in-flight invocation_id and calls
   close_invocation. Idempotent.

3. detached_child_trace_ids side-cache was never cleared on
   fan-out node completion. Cyclic graphs re-entering the same
   fan-out would accumulate prior-iteration trace ids into the next
   iteration's list, overwriting the link metadata. Pop the entry
   in _handle_completed's fan-out-completion branch alongside the
   existing fan_out_parent_node_name pop.

4. _resolve_llm_parent_observation_id docstring claimed a
   leaf-ancestor walk fallback that the impl doesn't have. The
   actual precedence (exact-leaf > per-instance fan-out dispatch >
   subgraph dispatch longest-prefix-first > None) is correct
   because the dispatch fallbacks cover the wrapped-call cases an
   ancestor walk would have caught. Docstring updated to describe
   the real chain.

5. cast("list[str]", link_ids) used the string-form unnecessarily.
   With from __future__ import annotations on, the type expression
   form cast(list[str], link_ids) is the right shape and survives
   strict pyright settings.

New unit test verifies the close_invocation drain path: a graph
whose last subtree is a subgraph leaves the dispatch observation
in-flight after invoke + drain; shutdown() ends it cleanly.

Author: chris-colinsky

src/openarmature/observability/langfuse/observer.py

@@ -306,6 +306,12 @@ def _handle_completed(self, event: NodeEvent) -> None:
                 if len(prefix) > len(event.namespace) and prefix[: len(event.namespace)] == event.namespace:
                     self._close_fan_out_instance_dispatch_observation(inv_state, prefix)
             inv_state.fan_out_parent_node_name.pop(event.namespace, None)
+            # Clear the detached-child-trace-ids accumulator for this
+            # fan-out node — cyclic execution that re-enters the same
+            # fan-out starts the next iteration with a fresh list
+            # rather than appending to the previous iteration's
+            # accumulator and overwriting the prior link metadata.
+            inv_state.detached_child_trace_ids.pop(event.namespace, None)
 
         key = self._key_for(event)
         observation = inv_state.open_observations.pop(key, None)
@@ -566,12 +572,18 @@ def _open_detached_subgraph_trace(
             if sg is not None:
                 parent_observation_id = sg.handle.id
                 break
-        self.client.span(
+        # Zero-duration link observation in the main Trace — it
+        # exists only to surface the cross-Trace reference via
+        # metadata.detached_child_trace_ids; close it immediately so
+        # nothing perceives it as in-flight. Mirrors the OTel
+        # observer's synthetic-event zero-duration spans.
+        link_handle = self.client.span(
             trace_id=inv_state.trace_id,
             name=prefix[-1],
             metadata=link_metadata,
             parent_observation_id=parent_observation_id,
         )
+        link_handle.end()
         # Open the detached Trace + the dispatch observation that
         # subtree descendants parent under.
         detached_metadata: dict[str, Any] = {"detached_from_invocation_id": inv_state.trace_id}
@@ -685,6 +697,59 @@ def _find_fan_out_node_observation(
                 return observation
         return None
 
+    # ------------------------------------------------------------------
+    # Lifecycle: close_invocation / shutdown
+    # ------------------------------------------------------------------
+
+    def close_invocation(self, invocation_id: str) -> None:
+        """Drain still-open observations for ``invocation_id``.
+
+        Synthetic dispatch observations only close on cursor-move when
+        a subsequent event arrives with a different namespace prefix.
+        For a subgraph or fan-out that's the last subtree of an
+        invocation, no follow-up event triggers the close — this
+        method walks the per-invocation state and ends anything left
+        in child→parent order so the Langfuse-side observations don't
+        stay perpetually in-flight.
+
+        Idempotent: calling twice (or for an invocation_id with no
+        open state) is a no-op.
+        """
+        inv_state = self._inv_states.pop(invocation_id, None)
+        if inv_state is None:
+            return
+        # Order: deepest leaves first so parents see all children
+        # closed before they end. LLM observations → leaf nodes
+        # (sorted deepest-first by namespace length) → per-instance
+        # fan-out dispatches → subgraph dispatches.
+        for call_id in list(inv_state.open_llm_observations.keys()):
+            obs = inv_state.open_llm_observations.pop(call_id, None)
+            if obs is not None:
+                obs.handle.end()
+        for key in sorted(
+            inv_state.open_observations.keys(),
+            key=lambda k: -len(k[0]),
+        ):
+            obs = inv_state.open_observations.pop(key, None)
+            if obs is not None:
+                obs.handle.end()
+        for prefix in list(inv_state.fan_out_instance_observations.keys()):
+            self._close_fan_out_instance_dispatch_observation(inv_state, prefix)
+        for prefix in sorted(
+            inv_state.subgraph_observations.keys(),
+            key=lambda p: -len(p),
+        ):
+            self._close_subgraph_observation(inv_state, prefix)
+
+    def shutdown(self) -> None:
+        """Drain every in-flight invocation. Use for long-lived
+        observers shared across requests; CLI / one-shot processes
+        typically call this from a ``finally`` block alongside
+        ``compiled.drain()``.
+        """
+        for invocation_id in list(self._inv_states.keys()):
+            self.close_invocation(invocation_id)
+
     def _observation_metadata(self, event: NodeEvent, correlation_id: str | None) -> dict[str, Any]:
         # §8.4.2 observation-level mapping. Fields below mirror the
         # OTel observer's _node_attrs() output, renamed for Langfuse's
@@ -783,11 +848,16 @@ def _resolve_llm_parent_observation_id(
         self, inv_state: _InvState, payload: LlmEventPayload
     ) -> str | None:
         # Calling-node identity comes from the payload (set at
-        # dispatch time per llm-provider §5.5). Try the exact-match
-        # leaf node first; if absent (LLM call fired from a wrapper
-        # node only, or the calling node already completed), fall
-        # back to the per-instance / subgraph dispatch chain, then to
-        # the leaf ancestor walk.
+        # dispatch time per llm-provider §5.5). Precedence:
+        #   1. Exact-match leaf node at the calling key.
+        #   2. Per-instance fan-out dispatch observation when the
+        #      call originated inside a fan-out instance.
+        #   3. Subgraph dispatch observations along the calling
+        #      namespace prefix, walked longest-prefix-first.
+        #   4. None — Trace becomes the implicit parent.
+        # The dispatch fallbacks cover the wrapped-call cases the
+        # exact-match miss would otherwise need a leaf-ancestor walk
+        # to handle.
         key: _StackKey = (
             payload.calling_namespace_prefix,
             payload.calling_attempt_index,

tests/unit/test_observability_langfuse.py

@@ -314,11 +314,37 @@ async def _worker(_s: _WorkerState) -> Any:
     # The fan-out node's metadata accumulates all 3 detached trace ids.
     link_ids = fan_node.metadata.get("detached_child_trace_ids")
     assert isinstance(link_ids, list)
-    assert set(cast("list[str]", link_ids)) == {t.id for t in detached_traces}
+    assert set(cast(list[str], link_ids)) == {t.id for t in detached_traces}
 
     # Each detached Trace has its own per-instance dispatch with a
     # worker observation under it.
     for t in detached_traces:
         dispatch = _find_observation(t, "fan")
         worker = _find_observation(t, "worker")
         assert worker.parent_observation_id == dispatch.id
+
+
+async def test_subgraph_dispatch_observation_ended_on_invocation_close() -> None:
+    # Synthetic dispatch observations close on cursor-move; without
+    # the close_invocation drain a subgraph at the tail of an
+    # invocation would leave its dispatch in-flight forever. Verifies
+    # the drain path ends everything.
+    inner = (
+        GraphBuilder(_S)
+        .add_node("inner_a", lambda _s: _record("inner_a"))
+        .add_edge("inner_a", END)
+        .set_entry("inner_a")
+        .compile()
+    )
+    parent = GraphBuilder(_S).add_subgraph_node("sub", inner).add_edge("sub", END).set_entry("sub").compile()
+    graph, client, observer = _attach(parent)
+
+    await graph.invoke(_S())
+    await graph.drain()
+    # Without explicit close_invocation, the sub dispatch would still
+    # be in-flight (ended=False). Call shutdown() to drain.
+    observer.shutdown()
+
+    trace = next(iter(client.traces.values()))
+    for obs in trace.observations:
+        assert obs.ended, f"observation {obs.name!r} not ended after shutdown()"