@@ -1665,6 +1665,89 @@ async def _double(s: _ChildState) -> dict[str, int]:
16651665 assert len (double_spans ) == 5 , f"expected 5 per-instance node spans; got { len (double_spans )} "
16661666
16671667
1668+ async def test_detached_fan_out_instance_error_status_on_both_spans () -> None :
1669+ # Proposal 0061 / §4.2 (v0.15.0 release-review item 2): when a DETACHED
1670+ # fan-out instance's subgraph raises, ERROR surfaces on BOTH the parent's
1671+ # fan-out node span (parent trace) and the detached instance's invocation
1672+ # span (its own trace, carrying the §4 category + an OTel exception event).
1673+ # This is the fan-out analog of the detached-subgraph case (conformance
1674+ # fixture 008 case 3 covers only the subgraph path).
1675+ from opentelemetry .trace import StatusCode
1676+
1677+ from openarmature .graph import RuntimeGraphError
1678+
1679+ class _ParentState (State ):
1680+ items : list [int ] = Field (default_factory = list [int ])
1681+ results : list [int ] = Field (default_factory = list [int ])
1682+
1683+ class _ChildState (State ):
1684+ item : int = 0
1685+ out : int = 0
1686+
1687+ async def _raise (_s : _ChildState ) -> dict [str , int ]:
1688+ raise RuntimeError ("boom" )
1689+
1690+ inner = (
1691+ GraphBuilder (_ChildState )
1692+ .add_node ("compute" , _raise )
1693+ .add_edge ("compute" , END )
1694+ .set_entry ("compute" )
1695+ .compile ()
1696+ )
1697+ parent = (
1698+ GraphBuilder (_ParentState )
1699+ .add_fan_out_node (
1700+ "score" ,
1701+ subgraph = inner ,
1702+ collect_field = "out" ,
1703+ target_field = "results" ,
1704+ items_field = "items" ,
1705+ item_field = "item" ,
1706+ )
1707+ .add_edge ("score" , END )
1708+ .set_entry ("score" )
1709+ .compile ()
1710+ )
1711+ exporter = InMemorySpanExporter ()
1712+ observer = OTelObserver (
1713+ span_processor = SimpleSpanProcessor (exporter ),
1714+ detached_fan_outs = frozenset ({"score" }),
1715+ )
1716+ parent .attach_observer (observer )
1717+ with pytest .raises (RuntimeGraphError ):
1718+ await parent .invoke (_ParentState (items = [5 ]))
1719+ await parent .drain ()
1720+ observer .shutdown ()
1721+
1722+ spans = exporter .get_finished_spans ()
1723+ # The detached instance trace is the one containing the raising inner
1724+ # node (`score` appears in BOTH traces -- the parent fan-out node span
1725+ # and the detached instance's own root span -- so key off `compute`).
1726+ compute = next (s for s in spans if s .name == "compute" )
1727+ assert compute .context is not None
1728+ detached_trace_id = compute .context .trace_id
1729+ # §4.2: the detached instance's invocation span is its own trace's
1730+ # authoritative carrier -- ERROR + the §4 category + an OTel exception
1731+ # event (mirroring the detached-subgraph case).
1732+ detached_inv = next (
1733+ s
1734+ for s in spans
1735+ if s .name == "openarmature.invocation"
1736+ and s .context is not None
1737+ and s .context .trace_id == detached_trace_id
1738+ )
1739+ assert detached_inv .status .status_code == StatusCode .ERROR
1740+ assert dict (detached_inv .attributes or {}).get ("openarmature.error.category" ) == "node_exception"
1741+ assert any (e .name == "exception" for e in detached_inv .events )
1742+ # And the parent-trace fan-out node span (the §4.4 Link carrier) is ERROR.
1743+ parent_score = next (
1744+ s
1745+ for s in spans
1746+ if s .name == "score" and s .context is not None and s .context .trace_id != detached_trace_id
1747+ )
1748+ assert parent_score .status .status_code == StatusCode .ERROR
1749+
1750+
16681751async def test_concurrent_fan_out_llm_spans_parent_under_calling_instance () -> None :
16691752 """Under concurrent fan-out: each instance's
16701753 ``openarmature.llm.complete`` span MUST parent under that
0 commit comments