Address PR 85 review: header tweak + two coverage tests

Updates the module-header comment in
tests/conformance/test_observability_langfuse.py to clarify that
the harness *infrastructure* supports the 031/032/033 topology
shapes while activation is currently deferred. Drops the
forward-reference to fixture 027 in an internal comment that
referenced fixtures this module doesn't currently run.

Adds two focused unit tests for the observer behavior changes that
were only exercised by deferred conformance fixtures:

- test_entry_node_resolves_to_wrapper_when_entry_is_subgraph pins
  the entry_node correctness fix: when the outer entry IS a
  SubgraphNode, the first event arrives from inside the subgraph
  (event.namespace = (wrapper, inner), node_name = inner). Both
  trace.name and trace.metadata.entry_node must resolve to the
  wrapper, not the inner node.

- test_detached_subgraph_subgraph_name_placement pins the audit
  fix from coord-thread msg 07: in detached mode the wrapper role
  migrates to the detached trace. The parent trace's link
  observation must not carry subgraph_name; the detached trace's
  dispatch observation must.

Both tests sit alongside the existing dispatch-synthesis tests.

Author: chris-colinsky

tests/conformance/test_observability_langfuse.py

@@ -1,8 +1,15 @@
-# Spec mapping (observability §8): drives the Langfuse mapping fixtures
-# (022/023/024 basic/generation/prompt-linkage, plus 031/032/033 graph
-# topology) against the in-memory LangfuseObserver client. Sibling of
-# test_observability.py (OTel mapping); shares no harness state with
-# the OTel side — each fixture builds its own graph + observer instance.
+# Spec mapping (observability §8): drives the Langfuse mapping
+# fixtures (022 basic-trace, 023 generation-rendering, 024
+# prompt-linkage) against the in-memory LangfuseObserver client.
+# Sibling of test_observability.py (OTel mapping); shares no harness
+# state with the OTel side — each fixture builds its own graph +
+# observer instance.
+#
+# The harness also supports the graph-topology shapes used by
+# 031/032/033 (subgraph / fan-out / detached-trace) via the
+# cross-capability adapter.build_graph helper, but activation of
+# those three fixtures is currently deferred — see the
+# `_LANGFUSE_FIXTURES` frozenset comment for the gating questions.
 
 """Run spec observability Langfuse conformance fixtures."""
 
@@ -245,7 +252,7 @@ async def _run_case(case: Mapping[str, Any]) -> None:
     # ---- Graph build
     # Two paths: topology fixtures (031/032/033) need the full
     # ``adapter.build_graph`` machinery for subgraph / fan_out shapes;
-    # LLM/prompt fixtures (022/023/024/027) use the simpler hand-rolled
+    # LLM/prompt fixtures (022/023/024) use the simpler hand-rolled
     # per-node build that knows about ``calls_llm`` / ``renders_prompt``.
     if _has_topology_constructs(case):
         subgraphs = _compile_subgraphs(case)

tests/unit/test_observability_langfuse.py

@@ -196,6 +196,32 @@ def _find_observation(trace: LangfuseTrace, name: str) -> LangfuseObservation:
     raise AssertionError(f"observation {name!r} not in trace {trace.id!r}")
 
 
+async def test_entry_node_resolves_to_wrapper_when_entry_is_subgraph() -> None:
+    # When the outer entry IS a SubgraphNode, the first event the
+    # observer sees comes from inside the subgraph
+    # (event.namespace = (wrapper, inner), event.node_name = inner).
+    # `entry_node` and trace.name MUST resolve to the wrapper node
+    # name (event.namespace[0]), not the inner node name.
+    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, _ = _attach(parent)
+
+    await graph.invoke(_S())
+    await graph.drain()
+
+    trace = next(iter(client.traces.values()))
+    assert trace.name == "sub", f"trace name should be the wrapper, got {trace.name!r}"
+    assert trace.metadata.get("entry_node") == "sub", (
+        f"entry_node should be the wrapper, got {trace.metadata.get('entry_node')!r}"
+    )
+
+
 async def test_subgraph_dispatch_observation_parents_inner_node() -> None:
     inner = (
         GraphBuilder(_S)
@@ -294,6 +320,40 @@ async def test_detached_subgraph_opens_separate_trace() -> None:
     assert inner_node.parent_observation_id == detached_dispatch.id
 
 
+async def test_detached_subgraph_subgraph_name_placement() -> None:
+    # Per coord thread `discuss-observability-langfuse-mapping` msg 07
+    # and the wrapper-role-migration framing: in detached mode the
+    # wrapper role migrates to the detached trace. The parent trace's
+    # link observation IS the SubgraphNode span (no wrapper role) and
+    # MUST NOT carry `subgraph_name`. The detached trace's dispatch
+    # observation IS the migrated wrapper and MUST carry it.
+    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, _ = _attach_with_detached(parent, detached_subgraphs=frozenset({"sub"}))
+
+    await graph.invoke(_S())
+    await graph.drain()
+
+    main = next(t for t in client.traces.values() if "detached_from_invocation_id" not in t.metadata)
+    detached = next(t for t in client.traces.values() if "detached_from_invocation_id" in t.metadata)
+
+    link_obs = _find_observation(main, "sub")
+    assert "subgraph_name" not in link_obs.metadata, (
+        f"link observation MUST NOT carry subgraph_name; got {link_obs.metadata!r}"
+    )
+
+    detached_dispatch = _find_observation(detached, "sub")
+    assert "subgraph_name" in detached_dispatch.metadata, (
+        f"detached dispatch MUST carry subgraph_name; got {detached_dispatch.metadata!r}"
+    )
+
+
 async def test_detached_fan_out_each_instance_gets_trace() -> None:
     async def _worker(_s: _WorkerState) -> Any:
         return {"result": "done"}