Wire fixture 084 and extend conformance harness

Activate observability fixture 084 (Langfuse session/user promotion):
cases 2/3/4 (not session-bound, userId present additively, userId
absent) run now; the session-bound cases 1/5 defer per-case until the
sessions capability (0020) supplies a session_id source.

Extend the fixture-parsing models for 084's shapes: ObservabilityExpected
gains langfuse_trace / langfuse_traces (and the matching discriminator
keys), and CaseSpec.invocations widens to int | list to carry case 5's
multi-invocation specs alongside the existing run-count usage.

Author: chris-colinsky

tests/conformance/harness/expectations.py

@@ -197,6 +197,11 @@ class ObservabilityExpected(_ForbidExtras):
     determinism_check: dict[str, Any] | None = None
     # Multi-invocation fixtures (009 cross-cutting, 011 determinism).
     invocation_count: int | None = None
+    # Langfuse Trace-level expectations (proposal 0064, fixture 084): the
+    # single-trace shape (sessionId / userId / metadata) and the
+    # multi-invocation grouping case's per-trace list.
+    langfuse_trace: dict[str, Any] | None = None
+    langfuse_traces: list[dict[str, Any]] | None = None
 
 
 # ---------------------------------------------------------------------------
@@ -240,6 +245,9 @@ class ObservabilityExpected(_ForbidExtras):
         "no_llm_provider_span",
         "determinism_check",
         "invocation_count",
+        # proposal 0064 (fixture 084) Langfuse Trace-level expectations
+        "langfuse_trace",
+        "langfuse_traces",
     }
 )
 

tests/conformance/harness/fixtures.py

@@ -134,7 +134,9 @@ class CaseSpec(BaseModel):
     caller_correlation_id: str | None = None
     # observability — mock LLM responses + per-case run config.
     mock_llm: list[MockResponse] | None = None
-    invocations: int | None = None
+    # ``int`` = a run-count (fixtures 009 / 011); ``list`` = per-invocation
+    # specs for a multi-invocation fixture (proposal 0064 fixture 084 case 5).
+    invocations: int | list[dict[str, Any]] | None = None
 
 
 # ---------------------------------------------------------------------------

tests/conformance/test_observability.py

@@ -153,6 +153,11 @@ def _reset_otel_global_tracer_provider(restore_to: object) -> None:
         # (v0.61.0) resolves the detached-invocation-span shape case 2
         # presupposed — the whole fixture was unwired pending that.
         "058-implementation-attribution-otel",
+        # v0.62.0 — proposal 0064 (Langfuse trace.sessionId / trace.userId
+        # population). Cases 2/3/4 (not session-bound + userId promotion)
+        # run; session-bound cases 1/5 defer until the sessions capability
+        # (0020) supplies openarmature.session_id.
+        "084-langfuse-session-user-promotion",
     }
 )
 
@@ -246,6 +251,8 @@ async def test_observability_fixture(fixture_path: Path) -> None:
         await _run_fixture_056(spec)
     elif fixture_id == "058-implementation-attribution-otel":
         await _run_fixture_058(spec)
+    elif fixture_id == "084-langfuse-session-user-promotion":
+        await _run_fixture_084(spec)
     elif fixture_id in {
         "012-otel-llm-payload-default-off",
         "013-otel-llm-payload-enabled",
@@ -2175,6 +2182,74 @@ async def delete(self, invocation_id: str) -> None:
     )
 
 
+# ---------------------------------------------------------------------------
+# Fixture 084 — Langfuse session/user promotion (proposal 0064)
+# ---------------------------------------------------------------------------
+
+
+async def _run_fixture_084(spec: Mapping[str, Any]) -> None:
+    from openarmature.observability.langfuse import (  # noqa: PLC0415
+        InMemoryLangfuseClient,
+        LangfuseObserver,
+    )
+
+    # Proposal 0064 §8.4.1. Cases 1 + 5 are session-bound: they supply
+    # session_id at invoke(), which needs the sessions capability
+    # (proposal 0020, §5.6) to surface openarmature.session_id. That is
+    # unimplemented in python until v0.19.0, so trace.sessionId has no
+    # source and these cases defer (per-case continue). Cases 2/3/4 (not
+    # session-bound + the userId promotion) run now.
+    _deferred_cases = {
+        "session_bound_sets_trace_session_id",
+        "multi_invocation_shared_session_groups",
+    }
+    cases = cast("list[dict[str, Any]]", spec["cases"])
+    for case in cases:
+        case_name = cast("str", case["name"])
+        if case_name in _deferred_cases:
+            continue
+        try:
+            client = InMemoryLangfuseClient()
+            observer = LangfuseObserver(client=client)
+            trace: list[str] = []
+            built = build_graph(case, trace=trace)
+            compiled = built.builder.compile()
+            compiled.attach_observer(observer)
+            initial_state = built.initial_state(case.get("initial_state", {}))
+            caller_metadata = cast("dict[str, Any] | None", case.get("caller_metadata"))
+            if caller_metadata is not None:
+                await compiled.invoke(initial_state, metadata=caller_metadata)
+            else:
+                await compiled.invoke(initial_state)
+            await compiled.drain()
+            observer.shutdown()
+
+            assert len(client.traces) == 1, f"expected 1 trace, got {len(client.traces)}"
+            lf_trace = next(iter(client.traces.values()))
+            expected = cast("dict[str, Any]", case["expected"]["langfuse_trace"])
+            # trace.sessionId is unset for the runnable cases (no session
+            # source until 0020).
+            assert lf_trace.session_id == expected.get("sessionId"), (
+                f"sessionId: got {lf_trace.session_id!r}, expected {expected.get('sessionId')!r}"
+            )
+            # trace.userId: promoted from the userId caller key (case 3),
+            # unset otherwise (cases 2/4).
+            assert lf_trace.user_id == expected.get("userId"), (
+                f"userId: got {lf_trace.user_id!r}, expected {expected.get('userId')!r}"
+            )
+            # Additive promotion + unaffected metadata: every concrete
+            # (non-placeholder) expected metadata key also lands top-level.
+            expected_md = cast("dict[str, Any]", expected.get("metadata") or {})
+            for key, val in expected_md.items():
+                if isinstance(val, str) and val.startswith("<") and val.endswith(">"):
+                    continue
+                assert lf_trace.metadata.get(key) == val, (
+                    f"metadata.{key}: got {lf_trace.metadata.get(key)!r}, expected {val!r}"
+                )
+        except AssertionError as e:
+            raise AssertionError(f"case {case_name!r}: {e}") from e
+
+
 # ---------------------------------------------------------------------------
 # Fixture 010 — log correlation
 #