From 7cb312600f92506da7d04517dfa10e192ff02dea Mon Sep 17 00:00:00 2001
From: chris-colinsky <chris@lunarcommand.xyz>
Date: Fri, 16 Oct 2026 19:29:12 -0700
Subject: [PATCH 1/5] Wire the observability per-fixture driver keys

The OTel drivers for fixtures 004, 005, 009 and 011 asserted their
claims with hardcoded values instead of reading the fixture, so an
edited fixture value went unnoticed. Each now reads its expected keys
and is registered in _DRIVER_EXPECTED_KEYS, so a key it does not read
fails the run.

- 004 matches the declared span_tree and reads no_edge_spans as a
  closed world over span names.
- 005 matches span_tree, span_tree_private and span_tree_global, and
  reads no_llm_provider_span and no_openarmature_spans_on_global. The
  node emits the external span itself, as also_emits_via_global_tracer
  declares, and a leak to the global provider is found by
  instrumentation scope rather than span name.
- 009 runs every case through one driver. It matches span_tree, counts
  traces against invocation_count and checks each correlation_id
  invariant by name. A `<uuid>` placeholder now matches any canonical
  UUIDv4 (conformance-adapter 5.10).
- 011 compares the runs and aspects determinism_check names, applies
  the ignore_attributes patterns and checks its two invariants.

The Langfuse runner now reads 037's resume.from_first_run and builds
039 case 3 from the fixture's subgraph_call and fan_out declarations.

The expected-key guard runs after the hollow-case check, as in the
Langfuse runner, so an invariants-only case is reported as hollow. The
harness fidelity test now also fails when a registered driver is still
listed as unguarded. The nested-key ledger drops the ten keys these
drivers now read.
---
 tests/conformance/harness/vocabulary.py       |  14 +-
 tests/conformance/test_harness_fidelity.py    |  11 +-
 tests/conformance/test_observability.py       | 475 ++++++++++--------
 .../test_observability_langfuse.py            |  67 ++-
 4 files changed, 310 insertions(+), 257 deletions(-)

diff --git a/tests/conformance/harness/vocabulary.py b/tests/conformance/harness/vocabulary.py
index 033e370..73e8dd7 100644
--- a/tests/conformance/harness/vocabulary.py
+++ b/tests/conformance/harness/vocabulary.py
@@ -287,19 +287,9 @@
     # post_state, which graph-engine section 6 dispatch timing and fixture 061 do
     # not allow. Awaiting a spec ruling.
     ("expected", "expected_observer_event"): ("006-middleware-error-recovery",),
-    # observability per-fixture drivers. Some assert the claim by hardcoding it
-    # rather than reading the key, so a changed fixture value would go unnoticed.
-    ("nodes.*", "also_emits_via_global_tracer"): ("005-otel-llm-provider-span-nested",),
-    ("expected", "determinism_check"): ("011-otel-determinism",),
-    ("resume", "from_first_run"): ("037-langfuse-trace-input-output",),
-    ("expected", "invocation_count"): ("009-otel-correlation-id-cross-cutting",),
-    ("expected", "no_edge_spans"): ("004-otel-routing-error-attribution",),
-    ("expected", "no_llm_provider_span"): ("005-otel-llm-provider-span-nested",),
-    ("expected", "no_openarmature_spans_on_global"): ("005-otel-llm-provider-span-nested",),
+    # observability: the detached-trace drivers check their trees by hand rather
+    # than reading these keys, so a changed fixture value would go unnoticed.
     ("expected", "parent_trace"): ("008-otel-detached-trace-mode",),
-    ("expected", "span_tree_global"): ("005-otel-llm-provider-span-nested",),
-    ("expected", "span_tree_private"): ("005-otel-llm-provider-span-nested",),
-    ("nodes.*", "subgraph_call"): ("039-nested-lineage-augmentation",),
     ("expected", "traces"): (
         "008-otel-detached-trace-mode",
         "058-implementation-attribution-otel",
diff --git a/tests/conformance/test_harness_fidelity.py b/tests/conformance/test_harness_fidelity.py
index b599de8..1e72903 100644
--- a/tests/conformance/test_harness_fidelity.py
+++ b/tests/conformance/test_harness_fidelity.py
@@ -32,14 +32,10 @@
         "_run_fixture_001",
         "_run_fixture_002",
         "_run_fixture_003",
-        "_run_fixture_004",
-        "_run_fixture_005",
         "_run_fixture_006",
         "_run_fixture_007",
         "_run_fixture_008",
-        "_run_fixture_009",
         "_run_fixture_010",
-        "_run_fixture_011",
         "_run_fixture_028",
         "_run_fixture_038",
         "_run_fixture_050",
@@ -92,6 +88,13 @@ def test_every_dispatched_driver_is_registered_or_explicitly_unguarded() -> None
     # And the lists must not accumulate names for drivers that no longer exist.
     stale = sorted((_UNGUARDED_SINGLE_FIXTURE_DRIVERS | _UNGUARDED_SHARED_DRIVERS) - dispatched)
     assert not stale, f"unguarded lists name drivers nothing dispatches to: {stale}"
+    # Nor keep a driver once it is registered: the subset check above is satisfied
+    # either way, so the list would go on claiming a guarded driver is unguarded.
+    registered = sorted(
+        set(otel_runner._DRIVER_EXPECTED_KEYS)  # noqa: SLF001
+        & (_UNGUARDED_SINGLE_FIXTURE_DRIVERS | _UNGUARDED_SHARED_DRIVERS)
+    )
+    assert not registered, f"unguarded lists name drivers registered in `_DRIVER_EXPECTED_KEYS`: {registered}"
 
 
 def _invariant_names_read_by(func: object, param: str) -> set[str]:
diff --git a/tests/conformance/test_observability.py b/tests/conformance/test_observability.py
index 02ba41d..ee509c6 100644
--- a/tests/conformance/test_observability.py
+++ b/tests/conformance/test_observability.py
@@ -29,9 +29,11 @@
 from __future__ import annotations
 
 import copy
+import fnmatch
 import functools
 import re
 import uuid
+from collections import Counter
 from collections.abc import Callable, Mapping, Sequence
 from collections.abc import Set as AbstractSet
 from pathlib import Path
@@ -614,6 +616,10 @@ def _reset_otel_global_tracer_provider(restore_to: object) -> None:
 )
 
 
+def _is_uuidv4(value: Any) -> bool:
+    return isinstance(value, str) and _UUIDV4_RE.match(value) is not None
+
+
 def _fixture_paths() -> list[Path]:
     return sorted(CONFORMANCE_DIR.glob("[0-9][0-9][0-9]-*.yaml"))
 
@@ -710,6 +716,18 @@ def _reject_unsupported_capability_gate(fixture_id: str, spec: Mapping[str, Any]
     # _run_llm_payload_case reads span_tree ONLY. Claiming observers is what let
     # 149 be re-routed here with its contains_event half silently dropped.
     "_run_llm_payload_fixture": frozenset({"span_tree", "invariants"}),
+    "_run_fixture_004": frozenset({"span_tree", "no_edge_spans"}),
+    "_run_fixture_005": frozenset(
+        {
+            "span_tree",
+            "span_tree_private",
+            "span_tree_global",
+            "no_llm_provider_span",
+            "no_openarmature_spans_on_global",
+        }
+    ),
+    "_run_fixture_009": frozenset({"span_tree", "invocation_count", "invariants"}),
+    "_run_fixture_011": frozenset({"determinism_check", "invariants"}),
 }
 
 
@@ -792,8 +810,6 @@ async def test_observability_fixture(fixture_path: Path) -> None:
     _reject_unsupported_capability_gate(fixture_id, spec)
     from .harness.capabilities import assert_case_asserts_something  # noqa: PLC0415
 
-    _assert_driver_reads_every_expected_key(fixture_id, spec)
-
     # `or [spec]` because a single-case fixture puts `expected` at the top
     # level with no `cases` list; `or []` skipped seven of them entirely.
     for _case in cast("list[dict[str, Any]]", spec.get("cases") or [spec]):
@@ -802,6 +818,9 @@ async def test_observability_fixture(fixture_path: Path) -> None:
             cast("str", _case.get("name") or "<unnamed>"),
             cast("Mapping[str, Any]", _case.get("expected") or {}),
         )
+    # After the hollow-case check, as in the Langfuse runner, so an invariants-only
+    # case on a registered driver is reported as hollow, not as an unread key.
+    _assert_driver_reads_every_expected_key(fixture_id, spec)
     if fixture_id == "001-otel-basic-trace":
         await _run_fixture_001(spec)
     elif fixture_id == "002-otel-subgraph-hierarchy":
@@ -1234,13 +1253,8 @@ async def _run_fixture_004(spec: Mapping[str, Any]) -> None:
     ``_handle_completed`` ERROR-mapping path picks this up
     automatically — no observer-side change needed for the swap.
 
-    Driver verifies: the ``pick`` node's span ends ERROR with
-    ``status_description == "routing_error"``, an ``exception``
-    event recorded, and the ``openarmature.error.category``
-    attribute. No span for the edge function (no ``edge_spans``) —
-    edge logic is folded into the preceding node span."""
-    from opentelemetry.trace import StatusCode
-
+    Driver verifies: the declared ``span_tree``, including the ERROR
+    span's description and exception event, and ``no_edge_spans``."""
     from openarmature.graph import RuntimeGraphError
 
     observer, exporter = _build_observer()
@@ -1255,51 +1269,31 @@ async def _run_fixture_004(spec: Mapping[str, Any]) -> None:
     await compiled.drain()
     observer.shutdown()
     spans = exporter.get_finished_spans()
+    expected = cast("dict[str, Any]", spec["expected"])
 
-    by_name = {s.name: s for s in spans}
-
-    pick = by_name.get("pick")
-    assert pick is not None
-    assert pick.status.status_code == StatusCode.ERROR, (
-        f"preceding node 'pick' span MUST be ERROR; got {pick.status.status_code}"
-    )
-    assert pick.status.description == "routing_error", (
-        f"preceding node 'pick' span status_description MUST be 'routing_error'; "
-        f"got {pick.status.description!r}"
-    )
-    pick_attrs = dict(pick.attributes or {})
-    assert pick_attrs.get("openarmature.error.category") == "routing_error"
-    # Exception event recorded on the span via record_exception.
-    exception_events = [e for e in pick.events if e.name == "exception"]
-    event_names = [e.name for e in pick.events]
-    assert len(exception_events) >= 1, (
-        f"'pick' MUST have at least one 'exception' event recorded; got {event_names}"
-    )
+    expected_tree = cast("list[dict[str, Any]]", expected["span_tree"])
+    inv_root = next((s for s in spans if s.name == "openarmature.invocation" and s.parent is None), None)
+    assert inv_root is not None, f"invocation root span missing; got {[s.name for s in spans]}"
+    _assert_span_tree_matches(spans, [inv_root], expected_tree)
+    _assert_error_span_extras(spans, expected_tree)
 
-    # Per fixture 004's "no_edge_spans: true" — the edge function
-    # itself does not produce a separate span; the routing error is
-    # folded into the preceding node's span.
-    edge_span_names = {"edge", "openarmature.edge", "edge_function"}
-    edge_spans = [s for s in spans if s.name in edge_span_names]
-    assert edge_spans == [], (
-        f"there MUST be no separate edge-function spans per §4.2; got {[s.name for s in edge_spans]}"
-    )
+    # §4.2: edge logic folds into the preceding node's span. Checked as a closed
+    # world over span names, so an edge span fails whatever it is called.
+    no_edge_spans = expected.get("no_edge_spans")
+    if no_edge_spans is not None:
+        assert no_edge_spans is True, f"no_edge_spans: only `true` is supported; got {no_edge_spans!r}"
+        allowed = {"openarmature.invocation", *cast("dict[str, Any]", spec["nodes"])}
+        extra = sorted(s.name for s in spans if s.name not in allowed)
+        assert not extra, f"spans that are neither the invocation nor a node (§4.2): {extra}"
 
     # Unreachable nodes never fire spans (they were unreached).
+    span_names = {s.name for s in spans}
     for unreachable in ("unreachable_a", "unreachable_b"):
-        assert unreachable not in by_name, f"{unreachable!r} MUST not produce a span — never reached"
-
-    # Invocation span ends ERROR per the §4.2 invocation-status
-    # propagation contract.
-    inv = by_name.get("openarmature.invocation")
-    assert inv is not None
-    assert inv.status.status_code == StatusCode.ERROR, (
-        f"invocation span MUST end ERROR when a child errors; got {inv.status.status_code}"
-    )
+        assert unreachable not in span_names, f"{unreachable!r} MUST not produce a span — never reached"
 
 
 # ---------------------------------------------------------------------------
-# Fixture 007 — retry attempt spans
+# Fixture 006 — fan-out instance attribution
 # ---------------------------------------------------------------------------
 
 
@@ -1400,6 +1394,11 @@ async def _run_fixture_006_case(case: Mapping[str, Any], subgraphs: Mapping[str,
         )
 
 
+# ---------------------------------------------------------------------------
+# Fixture 007 — retry attempt spans
+# ---------------------------------------------------------------------------
+
+
 async def _run_fixture_007(spec: Mapping[str, Any]) -> None:
     """Two sub-cases:
 
@@ -1542,31 +1541,19 @@ def _classifier(exc: Exception, _state: Any, _transient: frozenset[str] = transi
 # ---------------------------------------------------------------------------
 
 
-# Spec-canonical attributes that are non-deterministic by design and
-# MUST be excluded from determinism-comparison runs. Per spec
-# coordination on the fixture (08-spec-prep-sync-confirmed reaffirms
-# the §3.2 + §5.1 distinction): caller-supplied correlation_id is
-# deterministic; auto-generated UUIDv4 is not. Fixture 011 omits
-# ``caller_correlation_id``, so the auto-generated correlation_id IS
-# in the ignore set for this fixture.
-_DETERMINISM_IGNORED_ATTRS: frozenset[str] = frozenset(
-    {
-        "openarmature.invocation_id",
-        "openarmature.correlation_id",
-    }
+# Each `determinism_check.compare` aspect names one part of a span's signature.
+_DETERMINISM_ASPECTS = frozenset(
+    {"span_tree_structure", "span_names", "span_status", "span_attributes_excluding_ignored"}
 )
+_DETERMINISM_INVARIANTS = frozenset({"same_execution_path", "span_count_equal_across_runs"})
 
 
 async def _run_fixture_011(spec: Mapping[str, Any]) -> None:
-    """Deterministic span content is identical across two
-    invocations of the same graph with the same input. The
-    signature compared per-span:
-    ``(name, status_code, parent_name, attrs ∖ ignored_set)``.
-    Parent linkage is encoded as the parent span's NAME rather
-    than its span_id (span_ids are non-deterministic per OTel SDK's
-    default RandomIdGenerator); a hierarchy regression where a
-    node reparented to a different ancestor surfaces as a
-    parent_name divergence."""
+    """Deterministic span content is identical across repeated
+    invocations of the same graph with the same input. Each span's
+    signature holds the parts ``determinism_check.compare`` selects,
+    with ``ignore_attributes`` removed, and the runs it names must
+    produce the same multiset of signatures."""
     cases = cast("list[dict[str, Any]]", spec["cases"])
     for case in cases:
         case_name = cast("str", case["name"])
@@ -1576,6 +1563,34 @@ async def _run_fixture_011(spec: Mapping[str, Any]) -> None:
             raise AssertionError(f"case {case_name!r}: {e}") from e
 
 
+def _determinism_signature(
+    span: Any, by_id: Mapping[int, Any], aspects: AbstractSet[str], ignored: Sequence[str]
+) -> tuple[Any, ...]:
+    parts: list[Any] = []
+    if "span_names" in aspects:
+        parts.append(span.name)
+    if "span_status" in aspects:
+        parts.append(span.status.status_code.name)
+    if "span_tree_structure" in aspects:
+        # Span ids are random per run, so the hierarchy is the chain of ancestor
+        # names. A span moved under a different ancestor changes it.
+        ancestors: list[str] = []
+        parent = span.parent
+        while parent is not None and (above := by_id.get(parent.span_id)) is not None:
+            ancestors.append(cast("str", above.name))
+            parent = above.parent
+        parts.append(tuple(ancestors))
+    if "span_attributes_excluding_ignored" in aspects:
+        attrs = dict(span.attributes or {})
+        kept = (
+            (k, _normalize_attr_value(v))
+            for k, v in attrs.items()
+            if not any(fnmatch.fnmatchcase(k, pattern) for pattern in ignored)
+        )
+        parts.append(tuple(sorted(kept)))
+    return tuple(parts)
+
+
 async def _run_fixture_011_case(case: Mapping[str, Any]) -> None:
     # Translate the fixture's ``when:`` conditional-edge syntax
     # (``when: {field: counter, gt: 0}``) into the adapter's
@@ -1588,11 +1603,30 @@ async def _run_fixture_011_case(case: Mapping[str, Any]) -> None:
     # branch always fires under identical inputs. (The adapter has
     # no generic ``gt``/``lt`` edge conditions.)
     case_for_build = _translate_011_when_edges(case)
-
     invocations = int(case.get("invocations", 2))
-    assert invocations == 2, f"fixture 011: expected invocations=2; got {invocations}"
+
+    expected = cast("dict[str, Any]", case["expected"])
+    check = cast("dict[str, Any]", expected["determinism_check"])
+    compare_runs = cast("list[int]", check["compare_runs"])
+    assert len(set(compare_runs)) >= 2 and all(0 <= i < invocations for i in compare_runs), (
+        f"determinism_check.compare_runs {compare_runs} must name two or more of the {invocations} runs"
+    )
+    ignored = cast("list[str]", check.get("ignore_attributes") or [])
+    aspects = frozenset(cast("list[str]", check["compare"]))
+    unknown_aspects = sorted(aspects - _DETERMINISM_ASPECTS)
+    assert not unknown_aspects, f"determinism_check.compare: unsupported aspect(s) {unknown_aspects}"
+    require_equal = check.get("require_equal")
+    assert require_equal is True, (
+        f"determinism_check.require_equal: only `true` is supported; got {require_equal!r}"
+    )
+    invariants = cast("dict[str, Any]", expected.get("invariants") or {})
+    unknown_invariants = sorted(set(invariants) - _DETERMINISM_INVARIANTS)
+    assert not unknown_invariants, f"unsupported invariant(s) {unknown_invariants}"
+    for name, value in invariants.items():
+        assert value is True, f"invariants.{name}: only `true` is supported; got {value!r}"
 
     runs: list[list[Any]] = []
+    paths: list[list[str]] = []
     for _ in range(invocations):
         observer, exporter = _build_observer()
         trace_log: list[str] = []
@@ -1604,52 +1638,27 @@ async def _run_fixture_011_case(case: Mapping[str, Any]) -> None:
         await compiled.drain()
         observer.shutdown()
         runs.append(list(exporter.get_finished_spans()))
+        paths.append(trace_log)
 
-    assert len(runs[0]) == len(runs[1]), (
-        f"deterministic input MUST produce equal span counts; got {len(runs[0])} vs {len(runs[1])}"
-    )
-
-    # Compare each span's structural signature across runs. Span
-    # span_ids are non-deterministic, so we encode the parent
-    # linkage by looking up parent.span_id in the same run's
-    # by-id map and including the parent's NAME in the signature.
-    # That way a hierarchy regression (e.g., a node reparented
-    # from invocation to a sibling) shows up as a signature
-    # difference even though both spans' own attributes are
-    # unchanged.
-    def _signature(
-        span: Any, by_id: Mapping[int, Any]
-    ) -> tuple[str, str, str | None, tuple[tuple[str, Any], ...]]:
-        attrs = dict(span.attributes or {})
-        deterministic_items = sorted(
-            (k, _normalize_attr_value(v)) for k, v in attrs.items() if k not in _DETERMINISM_IGNORED_ATTRS
-        )
-        parent_name: str | None = None
-        if span.parent is not None:
-            parent_span = by_id.get(span.parent.span_id)
-            if parent_span is not None:
-                parent_name = cast("str", parent_span.name)
-        return (
-            cast("str", span.name),
-            str(span.status.status_code),
-            parent_name,
-            tuple(deterministic_items),
-        )
-
-    by_id_run_0: dict[int, Any] = {}
-    for s in runs[0]:
-        if s.context is not None:
-            by_id_run_0[s.context.span_id] = s
-    by_id_run_1: dict[int, Any] = {}
-    for s in runs[1]:
-        if s.context is not None:
-            by_id_run_1[s.context.span_id] = s
-    sig_run_0 = sorted(_signature(s, by_id_run_0) for s in runs[0])
-    sig_run_1 = sorted(_signature(s, by_id_run_1) for s in runs[1])
-    assert sig_run_0 == sig_run_1, (
-        f"deterministic span content MUST match across runs; "
-        f"first divergence: run_0={sig_run_0!r} vs run_1={sig_run_1!r}"
-    )
+    # The invariants are claims across all runs; the adapter's trace records the
+    # nodes each run executed, independent of what the observer emitted.
+    if "same_execution_path" in invariants:
+        assert all(path == paths[0] for path in paths), f"same_execution_path: runs took {paths}"
+    if "span_count_equal_across_runs" in invariants:
+        counts = [len(run) for run in runs]
+        assert len(set(counts)) == 1, f"span_count_equal_across_runs: span counts {counts}"
+
+    signatures: list[Counter[tuple[Any, ...]]] = []
+    for i in compare_runs:
+        by_id = {s.context.span_id: s for s in runs[i] if s.context is not None}
+        signatures.append(Counter(_determinism_signature(s, by_id, aspects, ignored) for s in runs[i]))
+    first = compare_runs[0]
+    for i, signature in zip(compare_runs[1:], signatures[1:], strict=True):
+        assert signature == signatures[0], (
+            f"deterministic span content MUST match across runs; only in run {first}: "
+            f"{sorted(signatures[0] - signature, key=repr)}; only in run {i}: "
+            f"{sorted(signature - signatures[0], key=repr)}"
+        )
 
 
 async def _run_fixture_028(spec: Mapping[str, Any]) -> None:
@@ -2173,65 +2182,84 @@ async def _run_fixture_009(spec: Mapping[str, Any]) -> None:
             raise AssertionError(f"case {case_name!r}: {e}") from e
 
 
-async def _run_fixture_009_case(case: Mapping[str, Any]) -> None:
-    case_name = case["name"]
-    if case_name == "context_reset_between_invocations":
-        # Two back-to-back invocations of the same compiled graph;
-        # each MUST get its own UUIDv4 (distinct from the other).
-        observer, exporter = _build_observer()
-        # Build the graph ONCE so both invocations share it.
-        from .adapter import build_graph as _bg
+# Names `_run_fixture_009_case` implements.
+_CORRELATION_ID_INVARIANTS = frozenset(
+    {
+        "correlation_id_uniform_across_tree",
+        "correlation_id_is_uuidv4",
+        "each_invocation_correlation_id_uniform",
+        "correlation_ids_differ_across_invocations",
+        "all_correlation_ids_uuidv4",
+    }
+)
 
-        built = _bg(case)
-        compiled = built.builder.compile()
-        compiled.attach_observer(observer)
-        for _ in range(int(case.get("invocations", 2))):
-            await compiled.invoke(built.initial_state(case.get("initial_state", {})))
-            await compiled.drain()
-        observer.shutdown()
-        spans = exporter.get_finished_spans()
 
-        # Group spans by trace_id (each invocation has its own trace).
-        by_trace: dict[int, list[Any]] = {}
-        for s in spans:
-            tid = s.context.trace_id
-            by_trace.setdefault(tid, []).append(s)
-        assert len(by_trace) == 2, f"expected 2 distinct traces (one per invocation); got {len(by_trace)}"
-        # Each invocation's spans share one correlation_id.
-        per_invocation_cids: list[str] = []
-        for trace_spans in by_trace.values():
-            cids = _all_correlation_ids(trace_spans)
-            assert len(cids) == 1, f"each invocation MUST uniformly carry one correlation_id; got {cids}"
-            cid = next(iter(cids))
-            assert _UUIDV4_RE.match(cid), f"auto-generated correlation_id MUST be UUIDv4; got {cid!r}"
-            per_invocation_cids.append(cid)
-        # Cross-invocation: distinct.
-        assert per_invocation_cids[0] != per_invocation_cids[1], (
-            "back-to-back invocations MUST get distinct correlation_ids"
-        )
-        return
+async def _run_fixture_009_case(case: Mapping[str, Any]) -> None:
+    expected = cast("dict[str, Any]", case["expected"])
+    _assert_invariants_recognized(case, _CORRELATION_ID_INVARIANTS, "fixture 009")
+    invariants = cast("dict[str, Any]", expected.get("invariants") or {})
+    for name, value in invariants.items():
+        if name != "correlation_id_uniform_across_tree":
+            assert value is True, f"invariants.{name}: only `true` is supported; got {value!r}"
 
-    # Sub-cases 1 & 2 (single-invocation).
     observer, exporter = _build_observer()
-    await _run_graph(case, observer, correlation_id=case.get("caller_correlation_id"))
+    built = build_graph(case)
+    compiled = built.builder.compile()
+    compiled.attach_observer(observer)
+    # Every run shares one compiled graph, so a correlation_id the context
+    # primitive failed to reset would carry into the next invocation.
+    for _ in range(int(case.get("invocations", 1))):
+        await compiled.invoke(
+            built.initial_state(case.get("initial_state", {})),
+            correlation_id=case.get("caller_correlation_id"),
+        )
+        await compiled.drain()
     observer.shutdown()
     spans = exporter.get_finished_spans()
-    cids = _all_correlation_ids(spans)
-    assert len(cids) == 1, f"every span MUST carry the same correlation_id; got {cids}"
-    cid = next(iter(cids))
-    expected = case.get("caller_correlation_id")
-    if expected is not None:
-        # Caller-supplied → exact match.
-        assert cid == expected, (
-            f"caller_correlation_id MUST be used verbatim; got {cid!r}, expected {expected!r}"
+    assert spans, "no spans were recorded, so every correlation_id claim would pass vacuously"
+
+    if (tree := expected.get("span_tree")) is not None:
+        inv_root = next((s for s in spans if s.name == "openarmature.invocation" and s.parent is None), None)
+        assert inv_root is not None, f"invocation root span missing; got {[s.name for s in spans]}"
+        _assert_span_tree_matches(spans, [inv_root], cast("list[dict[str, Any]]", tree))
+
+    # Each invocation is its own trace.
+    by_trace: dict[int, list[Any]] = {}
+    for s in spans:
+        by_trace.setdefault(s.context.trace_id, []).append(s)
+    if (invocation_count := expected.get("invocation_count")) is not None:
+        assert len(by_trace) == int(invocation_count), (
+            f"expected {invocation_count} distinct traces (one per invocation); got {len(by_trace)}"
+        )
+
+    all_cids = _all_correlation_ids(spans)
+    per_trace_cids = [_all_correlation_ids(trace_spans) for trace_spans in by_trace.values()]
+    not_uuidv4 = sorted(repr(c) for c in all_cids if not _is_uuidv4(c))
+    if (uniform := invariants.get("correlation_id_uniform_across_tree")) is not None:
+        assert len(all_cids) == 1, f"every span MUST carry the same correlation_id; got {all_cids}"
+        assert _attr_equal(next(iter(all_cids)), uniform), (
+            f"correlation_id_uniform_across_tree: expected {uniform!r}; got {all_cids}"
+        )
+    if "correlation_id_is_uuidv4" in invariants:
+        assert not not_uuidv4, f"correlation_id MUST be UUIDv4; got {not_uuidv4}"
+    if "all_correlation_ids_uuidv4" in invariants:
+        assert not not_uuidv4, f"every correlation_id MUST be UUIDv4; got {not_uuidv4}"
+    if "each_invocation_correlation_id_uniform" in invariants:
+        assert all(len(cids) == 1 for cids in per_trace_cids), (
+            f"each invocation MUST uniformly carry one correlation_id; got {per_trace_cids}"
+        )
+    if "correlation_ids_differ_across_invocations" in invariants:
+        assert len(per_trace_cids) >= 2, (
+            f"correlation_ids_differ_across_invocations needs two or more traces; got {len(per_trace_cids)}"
+        )
+        flat = [cid for cids in per_trace_cids for cid in cids]
+        assert len(flat) == len(set(flat)), (
+            f"back-to-back invocations MUST get distinct correlation_ids; got {per_trace_cids}"
         )
-    else:
-        # Auto-generated → UUIDv4.
-        assert _UUIDV4_RE.match(cid), f"auto-generated correlation_id MUST be UUIDv4; got {cid!r}"
 
 
 # ---------------------------------------------------------------------------
-# Fixture 005, 008 placeholders — driven below in subsequent commits
+# Fixtures 005, 008, 038, 110, 132 and 133, with the shared span-tree matcher
 # ---------------------------------------------------------------------------
 
 
@@ -2254,10 +2282,21 @@ async def _run_fixture_005(spec: Mapping[str, Any]) -> None:
             raise AssertionError(f"case {case_name!r}: {e}") from e
 
 
+# The tracer the harness's node emits through on the global provider, standing
+# in for an auto-instrumentation library.
+_EXTERNAL_TRACER_SCOPE = "external-instrumentation"
+
+
+def _exporter_roots(spans: Sequence[Any]) -> list[Any]:
+    """The spans whose parent did not reach the same exporter."""
+    ids = {s.context.span_id for s in spans if s.context is not None}
+    return [s for s in spans if s.parent is None or s.parent.span_id not in ids]
+
+
 async def _run_fixture_005_case(case: Mapping[str, Any]) -> None:
-    case_name = case["name"]
     disable_llm_spans = bool(case.get("disable_llm_spans", False))
     caller_global_active = bool(case.get("caller_global_otel_active", False))
+    expected = cast("dict[str, Any]", case["expected"])
 
     from opentelemetry import trace as otel_trace
     from opentelemetry.sdk.trace import TracerProvider
@@ -2291,6 +2330,7 @@ async def _run_fixture_005_case(case: Mapping[str, Any]) -> None:
             once._done = False  # pyright: ignore[reportPrivateUsage]
         otel_trace.set_tracer_provider(global_provider)
 
+    global_spans: Sequence[Any] = ()
     try:
         private_exporter = InMemorySpanExporter()
         observer = OTelObserver(
@@ -2305,29 +2345,12 @@ async def _run_fixture_005_case(case: Mapping[str, Any]) -> None:
         # Drive the graph. The ``calls_llm`` node body reads the mock
         # responses set up in ``_build_graph_with_mock_llm`` (httpx
         # MockTransport keyed off the response queue).
-        initial_state_cls = graph.state_cls
-        final = await graph.invoke(initial_state_cls())
+        await graph.invoke(graph.state_cls())
         await graph.drain()
         observer.shutdown()
         private_spans = private_exporter.get_finished_spans()
-
-        # Sub-case 3: external span emitted by the harness through the
-        # global tracer (simulating auto-instrumentation).
-        if caller_global_active:
-            global_tracer = otel_trace.get_tracer("external-instrumentation")
-            with global_tracer.start_as_current_span("external.llm.call"):
-                pass
-            assert global_exporter is not None
+        if global_exporter is not None:
             global_spans = global_exporter.get_finished_spans()
-            assert len(global_spans) == 1, (
-                f"global exporter MUST see exactly one external span; got {len(global_spans)}"
-            )
-            assert global_spans[0].name == "external.llm.call"
-            # The load-bearing isolation check.
-            for s in global_spans:
-                assert not s.name.startswith("openarmature."), (
-                    f"openarmature spans MUST NOT leak to the global provider; got {s.name!r}"
-                )
     finally:
         if caller_global_active and prior_global is not None:
             # OTel SDK 1.x makes set_tracer_provider one-shot: once a
@@ -2336,44 +2359,39 @@ async def _run_fixture_005_case(case: Mapping[str, Any]) -> None:
             # the global doesn't leak into subsequent tests.
             _reset_otel_global_tracer_provider(prior_global)
 
-    # Common assertions: the LLM span presence/absence + (when
-    # present) attributes + parent-child to the calling node.
-    llm_spans = [s for s in private_spans if s.name == "openarmature.llm.complete"]
-    if disable_llm_spans:
-        assert not llm_spans, (
-            f"disable_llm_spans=True MUST suppress LLM span emission; got {len(llm_spans)} llm spans"
-        )
-        # ask_llm node span still emits.
-        assert any(s.name == "ask_llm" for s in private_spans)
-        return
-
-    # default + external_auto_instrumentation_active — LLM span
-    # MUST emit with the spec §5.5 attributes.
-    assert len(llm_spans) == 1, (
-        f"expected one LLM span; got {len(llm_spans)}: {[s.name for s in private_spans]}"
-    )
-    llm = llm_spans[0]
-    attrs = dict(llm.attributes or {})
-    assert attrs.get("openarmature.llm.model") == "test-model"
-    if case_name == "default":
-        # Sub-case 1 asserts the full attribute set.
-        assert attrs.get("openarmature.llm.finish_reason") == "stop"
-        assert attrs.get("openarmature.llm.usage.prompt_tokens") == 5
-        assert attrs.get("openarmature.llm.usage.completion_tokens") == 1
-        assert attrs.get("openarmature.llm.usage.total_tokens") == 6
-    # Parent: the ask_llm node span.
-    ask_llm = next((s for s in private_spans if s.name == "ask_llm"), None)
-    assert ask_llm is not None, "expected ask_llm node span"
-    llm_parent = llm.parent
-    ask_llm_ctx = ask_llm.context
-    assert llm_parent is not None and ask_llm_ctx is not None
-    assert llm_parent.span_id == ask_llm_ctx.span_id, (
-        "openarmature.llm.complete MUST be parented under the calling node span"
-    )
-    # Final state was updated by the calls_llm node (msg = "hello"
-    # for default; "hi back" for disable_llm_spans path that we
-    # already returned from above).
-    assert "msg" in dir(final)
+    # `span_tree` and `span_tree_private` both describe openarmature's own exporter.
+    for tree in (expected.get("span_tree"), expected.get("span_tree_private")):
+        if tree is not None:
+            _assert_span_tree_matches(
+                private_spans, _exporter_roots(private_spans), cast("list[dict[str, Any]]", tree)
+            )
+    if (tree := expected.get("span_tree_global")) is not None:
+        assert global_exporter is not None, "span_tree_global needs caller_global_otel_active"
+        _assert_span_tree_matches(
+            global_spans, _exporter_roots(global_spans), cast("list[dict[str, Any]]", tree)
+        )
+
+    no_llm_span = expected.get("no_llm_provider_span")
+    if no_llm_span is not None:
+        assert no_llm_span is True, f"no_llm_provider_span: only `true` is supported; got {no_llm_span!r}"
+        llm_spans = [s for s in private_spans if s.name == "openarmature.llm.complete"]
+        assert not llm_spans, f"expected no LLM provider span; got {len(llm_spans)}"
+
+    no_oa_on_global = expected.get("no_openarmature_spans_on_global")
+    if no_oa_on_global is not None:
+        assert no_oa_on_global is True, (
+            f"no_openarmature_spans_on_global: only `true` is supported; got {no_oa_on_global!r}"
+        )
+        assert global_exporter is not None, "no_openarmature_spans_on_global needs caller_global_otel_active"
+        # §6 TracerProvider isolation. Only the harness's external tracer writes to
+        # the global provider here, so any other scope is openarmature's, whatever
+        # the span is named.
+        leaked = [
+            (s.name, getattr(s.instrumentation_scope, "name", None))
+            for s in global_spans
+            if getattr(s.instrumentation_scope, "name", None) != _EXTERNAL_TRACER_SCOPE
+        ]
+        assert not leaked, f"openarmature spans reached the global TracerProvider: {leaked}"
 
 
 def _build_graph_with_mock_llm(case: Mapping[str, Any]) -> tuple[Any, list[Any]]:
@@ -2383,6 +2401,7 @@ def _build_graph_with_mock_llm(case: Mapping[str, Any]) -> tuple[Any, list[Any]]
     import json
 
     import httpx
+    from opentelemetry import trace as otel_trace
 
     from openarmature.graph import GraphBuilder
     from openarmature.llm import OpenAIProvider, UserMessage
@@ -2425,8 +2444,17 @@ def _handler(request: httpx.Request) -> httpx.Response:
         UserMessage(content=m["content"]) for m in messages_spec if m.get("role") == "user"
     ]
 
+    global_span = cast("dict[str, Any] | None", nodes[entry_name].get("also_emits_via_global_tracer"))
+
     async def ask_llm_body(_s: Any) -> dict[str, str]:
-        response = await provider.complete(messages)
+        if global_span is None:
+            response = await provider.complete(messages)
+        else:
+            # Auto-instrumentation wraps the provider call from inside the node,
+            # through whatever the global provider is at that moment.
+            tracer = otel_trace.get_tracer(_EXTERNAL_TRACER_SCOPE)
+            with tracer.start_as_current_span(cast("str", global_span["span_name"])):
+                response = await provider.complete(messages)
         return {stores_in: response.message.content or ""}
 
     builder = (
@@ -2672,6 +2700,9 @@ def _attr_equal(actual: Any, expected: Any) -> bool:
     # through this matcher, so nothing surfaced it before.
     if isinstance(expected, list) and isinstance(actual, tuple):
         return list(cast("tuple[Any, ...]", actual)) == cast("list[Any]", expected)
+    # conformance-adapter §5.10: `<uuid>` matches any canonical UUIDv4.
+    if expected == "<uuid>":
+        return _is_uuidv4(actual)
     return bool(actual == expected)
 
 
diff --git a/tests/conformance/test_observability_langfuse.py b/tests/conformance/test_observability_langfuse.py
index dd4a942..d764e17 100644
--- a/tests/conformance/test_observability_langfuse.py
+++ b/tests/conformance/test_observability_langfuse.py
@@ -1055,16 +1055,29 @@ def _build_039_case3(
     provider: OpenAIProvider | None,
     prompt_manager: PromptManager | None,
 ) -> tuple[Any, Any]:
-    # Case 3: a serial subgraph wrapper (`wrap`) descends into `wrapped_fan_out`,
-    # whose `pick` fan-out runs per-product; each instance augments note=<id>.
-    # The wrapper span must carry the augmentation (last-writer) per 0045's
-    # lineage-aware rule, the fan-out NODE must not.
-    # The fan-out places each outer product into per_product's item_field slot;
-    # the augment middleware reads <id> from it. per_product's declared state
-    # ({picked}) lacks the slot, so inject it (mirrors _synthesize_fan_out_
+    # Case 3: the entry node's `subgraph_call` descends into a serial subgraph
+    # whose entry node fans out per product; each instance augments metadata from
+    # its item. The wrapper span must carry the augmentation (last-writer) per
+    # 0045's lineage-aware rule, the fan-out NODE must not.
+    # The fan-out places each outer product into the inner subgraph's item_field
+    # slot; the augment middleware reads from it. The inner subgraph's declared
+    # state lacks the slot, so inject it (mirrors _synthesize_fan_out_
     # aggregation on the generic 029 path).
     assert provider is not None, "039 cases declare mock_llm, so the provider must be set"
-    per_product_spec = copy.deepcopy(cast("dict[str, Any]", case["inner_subgraphs"]["per_product"]))
+    wrap_name = cast("str", case["entry"])
+    call = cast("dict[str, Any]", case["nodes"][wrap_name]["subgraph_call"])
+    subgraphs = cast("dict[str, Any]", case["subgraphs"])
+    wrapped_name = cast("str", call["subgraph"])
+    assert wrapped_name in subgraphs, (
+        f"subgraph_call names {wrapped_name!r}; the case declares {sorted(subgraphs)}"
+    )
+    wrapped_spec = cast("dict[str, Any]", subgraphs[wrapped_name])
+    fan_out_name = cast("str", wrapped_spec["entry"])
+    fan_out = cast("dict[str, Any]", wrapped_spec["nodes"][fan_out_name]["fan_out"])
+    items_field = cast("str", fan_out["items_field"])
+    per_product_spec = copy.deepcopy(
+        cast("dict[str, Any]", case["inner_subgraphs"][cast("str", fan_out["inner_subgraph"])])
+    )
     per_product_spec.setdefault("state", {}).setdefault("fields", {}).setdefault(
         "oa_fan_out_item", {"type": "dict", "default": {}}
     )
@@ -1074,40 +1087,51 @@ def _build_039_case3(
         prompt_manager=prompt_manager,
         render_variables={},
     )
+    # The fixture under-declares this topology. Its call has no `inputs`, and
+    # graph-engine §2 projects nothing in by default, so the fan-out would see
+    # no items; neither state declares the items field either. The builder
+    # projects and declares it under its own name, pending a spec ruling. It
+    # also types `picks` and `result` as plain lists, since the declared
+    # `list<dict>` and `dict` cannot hold the strings the instances collect.
     wrap_state_cls = build_state_cls(
         "Wrapped039C3",
         {
             "picks": {"type": "list", "reducer": "append", "default": []},
-            "products": {"type": "list<dict>", "default": []},
+            items_field: {"type": "list<dict>", "default": []},
             "oa_fan_out_item": {"type": "dict", "default": {}},
         },
     )
     wrap_builder: GraphBuilder[Any] = GraphBuilder(wrap_state_cls)
-    wrap_builder.set_entry("pick")
+    wrap_builder.set_entry(fan_out_name)
     wrap_builder.add_fan_out_node(
-        "pick",
+        fan_out_name,
         subgraph=per_product,
-        items_field="products",
+        items_field=items_field,
         item_field="oa_fan_out_item",
         collect_field="picked",
         target_field="picks",
-        instance_middleware=[_make_augment_instance_middleware({"note": "id"}, "oa_fan_out_item")],
+        instance_middleware=[
+            _make_augment_instance_middleware(
+                cast("dict[str, str]", fan_out["augment_metadata_from_field"]), "oa_fan_out_item"
+            )
+        ],
     )
-    wrap_builder.add_edge("pick", END)
+    wrap_builder.add_edge(fan_out_name, END)
     wrapped_fan_out = wrap_builder.compile()
 
     outer_state_cls = build_state_cls(
         "Outer039C3",
-        {"result": {"type": "list", "default": []}, "products": {"type": "list<dict>", "default": []}},
+        {"result": {"type": "list", "default": []}, items_field: {"type": "list<dict>", "default": []}},
     )
+    inputs = cast("dict[str, str]", call["inputs"]) if "inputs" in call else {items_field: items_field}
     outer_builder: GraphBuilder[Any] = GraphBuilder(outer_state_cls)
-    outer_builder.set_entry("wrap")
+    outer_builder.set_entry(wrap_name)
     outer_builder.add_subgraph_node(
-        "wrap",
+        wrap_name,
         wrapped_fan_out,
-        ExplicitMapping(inputs={"products": "products"}, outputs={"result": "picks"}),
+        ExplicitMapping(inputs=inputs, outputs=cast("dict[str, str]", call["outputs"])),
     )
-    outer_builder.add_edge("wrap", END)
+    outer_builder.add_edge(wrap_name, END)
     graph = outer_builder.compile()
     initial = cast("dict[str, Any]", case.get("initial_state") or {})
     return graph, (lambda: outer_state_cls(**initial))
@@ -1457,6 +1481,11 @@ async def _run_resume_case(
     """
     from openarmature.graph.errors import RuntimeGraphError  # noqa: PLC0415
 
+    # conformance-adapter §5.6: `from_first_run: true` resumes the first run's
+    # invocation id, which is the only resume Step 2 performs.
+    from_first_run = cast("dict[str, Any]", case["resume"]).get("from_first_run")
+    assert from_first_run is True, f"resume.from_first_run: only `true` is supported; got {from_first_run!r}"
+
     # ---- Step 1: first invoke catches expected error
     first_run_expected_error = cast("dict[str, Any]", case.get("first_run_expected_error") or {})
     expected_category = cast("str", first_run_expected_error.get("category", "node_exception"))

From 39307ec78a7655f822783b3b420e1b143e178fdf Mon Sep 17 00:00:00 2001
From: chris-colinsky <chris@lunarcommand.xyz>
Date: Sat, 17 Oct 2026 07:12:47 -0700
Subject: [PATCH 2/5] Treat span-tree children lists as complete

The shared OTel span-tree matcher only checked that each declared
child existed, so a span the expected tree did not list passed
unnoticed.

A declared list is now the complete set at every level. `children: []`
asserts a leaf, a span the tree does not declare fails with its parent
named, and a node without `children` leaves them unchecked. The
Langfuse runner's observation matcher already treated a non-empty list
this way but skipped an empty one, so it now enforces `children: []`
as well.

004 now matches its tree from every root its exporter saw, so a span
outside the invocation's tree fails too. That covers its hand-written
check that the unreachable nodes emit no span, which is removed.

The observability suite's harness note says nothing about children,
so this reading is pending a spec ruling. Every fixture that runs today
passes under it.
---
 tests/conformance/test_observability.py | 31 +++++++++++++++++--------
 1 file changed, 21 insertions(+), 10 deletions(-)

diff --git a/tests/conformance/test_observability.py b/tests/conformance/test_observability.py
index ee509c6..2cd8663 100644
--- a/tests/conformance/test_observability.py
+++ b/tests/conformance/test_observability.py
@@ -1274,7 +1274,9 @@ async def _run_fixture_004(spec: Mapping[str, Any]) -> None:
     expected_tree = cast("list[dict[str, Any]]", expected["span_tree"])
     inv_root = next((s for s in spans if s.name == "openarmature.invocation" and s.parent is None), None)
     assert inv_root is not None, f"invocation root span missing; got {[s.name for s in spans]}"
-    _assert_span_tree_matches(spans, [inv_root], expected_tree)
+    # Matched from every root the exporter saw, so a span outside the
+    # invocation's tree (an unreachable node's, say) fails as well.
+    _assert_span_tree_matches(spans, _exporter_roots(spans), expected_tree)
     _assert_error_span_extras(spans, expected_tree)
 
     # §4.2: edge logic folds into the preceding node's span. Checked as a closed
@@ -1286,11 +1288,6 @@ async def _run_fixture_004(spec: Mapping[str, Any]) -> None:
         extra = sorted(s.name for s in spans if s.name not in allowed)
         assert not extra, f"spans that are neither the invocation nor a node (§4.2): {extra}"
 
-    # Unreachable nodes never fire spans (they were unreached).
-    span_names = {s.name for s in spans}
-    for unreachable in ("unreachable_a", "unreachable_b"):
-        assert unreachable not in span_names, f"{unreachable!r} MUST not produce a span — never reached"
-
 
 # ---------------------------------------------------------------------------
 # Fixture 006 — fan-out instance attribution
@@ -2707,12 +2704,18 @@ def _attr_equal(actual: Any, expected: Any) -> bool:
 
 
 def _assert_span_tree_matches(
-    all_spans: Sequence[Any], actual_roots: Sequence[Any], expected_nodes: Sequence[Mapping[str, Any]]
+    all_spans: Sequence[Any],
+    actual_roots: Sequence[Any],
+    expected_nodes: Sequence[Mapping[str, Any]],
+    *,
+    under: str = "the top level",
 ) -> None:
     """Recursive structural match: for every expected node, find a
     matching actual span by name + attribute-subset; recurse on its
     children.  Children are matched as a SET (order independent —
-    parallel-branches dispatch order isn't span-emission order)."""
+    parallel-branches dispatch order isn't span-emission order).
+    A declared list, at the top level or as ``children``, must account
+    for every span; a node without ``children`` leaves them unchecked."""
     actual_by_name: dict[str, list[Any]] = {}
     for root in actual_roots:
         actual_by_name.setdefault(root.name, []).append(root)
@@ -2765,7 +2768,14 @@ def _matches(span: Any, eattrs: dict[str, Any] = expected_attrs) -> bool:
             actual_children = [
                 s for s in all_spans if s.parent is not None and s.parent.span_id == matched_span_id
             ]
-            _assert_span_tree_matches(all_spans, actual_children, expected_children)
+            _assert_span_tree_matches(
+                all_spans, actual_children, expected_children, under=repr(expected_name)
+            )
+    # A declared list is the complete set: `children: []` asserts a leaf, and a
+    # span the tree does not list fails. The suite's harness note is silent on
+    # children, so this reading is pending a spec ruling.
+    undeclared = sorted(s.name for group in actual_by_name.values() for s in group)
+    assert not undeclared, f"spans under {under} that the expected tree does not declare: {undeclared}"
 
 
 async def _run_fixture_132(spec: Mapping[str, Any]) -> None:
@@ -5716,7 +5726,8 @@ def _assert_langfuse_observation_tree(
         if "metadata_truncation" in exp:
             _assert_metadata_truncation(match, cast("dict[str, Any]", exp["metadata_truncation"]))
         children = cast("list[dict[str, Any]] | None", exp.get("children"))
-        if children:
+        # `children: []` asserts a leaf, so an empty list recurses too.
+        if children is not None:
             _assert_langfuse_observation_tree(trace, children, parent_id=match.id)
     # Every observation under this parent must be accounted for. Leaving
     # leftovers unasserted lets an unexpected observation ride silently, the

From 61fc7d50384355940779b580f007519176996194 Mon Sep 17 00:00:00 2001
From: chris-colinsky <chris@lunarcommand.xyz>
Date: Sat, 17 Oct 2026 09:05:07 -0700
Subject: [PATCH 3/5] Tighten the 009 UUID and span-tree checks

_is_uuidv4 used re.match, and `$` also matches before a final
newline, so a UUID with a trailing newline passed the `<uuid>`
placeholder and 009's UUID invariants. It now uses fullmatch.

009 matched its span_tree from the invocation root only, so a stray
root span carrying the same correlation id passed both the tree and
the uniformity invariant in the single-invocation cases. It now
matches from every root the exporter saw, as 004 and 005 do.
---
 tests/conformance/test_observability.py | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/tests/conformance/test_observability.py b/tests/conformance/test_observability.py
index 2cd8663..03cdb2d 100644
--- a/tests/conformance/test_observability.py
+++ b/tests/conformance/test_observability.py
@@ -617,7 +617,7 @@ def _reset_otel_global_tracer_provider(restore_to: object) -> None:
 
 
 def _is_uuidv4(value: Any) -> bool:
-    return isinstance(value, str) and _UUIDV4_RE.match(value) is not None
+    return isinstance(value, str) and _UUIDV4_RE.fullmatch(value) is not None
 
 
 def _fixture_paths() -> list[Path]:
@@ -2218,7 +2218,8 @@ async def _run_fixture_009_case(case: Mapping[str, Any]) -> None:
     if (tree := expected.get("span_tree")) is not None:
         inv_root = next((s for s in spans if s.name == "openarmature.invocation" and s.parent is None), None)
         assert inv_root is not None, f"invocation root span missing; got {[s.name for s in spans]}"
-        _assert_span_tree_matches(spans, [inv_root], cast("list[dict[str, Any]]", tree))
+        # Matched from every root the exporter saw, so a stray root fails as well.
+        _assert_span_tree_matches(spans, _exporter_roots(spans), cast("list[dict[str, Any]]", tree))
 
     # Each invocation is its own trace.
     by_trace: dict[int, list[Any]] = {}

From 3d449af97c108880dd59f6d9d7ff7e13ca6c3b14 Mon Sep 17 00:00:00 2001
From: chris-colinsky <chris@lunarcommand.xyz>
Date: Sat, 17 Oct 2026 09:06:29 -0700
Subject: [PATCH 4/5] Require canonical UUIDs in the OTel runner

The OTel runner's UUIDv4 pattern was case-insensitive, so an
uppercase id passed the `<uuid>` placeholder and 009's invariants.
RFC 9562 writes UUIDs in lowercase, and the checkpoint runner's own
check already requires it. The pattern now does too.

The span test for pipeline-utilities 031 still called the pattern's
match() directly, so a trailing newline passed there. It now goes
through _is_uuidv4.
---
 tests/conformance/test_observability.py | 8 +++-----
 1 file changed, 3 insertions(+), 5 deletions(-)

diff --git a/tests/conformance/test_observability.py b/tests/conformance/test_observability.py
index 03cdb2d..665e8bb 100644
--- a/tests/conformance/test_observability.py
+++ b/tests/conformance/test_observability.py
@@ -610,10 +610,8 @@ def _reset_otel_global_tracer_provider(restore_to: object) -> None:
 
 
 # UUIDv4 canonical form: xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx (where y in {8,9,a,b}).
-_UUIDV4_RE = re.compile(
-    r"^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$",
-    re.IGNORECASE,
-)
+# RFC 9562 section 4: lowercase on output.
+_UUIDV4_RE = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$")
 
 
 def _is_uuidv4(value: Any) -> bool:
@@ -3901,7 +3899,7 @@ async def delete(self, invocation_id: str) -> None:
         )
     else:
         # Auto-generated → UUIDv4 form.
-        assert _UUIDV4_RE.match(first_correlation_id), (
+        assert _is_uuidv4(first_correlation_id), (
             f"auto-generated correlation_id MUST be UUIDv4; got {first_correlation_id!r}"
         )
 

From 568a65ab5e9a26f657b61a87f6ed3ed3cceed319 Mon Sep 17 00:00:00 2001
From: chris-colinsky <chris@lunarcommand.xyz>
Date: Sat, 17 Oct 2026 09:10:17 -0700
Subject: [PATCH 5/5] Match every span tree from all exporter roots

Eight more OTel drivers matched their span_tree from the invocation
root only. The tree matcher checks the top level against the roots
it is given, so a stray root span went unchecked in 038, 110, 124,
126-129, 131-133, 146, 147, 149, 152 and 153. They now pass every
root the exporter saw, as 004, 005 and 009 do, and the matcher's
docstring says why.
---
 tests/conformance/test_observability.py | 20 +++++++++++---------
 1 file changed, 11 insertions(+), 9 deletions(-)

diff --git a/tests/conformance/test_observability.py b/tests/conformance/test_observability.py
index 665e8bb..b6bc1ca 100644
--- a/tests/conformance/test_observability.py
+++ b/tests/conformance/test_observability.py
@@ -2636,7 +2636,7 @@ async def _ask_body(
         (s for s in spans if s.name == "openarmature.invocation" and cast("Any", s.parent) is None), None
     )
     assert inv_root is not None, f"invocation root span missing; got {[s.name for s in spans]}"
-    _assert_span_tree_matches(spans, [inv_root], expected_tree)
+    _assert_span_tree_matches(spans, _exporter_roots(spans), expected_tree)
 
     # ---- Invariants
     invariants = cast("dict[str, Any]", case["expected"].get("invariants") or {})
@@ -2714,7 +2714,9 @@ def _assert_span_tree_matches(
     children.  Children are matched as a SET (order independent —
     parallel-branches dispatch order isn't span-emission order).
     A declared list, at the top level or as ``children``, must account
-    for every span; a node without ``children`` leaves them unchecked."""
+    for every span; a node without ``children`` leaves them unchecked.
+    The top level sees only ``actual_roots``, so pass every root the
+    exporter saw (``_exporter_roots``) or a stray root goes unchecked."""
     actual_by_name: dict[str, list[Any]] = {}
     for root in actual_roots:
         actual_by_name.setdefault(root.name, []).append(root)
@@ -2955,7 +2957,7 @@ async def _ask_body(_s: Any) -> dict[str, str]:
         (s for s in spans if s.name == "openarmature.invocation" and cast("Any", s.parent) is None), None
     )
     assert inv_root is not None, f"invocation root span missing; got {[s.name for s in spans]}"
-    _assert_span_tree_matches(spans, [inv_root], expected_tree)
+    _assert_span_tree_matches(spans, _exporter_roots(spans), expected_tree)
 
     # ---- Invariants (fail loud on any declared-but-unhandled name).
     invariants = cast("dict[str, Any]", case["expected"].get("invariants") or {})
@@ -3296,7 +3298,7 @@ async def _wrapper_mw(state: Any, next_call: Any) -> Mapping[str, Any]:
         (s for s in spans if s.name == "openarmature.invocation" and cast("Any", s.parent) is None), None
     )
     assert inv_root is not None, f"invocation root span missing; got {[s.name for s in spans]}"
-    _assert_span_tree_matches(spans, [inv_root], expected_tree)
+    _assert_span_tree_matches(spans, _exporter_roots(spans), expected_tree)
 
     # ---- Invariants (fail loud on any declared-but-unhandled name).
     invariants = cast("dict[str, Any]", case["expected"].get("invariants") or {})
@@ -3397,7 +3399,7 @@ async def _run_fixture_110_case(case: Mapping[str, Any]) -> None:
     expected_tree = cast("list[dict[str, Any]]", case["expected"]["span_tree"])
     inv_root = next((s for s in spans if s.name == "openarmature.invocation" and s.parent is None), None)
     assert inv_root is not None, f"invocation root span missing; got {[s.name for s in spans]}"
-    _assert_span_tree_matches(spans, [inv_root], expected_tree)
+    _assert_span_tree_matches(spans, _exporter_roots(spans), expected_tree)
 
     # ---- when-skipped branches emit NO span. The span_tree match is
     # subset-based, so assert the skip explicitly: any declared branch absent
@@ -5101,7 +5103,7 @@ async def _run_token_budget_case(case: Mapping[str, Any]) -> None:
             None,
         )
         assert inv_root is not None, f"invocation root span missing; got {[s.name for s in spans]}"
-        _assert_span_tree_matches(spans, [inv_root], expected_tree)
+        _assert_span_tree_matches(spans, _exporter_roots(spans), expected_tree)
         # _assert_span_tree_matches ignores attributes_absent; assert the
         # declared-only / no-budget absence claims explicitly.
         _assert_error_span_extras(spans, expected_tree)
@@ -5325,7 +5327,7 @@ async def _run_structured_output_error_span_case(case: Mapping[str, Any]) -> Non
         None,
     )
     assert inv_root is not None, f"invocation root span missing; got {[s.name for s in spans]}"
-    _assert_span_tree_matches(spans, [inv_root], expected_tree)
+    _assert_span_tree_matches(spans, _exporter_roots(spans), expected_tree)
     # _assert_span_tree_matches ignores status_description / attributes_absent /
     # exception_recorded; assert those (124's payload redaction + §4.2 exception
     # event) explicitly so the fixture's central claims are actually verified.
@@ -7096,7 +7098,7 @@ async def _run_typed_event_with_span_case(case: Mapping[str, Any]) -> None:
             None,
         )
         assert inv_root is not None, f"invocation root span missing; got {[s.name for s in spans]}"
-        _assert_span_tree_matches(spans, [inv_root], expected_tree)
+        _assert_span_tree_matches(spans, _exporter_roots(spans), expected_tree)
         _assert_error_span_extras(spans, expected_tree)
 
     _assert_invariants_recognized(case, _MALFORMED_COUNTER_INVARIANTS, "malformed-counter")
@@ -7625,7 +7627,7 @@ async def _mw(state: Any, next_call: Any) -> Any:
         # while neither fixture declares a key one of them solely reads, so
         # assert that rather than leaving it to hold by luck.
         _assert_no_skipped_span_tree_keys(expected_tree)
-        _assert_span_tree_matches(spans, [inv_root], expected_tree)
+        _assert_span_tree_matches(spans, _exporter_roots(spans), expected_tree)
 
     _assert_invariants_recognized(case, _ORPHAN_FALLBACK_INVARIANTS, "orphan-fallback")
     _assert_orphan_fallback_invariants(case, spans, subgraph_specs, mock_bodies)