diff --git a/CHANGELOG.md b/CHANGELOG.md index 793a31c..e07369b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,6 +38,8 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). The ### Fixed +- **A detached fan-out instance no longer opens a second root that abandons the first** (observability §4.4). A detached fan-out's per-instance root is stored under `prefix + (instance index)`, but the ancestor walk that decides whether to open one tested the bare `prefix`, so the test never matched and the arm fired once per inner **node event** rather than once per instance. Each additional inner node re-opened the root, and the second open replaced both the root and its detached invocation span in the observer's state, so the first pair was never ended and never exported. The result for a caller: three traces where there should be two, one instance's inner nodes split across two of them, and spans carrying a parent id that nothing emitted. It needs two or more nodes in the instance subgraph to see, which is why it went unnoticed: with a single node there is one event and one open, and every existing test used that shape. The guard now tests the key the root is actually stored under. The neighbouring bare-prefix test is kept for depth and is documented as redundant rather than load-bearing, since the detached-subgraph path is already guarded a line earlier. + - **A dispatch span synthesized from a wrapper-issued call no longer loses its subgraph identity or its caller metadata** (observability §5.4 / §5.5 / §5.6, graph-engine §6). When a fan-out's `instance_middleware` (or a parallel branch's) issues a provider or tool call and then returns without calling `next_call`, no inner node event is ever emitted, so the dispatch span the observer synthesizes from that call had nothing to repair it. Two attributes went missing and stayed missing. `openarmature.subgraph.name` was empty despite a declared `subgraph_identity`, because the identity reached an observer only through an inner node event; `fan_out_config` now carries an optional `subgraph_identity` alongside its four required keys, and the observer caches it from the fan-out node's own `started` event, which always precedes its instances. The cross-cutting `openarmature.user.*` set was absent whenever a `FailureIsolatedEvent` was the event that synthesized the span, which happens when a wrapper sets metadata and then raises without making a provider call first; that event now carries the same optional `caller_invocation_metadata` the provider events carry, populated in the engine task, because the metadata is per-async-context per §3.4 and an observer resolves on the serial delivery queue where a live read sees the wrong context. The same change puts the cross-cutting set on the `openarmature.failure_isolated` marker span, which §5.6 requires on every span in the invocation and which previously carried none. Both event-surface additions are additive: graph-engine §6 requires `fan_out_config` to present all four of its keys and does not close the set, and `caller_invocation_metadata` is already an optional field on the provider events. They ship ahead of any spec text, unit-tested, with standardization requested so a second implementation matches. - **A fan-out inside a parallel branch no longer returns a sibling branch's results** (pipeline-utilities §9 / §11). **Correctness fix; a graph could return data it never computed.** The engine keyed a fan-out's in-flight execution state by its namespace, node name, and enclosing fan-out-instance lineage. A parallel branch contributes to none of those: branch names never enter the namespace, and a branch descent adds only `null` entries to the fan-out-instance chain. So two sibling branches whose subgraphs each held a fan-out node of the same name built an identical key, and the second branch found the first's instances already `completed` and rolled its results forward. It executed none of its own item bodies and emitted no inner node events, so nothing in a trace showed the work had not happened. Where the two branches resolved different item counts the collision surfaced instead as a `CheckpointRecordInvalid` raised on a fresh run with no checkpointer attached, complaining about a saved record that did not exist. The state key now also carries the enclosing branch lineage, so sibling branches are distinct. The defect was schedule-dependent, and a single `await` in an item body was enough to hide it. Note for checkpointed graphs: a branch-nested fan-out now re-runs every instance on resume rather than skipping completed ones, because the checkpoint record has no field that distinguishes sibling branches and applying one branch's skips to another would be worse than re-running. Resume stays correct for this shape and saves no work; lifting that needs a record-format change, and is under discussion. diff --git a/src/openarmature/observability/otel/observer.py b/src/openarmature/observability/otel/observer.py index 9152365..fe1fdbd 100644 --- a/src/openarmature/observability/otel/observer.py +++ b/src/openarmature/observability/otel/observer.py @@ -2658,6 +2658,32 @@ def _sync_subgraph_spans( prefix = namespace[:depth] if prefix in inv_state.subgraph_spans: continue + # `detached_roots` holds TWO key shapes. A detached SUBGRAPH root is + # stored under the bare prefix; a detached FAN-OUT INSTANCE root is + # stored under `prefix + (str(fan_out_index),)`. + # + # The bare prefix therefore never matched an instance root, so the + # instance arm below re-fired once per inner node event instead of + # once per instance. The second open replaced both dict entries, + # leaving the first root and its detached invocation span unended + # and unexported, and splitting one instance's inner nodes across + # traces with the first pointing at a parent nothing emitted. + # Visible only with two or more nodes in the instance subgraph: with + # one there is a single event and a single open. The instance arm + # now tests its OWN key. + # + # This test is REDUNDANT today, and is kept as depth rather than + # because anything reaches it. `_open_detached_subgraph_root` writes + # `subgraph_spans[prefix]` as well as `detached_roots[prefix]`, and + # the `subgraph_spans` guard immediately above runs first, so this + # one can never be the guard that fires. Established by mutation: + # deleting this leaves the whole suite green, while deleting the + # `subgraph_spans` guard is caught by conformance fixture 002. + # + # It is retained because a future path that populates + # `detached_roots` without `subgraph_spans` would need it, and the + # #279 rework may add exactly that. Delete it if that stops being + # plausible -- but do not keep it believing it is load-bearing. if prefix in inv_state.detached_roots: continue # The fan-out instance axis at THIS depth -- the chain entry for the @@ -2704,9 +2730,12 @@ def _sync_subgraph_spans( # fan-out (event.fan_out_index populated, fan-out NODE # name at ``prefix[-1]`` in the configured set). if event.fan_out_index is not None and prefix[-1] in self.detached_fan_outs: - self._open_detached_fan_out_instance_root( - inv_state, invocation_id, correlation_id, prefix, event - ) + # Guarded on the key the opener actually stores under, not the + # bare prefix the loop tests above. + if prefix + (str(event.fan_out_index),) not in inv_state.detached_roots: + self._open_detached_fan_out_instance_root( + inv_state, invocation_id, correlation_id, prefix, event + ) continue # Per spec §5.4 + proposal 0013: non-detached fan-out # instances get a synthetic per-instance dispatch span diff --git a/tests/unit/test_observability_otel.py b/tests/unit/test_observability_otel.py index 7c1b335..973f8e6 100644 --- a/tests/unit/test_observability_otel.py +++ b/tests/unit/test_observability_otel.py @@ -39,6 +39,7 @@ from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( InMemorySpanExporter, ) +from opentelemetry.trace import SpanContext from openarmature.checkpoint import InMemoryCheckpointer from openarmature.graph import ( @@ -6353,3 +6354,115 @@ def _sub(identity: str) -> CompiledGraph[Any]: assert identities == ["identity_a", "identity_b"], ( f"each branch's fan-out must keep its OWN declared identity, got {identities}" ) + + +def _span_ctx(span: ReadableSpan) -> SpanContext: + """The span's context. + + Optional on the base type; a finished SDK span always has one. Asserting + says so and checks it, where a cast would only silence the type checker. + Matches the `assert s.context is not None` idiom used elsewhere in this file. + """ + ctx = span.context + assert ctx is not None, f"finished span {span.name!r} carries no SpanContext" + return ctx + + +async def test_detached_fan_out_instance_opens_one_root_per_instance() -> None: + # `_sync_subgraph_spans` guards its detached fan-out arm with + # `if prefix in inv_state.detached_roots`, the BARE prefix, while + # `_open_detached_fan_out_instance_root` stores under + # `prefix + (str(fan_out_index),)`. The guard therefore never matches, and + # the arm fires once per INNER NODE EVENT rather than once per instance. + # + # With a single-node instance subgraph it is invisible: one event, one open. + # Two nodes is enough to see it, which is an ordinary graph rather than an + # exotic one. The second open replaces the first in both dicts, so the first + # root and its detached invocation span are never ended and never exported, + # and the inner nodes end up in different traces with the first pointing at + # a parent that does not exist. + # + # Measured before the fix: 3 traces instead of 2, node `a` alone in its own + # trace, and two spans carrying a parent span id that was never exported. + # + # Asserted through the exported spans rather than by counting calls, so the + # test states the trace-shape guarantee a consumer actually depends on + # rather than an implementation detail. + class _Top(State): + items: list[int] = [0] + out: list[int] = [] + + class _Leaf(State): + item: int = 0 + result: int = 0 + + async def _one(s: _Leaf) -> dict[str, Any]: + return {"result": s.item + 1} + + async def _two(s: _Leaf) -> dict[str, Any]: + return {"result": s.result + 1} + + leaf = ( + GraphBuilder(_Leaf) + .add_node("a", _one) + .add_node("b", _two) + .add_edge("a", "b") + .add_edge("b", END) + .set_entry("a") + .compile() + ) + graph = ( + GraphBuilder(_Top) + .add_fan_out_node( + "fo", + subgraph=leaf, + items_field="items", + item_field="item", + collect_field="result", + target_field="out", + ) + .add_edge("fo", END) + .set_entry("fo") + .compile() + ) + exporter = InMemorySpanExporter() + observer = OTelObserver( + span_processor=SimpleSpanProcessor(exporter), + detached_fan_outs=frozenset({"fo"}), + ) + graph.attach_observer(observer) + try: + await graph.invoke(_Top()) + await graph.drain() + finally: + observer.shutdown() + + spans = list(exporter.get_finished_spans()) + by_name = {s.name for s in spans} + # Non-vacuity: the shape really ran and really detached. + assert {"a", "b"} <= by_name, sorted(by_name) + + # Every referenced parent must itself have been exported. An abandoned root + # leaves its children pointing at a span id nothing emitted. + # + # `openarmature.invocation` is excluded deliberately, not to make the test + # pass: a detached invocation span is parented under a synthetic + # `NonRecordingSpan` that exists only to carry the new trace id and is never + # exported by design, with the fan-out node span holding a Link to it + # instead. Including it would flag that design rather than a defect. + exported = {_span_ctx(s).span_id for s in spans} + dangling = sorted( + s.name + for s in spans + if s.name != "openarmature.invocation" and getattr(s.parent, "span_id", None) not in (None, *exported) + ) + assert dangling == [], f"spans whose parent span was never exported: {dangling}" + + # One instance means one detached trace: its inner nodes belong together, and + # there is no third trace holding an abandoned root's leftovers. + trace_of = {s.name: _span_ctx(s).trace_id for s in spans} + assert trace_of["a"] == trace_of["b"], ( + "the two inner nodes of one fan-out instance landed in different traces" + ) + trace_count = len({_span_ctx(s).trace_id for s in spans}) + assert trace_count == 2, f"expected the parent trace plus one detached instance trace, got {trace_count}"