diff --git a/tests/conformance/test_harness_fidelity.py b/tests/conformance/test_harness_fidelity.py
index 3393920..b599de8 100644
--- a/tests/conformance/test_harness_fidelity.py
+++ b/tests/conformance/test_harness_fidelity.py
@@ -10,7 +10,7 @@
 
 import ast
 import inspect
-from typing import Any
+from typing import Any, cast
 
 import pytest
 
@@ -96,31 +96,126 @@ def test_every_dispatched_driver_is_registered_or_explicitly_unguarded() -> None
 
 def _invariant_names_read_by(func: object, param: str) -> set[str]:
     """String literals `func` looks up on the mapping named `param`."""
+    return _invariant_names_read_by_node(
+        ast.parse(inspect.getsource(func)),  # pyright: ignore[reportArgumentType]
+        param,
+    )
+
+
+def _invariant_helpers_called_by(func: object, module: object) -> list[object]:
+    """Module-level callees of `func` that take an `expected_invariants` param."""
+    # One level, deliberately. Reading a guard's own body misses names it
+    # delegates to a helper, which is not hypothetical: moving 148's two
+    # invariants into `_assert_generation_usage_omission` made the check below
+    # blind to them the moment it was written. Deeper recursion is the
+    # transitive-derivation problem tracked separately; one level covers the
+    # delegate-to-a-sibling shape that actually occurs here.
+    #
+    # Selected by SIGNATURE, not by an `_assert_` name prefix. A prefix filter
+    # silently excludes a future guard spelled `_check_...`, and excluding it
+    # from the walk is the fail-open direction this whole check exists to close.
     tree = ast.parse(inspect.getsource(func))  # pyright: ignore[reportArgumentType]
-    names: set[str] = set()
+    found: list[object] = []
+    for node in ast.walk(tree):
+        if not (isinstance(node, ast.Call) and isinstance(node.func, ast.Name)):
+            continue
+        helper = getattr(module, node.func.id, None)
+        if not callable(helper) or helper in found:
+            continue
+        try:
+            params = inspect.signature(helper).parameters
+        except (TypeError, ValueError):  # pragma: no cover - builtins
+            continue
+        if "expected_invariants" in params:
+            found.append(helper)
+    return found
+
+
+def _invariant_names_evaluated_by(func: object, param: str) -> set[str]:
+    """Names from `_invariant_names_read_by` whose `if` body actually does work."""
+    # A name is MENTIONED when `if invariants.get("x"):` appears; it is
+    # EVALUATED when that branch asserts something. `_assert_trace` contains a
+    # deliberate mention-only arm (`trace_id_equals_invocation_id`, whose body is
+    # a comment and `pass`), so counting mentions as evaluation lets a guard be
+    # gutted to `if x: pass` while every check here stays green.
+    tree = ast.parse(inspect.getsource(func))  # pyright: ignore[reportArgumentType]
+    # Guards spell the lookup two ways: inline in the test (`if inv.get("x"):`)
+    # and bound to a local first (`x = inv.get("x")` ... `if x:`), which is what
+    # a guard reading several names for one early-return does. Resolve the
+    # binding so the check does not quietly mandate one of the two idioms.
+    bound: dict[str, str] = {}
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Assign) and len(node.targets) == 1 and isinstance(node.targets[0], ast.Name):
+            names = _invariant_names_read_by_node(node.value, param)
+            if len(names) == 1:
+                bound[node.targets[0].id] = next(iter(names))
+
+    evaluated: set[str] = set()
     for node in ast.walk(tree):
+        if not isinstance(node, ast.If):
+            continue
+        body = [s for s in node.body if not _is_inert(s)]
+        if not body:
+            continue
+        evaluated |= _invariant_names_read_by_node(node.test, param)
+        evaluated |= {bound[n.id] for n in ast.walk(node.test) if isinstance(n, ast.Name) and n.id in bound}
+    return evaluated
+
+
+def _is_inert(stmt: ast.stmt) -> bool:
+    """A statement that asserts nothing about the invariant."""
+    # A BARE `return` counts as inert, which is not a nicety: a guard reading
+    # several names typically opens with `if not a and not b: return`, and that
+    # arm mentions every name while checking none of them. Counting it as
+    # evaluation let the gutted-to-`pass` mutant survive this whole check.
+    if isinstance(stmt, ast.Pass):
+        return True
+    if isinstance(stmt, ast.Return) and stmt.value is None:
+        return True
+    return (
+        isinstance(stmt, ast.Expr)
+        and isinstance(stmt.value, ast.Constant)
+        and isinstance(stmt.value.value, str)
+    )
+
+
+def _invariant_names_read_by_node(node: ast.AST, param: str) -> set[str]:
+    """The `_invariant_names_read_by` extraction, over an arbitrary subtree."""
+    names: set[str] = set()
+    for sub in ast.walk(node):
         if (
-            isinstance(node, ast.Call)
-            and isinstance(node.func, ast.Attribute)
-            and node.func.attr == "get"
-            and isinstance(node.func.value, ast.Name)
-            and node.func.value.id == param
-            and node.args
-            and isinstance(node.args[0], ast.Constant)
-            and isinstance(node.args[0].value, str)
+            isinstance(sub, ast.Call)
+            and isinstance(sub.func, ast.Attribute)
+            and sub.func.attr == "get"
+            and isinstance(sub.func.value, ast.Name)
+            and sub.func.value.id == param
+            and sub.args
+            and isinstance(sub.args[0], ast.Constant)
+            and isinstance(sub.args[0].value, str)
         ):
-            names.add(node.args[0].value)
+            names.add(sub.args[0].value)
         elif (
-            isinstance(node, ast.Subscript)
-            and isinstance(node.value, ast.Name)
-            and node.value.id == param
-            and isinstance(node.slice, ast.Constant)
-            and isinstance(node.slice.value, str)
+            isinstance(sub, ast.Subscript)
+            and isinstance(sub.value, ast.Name)
+            and sub.value.id == param
+            and isinstance(sub.slice, ast.Constant)
+            and isinstance(sub.slice.value, str)
         ):
-            names.add(node.slice.value)
+            names.add(sub.slice.value)
     return names
 
 
+# Per-trace invariant names whose arm is deliberately a no-op, with the reason.
+# Listed so the stale check can tell "documented as inert" from "silently gutted".
+_DOCUMENTARY_PER_TRACE_INVARIANTS = {
+    # §8.4.1 says trace.id == invocation_id, and there is no accessor for the
+    # invocation_id from outside the observer, so the claim degenerates to
+    # "trace.id matches the UUIDv4 pattern" -- already asserted via the
+    # `<uuid>` placeholder on the id itself.
+    "trace_id_equals_invocation_id",
+}
+
+
 def _span_dependent_invariants_in(func: object) -> set[str]:
     """Invariant names whose `if invariants.get(...)` body reads the span set."""
     tree = ast.parse(inspect.getsource(func))  # pyright: ignore[reportArgumentType]
@@ -189,15 +284,108 @@ def test_a_zero_span_run_cannot_satisfy_a_span_absence_claim(invariant: str, dri
         drive({"expected": {"invariants": {invariant: True}}})
 
 
-def test_per_trace_invariants_covers_everything_assert_trace_reads() -> None:
+def test_per_trace_invariants_matches_what_the_per_trace_guards_evaluate() -> None:
     # `_assert_multi_traces` filters the fixture's invariants down to
     # `_PER_TRACE_INVARIANTS` before delegating. A name `_assert_trace` checks
     # but that is absent from the set is therefore dropped on that path, and the
     # fixture declaring it still passes.
-    read = _invariant_names_read_by(langfuse_runner._assert_trace, "expected_invariants")  # noqa: SLF001
-    assert read, "parsed no invariant lookups out of _assert_trace, so this compared nothing"
+    entry = langfuse_runner._assert_trace  # noqa: SLF001
+    helpers = _invariant_helpers_called_by(entry, langfuse_runner)
+    # Non-vacuity on the callee walk: `_assert_trace` delegates today, so
+    # finding none means the walk broke and the check silently narrowed back to
+    # the entry point's own body.
+    assert helpers, "found no invariant helpers called by _assert_trace; the callee walk is broken"
+    read: set[str] = set()
+    evaluated: set[str] = set()
+    for func in [entry, *helpers]:
+        read |= _invariant_names_read_by(func, "expected_invariants")
+        evaluated |= _invariant_names_evaluated_by(func, "expected_invariants")
+    assert read, "parsed no invariant lookups out of the per-trace guards, so this compared nothing"
+    assert evaluated, "parsed no EVALUATED invariant arms; the body-triviality filter is broken"
     missing = sorted(read - set(langfuse_runner._PER_TRACE_INVARIANTS))  # noqa: SLF001
     assert not missing, (
-        f"`_assert_trace` checks {missing}, but `_PER_TRACE_INVARIANTS` omits them, so the "
+        f"the per-trace guards check {missing}, but `_PER_TRACE_INVARIANTS` omits them, so the "
         f"multi-trace runner discards those claims silently."
     )
+    # The other direction, and on EVALUATED rather than merely mentioned: a set
+    # entry whose arm is `if name: pass` reads as covered while asserting
+    # nothing, so gutting a guard to its `if` would otherwise stay green.
+    stale = sorted(
+        set(langfuse_runner._PER_TRACE_INVARIANTS)  # noqa: SLF001
+        - evaluated
+        - _DOCUMENTARY_PER_TRACE_INVARIANTS
+    )
+    assert not stale, (
+        f"`_PER_TRACE_INVARIANTS` names invariants no per-trace guard evaluates: {stale}. Either "
+        f"the guard was gutted, or the arm is deliberately inert and belongs in "
+        f"`_DOCUMENTARY_PER_TRACE_INVARIANTS` with its reason."
+    )
+
+
+def test_langfuse_harness_fixtures_are_all_driven_by_the_langfuse_runner() -> None:
+    # Activating a Langfuse-mapping fixture takes two independent edits in two
+    # files: adding it to `_LANGFUSE_HARNESS_FIXTURES` here, which makes the OTel
+    # runner skip it, and to `_LANGFUSE_FIXTURES` there, which makes the Langfuse
+    # runner parametrize it. Do only the first and the fixture runs NOWHERE: the
+    # OTel runner skips it saying it is tested by the sibling, the sibling never
+    # collects it, and the coverage guard counts it as accounted because the
+    # skip set is unioned into `accounted`. One green suite, zero assertions,
+    # and a skip message actively claiming coverage.
+    skipped_here = set(otel_runner._LANGFUSE_HARNESS_FIXTURES)  # noqa: SLF001
+    driven_there = set(langfuse_runner._LANGFUSE_FIXTURES)  # noqa: SLF001
+    assert skipped_here, "the skip set is empty, so the check below compared nothing"
+    undriven = sorted(skipped_here - driven_there)
+    assert not undriven, (
+        f"{undriven} are skipped by the OTel runner as 'fixture-tested by the Langfuse "
+        f"harness', but the Langfuse runner does not collect them, so they run nowhere while "
+        f"the coverage guard counts them as accounted. Add them to `_LANGFUSE_FIXTURES`."
+    )
+
+
+# Invariant names the Langfuse runner IMPLEMENTS, mapped to the fixture whose
+# claim rests on them. Most invariants in this corpus are documentary -- they
+# restate what a concrete directive already pins -- so a blanket "every declared
+# name must be read" guard would fail the ~50 working exactly as intended.
+# These are the opposite: the fixture's `langfuse_trace` block CANNOT express
+# the claim, so the invariant is the only thing asserting it.
+_LOAD_BEARING_LANGFUSE_INVARIANTS = {
+    # 148: `usage` is compared by iterating the EXPECTED keys, so an omitted
+    # `input` key -- the whole claim -- is invisible to the directive.
+    "generation_usage_input_omitted_when_prompt_tokens_null": (
+        "148-langfuse-generation-usage-omits-input-on-null-counter"
+    ),
+    "generation_usage_output_and_total_present_when_sound": (
+        "148-langfuse-generation-usage-omits-input-on-null-counter"
+    ),
+    # 156 declares no `level` at all, so this is its only expression. 155 pins
+    # `level: ERROR` concretely, which is why only the 156 direction is here.
+    "no_warning_level_under_budget": "156-langfuse-token-budget-under-budget-flag-false",
+}
+
+
+def test_load_bearing_langfuse_invariants_are_still_declared() -> None:
+    # A guard keyed on an invariant name goes dark the moment the fixture stops
+    # declaring that name -- a spec-side rename is the routine way that happens,
+    # and a pin bump brings those. Nothing else catches it: the guard early-
+    # returns, the `langfuse_trace` block passes on its own, the
+    # assert-something check is satisfied by that block's presence, and the
+    # unknown-directive check never descends into `expected.invariants`. The
+    # fixture reports green while asserting nothing it was activated for.
+    import yaml  # noqa: PLC0415
+
+    declared: dict[str, set[str]] = {}
+    for path in langfuse_runner._fixture_paths():  # noqa: SLF001
+        spec = cast("dict[str, Any]", yaml.safe_load(path.read_text()))
+        cases = cast("list[dict[str, Any]]", spec.get("cases") or [spec])
+        for case in cases:
+            expected = cast("dict[str, Any]", case.get("expected") or {})
+            for name in cast("dict[str, Any]", expected.get("invariants") or {}):
+                declared.setdefault(name, set()).add(path.stem)
+    assert declared, "parsed no invariants out of the corpus, so this compared nothing"
+
+    for name, fixture in sorted(_LOAD_BEARING_LANGFUSE_INVARIANTS.items()):
+        assert fixture in declared.get(name, set()), (
+            f"{fixture} no longer declares {name!r}, but the Langfuse runner still implements "
+            f"a guard keyed on that name. The guard is now dead and the fixture's claim rests "
+            f"on nothing. Re-point the guard at the fixture's new spelling."
+        )
diff --git a/tests/conformance/test_observability.py b/tests/conformance/test_observability.py
index 7a7add4..22b8fdb 100644
--- a/tests/conformance/test_observability.py
+++ b/tests/conformance/test_observability.py
@@ -257,6 +257,12 @@ def _reset_otel_global_tracer_provider(restore_to: object) -> None:
         "128-token-budget-under-budget-no-warning",
         "129-token-budget-absent-unchanged",
         "131-token-budget-on-structured-output-failure",
+        # 149 (proposal 0101): a MALFORMED wire counter ("abc" / -5 / true) is
+        # nulled by the real complete() mapping BEFORE the typed event and the
+        # span render. The event half is the fixture's stated discriminator --
+        # an impl sourcing the event from `raw` surfaces the wire value there --
+        # so it needs a driver reading both surfaces off one invocation.
+        "149-malformed-wire-counter-nulled-through-mapping-to-event-and-span",
         # v0.69.0 — proposal 0063 (tool-execution observability). A
         # calls_tool node enters the with_tool_call scope; the typed
         # ToolCallEvent / ToolCallFailedEvent drive the OTel tool span +
@@ -433,15 +439,6 @@ def _reset_otel_global_tracer_provider(restore_to: object) -> None:
     # embedding failure-metrics counterpart).
     # Diagnosed against their sibling drivers rather than guessed: each reaches a
     # driver and fails on a concrete gap, not on "no driver".
-    "149-malformed-wire-counter-nulled-through-mapping-to-event-and-span": (
-        "declares BOTH expected.span_tree and expected.observers.contains_event, and no driver "
-        "serves both: _run_typed_event_cases reads observers only, _run_llm_payload_case reads "
-        "span_tree only, and _run_token_budget_case reads both but builds the graph from a mock "
-        "shape that does not reproduce 149's response model / response id, so its span-tree match "
-        "fails on attributes rather than on behaviour. Needs a driver threading 149's mock_llm "
-        "through a typed collector AND a span exporter; the event half is the fixture's stated "
-        "discriminator, so wiring it under a span-only driver drops the assertion it exists for"
-    ),
     "119-otel-callable-branch-attempt-index-under-node-retry": (
         "the conformance adapter never translates a fixture node's YAML `middleware:` block, so "
         "the retry is never installed and the transient failure is terminal: only attempt 0 runs "
@@ -456,13 +453,6 @@ def _reset_otel_global_tracer_provider(restore_to: object) -> None:
     "153-otel-mixed-nesting-orphan-llm-fallback": (
         "same driver gap as 152, one nesting level deeper (KeyError: 'leaf_sg')"
     ),
-    "148-langfuse-generation-usage-omits-input-on-null-counter": (
-        "the sibling Langfuse runner DOES drive mock_llm generations asserting usage (023, 155, "
-        "156), so plumbing is not the gap. The gap is its `usage` comparator: it iterates the "
-        "EXPECTED keys, a subset match, so an OMITTED `input` key -- the whole claim -- cannot "
-        "be expressed. Wiring it needs an exact-map or usage_absent comparator first, or the "
-        "fixture passes against an impl emitting usage.input = null"
-    ),
     # Proposal 0109 (spec v0.104.0) token-budget failure-path parity.
 }
 
@@ -491,6 +481,11 @@ def _reset_otel_global_tracer_provider(restore_to: object) -> None:
         "059-implementation-attribution-langfuse",
         "155-langfuse-token-budget-exceeded-flag-on-failure",
         "156-langfuse-token-budget-under-budget-flag-false",
+        # 148 (proposal 0101): the Generation's fixed `usage` record omits a
+        # counter the provider did not report. The `usage` comparator is a
+        # subset match, so the OMISSION -- the whole claim -- is carried by the
+        # case's two invariants rather than by the langfuse_trace block.
+        "148-langfuse-generation-usage-omits-input-on-null-counter",
         # 134 -- proposal 0084 Langfuse parent resolution (nested exact-match +
         # orphan fallback). Driven by a dedicated hand-built runner in the
         # sibling harness; the generic topology path cannot model the
@@ -681,6 +676,8 @@ def _reject_unsupported_capability_gate(fixture_id: str, spec: Mapping[str, Any]
     ),
     "_run_rerank_fixture": frozenset({"span_tree", "metrics", "invariants", "observers", "langfuse_trace"}),
     "_run_token_budget_fixture": frozenset({"span_tree", "metrics", "invariants", "observers"}),
+    # 149's driver: reads the typed-event and span halves off one invocation.
+    "_run_typed_event_with_span_cases": frozenset({"span_tree", "observers", "invariants"}),
     # `invariants` is documentary here; `observers` is deliberately ABSENT because
     # _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.
@@ -903,6 +900,9 @@ async def test_observability_fixture(fixture_path: Path) -> None:
         "131-token-budget-on-structured-output-failure",
     }:
         await _run_token_budget_fixture(spec)
+    elif fixture_id == "149-malformed-wire-counter-nulled-through-mapping-to-event-and-span":
+        # Needs the request model bound independently of the mock; see the driver.
+        await _run_typed_event_with_span_cases(spec)
     elif fixture_id in {
         "092-tool-call-event-dispatch",
         "093-tool-call-failed-event-dispatch",
@@ -2632,6 +2632,19 @@ async def _ask_body(
             )
 
 
+def _attr_equal(actual: Any, expected: Any) -> bool:
+    """Compare a span attribute against a fixture's expected value."""
+    # OTel stores a sequence attribute as a TUPLE, while YAML parses the
+    # fixture's `[...]` into a LIST, and `['stop'] == ('stop',)` is False. A
+    # plain `==` therefore rejects every sequence-valued attribute, which read
+    # as "the implementation emitted the wrong value" rather than as a harness
+    # gap: 149 is the first fixture to assert `gen_ai.response.finish_reasons`
+    # 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)
+    return bool(actual == expected)
+
+
 def _assert_span_tree_matches(
     all_spans: Sequence[Any], actual_roots: Sequence[Any], expected_nodes: Sequence[Mapping[str, Any]]
 ) -> None:
@@ -2650,7 +2663,7 @@ def _assert_span_tree_matches(
 
         def _matches(span: Any, eattrs: dict[str, Any] = expected_attrs) -> bool:
             attrs = dict(span.attributes or {})
-            return all(attrs.get(k) == v for k, v in eattrs.items())
+            return all(_attr_equal(attrs.get(k), v) for k, v in eattrs.items())
 
         matching = [c for c in candidates if _matches(c)]
         assert len(matching) >= 1, (
@@ -4511,10 +4524,12 @@ def _walk(expected_entries: list[dict[str, Any]]) -> None:
             # ``attributes:`` block — exact match per key.
             for k, v in cast("dict[str, Any]", entry.get("attributes") or {}).items():
                 actual: Any = attrs.get(k)
-                # OTel attribute arrays come back as tuples; normalize.
-                if isinstance(v, list) and isinstance(actual, tuple):
-                    actual = list(cast("tuple[Any, ...]", actual))
-                assert actual == v, f"span {name!r} attribute {k!r} mismatch: expected {v!r}, got {actual!r}"
+                # Shared with the span-tree walker: one fixture directive
+                # (`attributes:`) is consumed by two comparators in this file,
+                # so a fix to attribute matching must not have two landing sites.
+                assert _attr_equal(actual, v), (
+                    f"span {name!r} attribute {k!r} mismatch: expected {v!r}, got {actual!r}"
+                )
             # ``attributes_absent:`` list of names that MUST NOT appear.
             absent = entry.get("attributes_absent")
             if absent:
@@ -6254,6 +6269,7 @@ def _build_simple_llm_graph(
     case: Mapping[str, Any],
     *,
     populate_caller_metadata: bool,
+    model: str | None = None,
 ) -> tuple[Any, type[Any], Any]:
     """Build a single-node graph that calls the LLM provider against a
     mock transport. Matches the simple entry → ask → END pattern used
@@ -6282,8 +6298,18 @@ def _build_simple_llm_graph(
     # §5.5.7 -- 068 needs this to differ from the provider-returned
     # response_model), else the model the first mock response reports
     # (050-056 path), else a default.
+    # A caller-supplied `model` sits BELOW the fixture's own declaration and above
+    # the mock fallback. Fixtures asserting that `gen_ai.request.model` and
+    # `gen_ai.response.model` DIFFER (144, 149) declare no `calls_llm.model`, so
+    # the mock fallback would bind the request model to the response model and
+    # collapse the distinction they exist to draw; `_run_llm_payload_case`, which
+    # drives 144, resolves that by hardcoding "test-model", and this parameter is
+    # the same convention made reusable. It must NOT outrank `calls_llm.model`:
+    # a driver's blanket default would then silently replace a request model a
+    # case declared, and any case not pinning the model attribute stays green.
     bound_model = (
         cast("str | None", calls_llm_spec.get("model"))
+        or model
         or _mock_model_from_first_response(case)
         or "test-model"
     )
@@ -6667,6 +6693,117 @@ async def _run_typed_event_cases(spec: Mapping[str, Any], *, expect_failure: boo
             raise AssertionError(f"case {case_name!r}: {e}") from e
 
 
+# Fixture 149's six invariant names, all DOCUMENTARY: each restates a claim the
+# case's concrete directives already pin, so there is nothing here for a guard to
+# evaluate. Recognized so they cannot read as unhandled.
+#
+# Documentary is a verdict, not an assumption, so the claims were mutation-tested
+# against src rather than read off the fixture. Coercing a malformed counter to 0
+# instead of nulling it fails `contains_event`; emitting the input-usage span
+# attribute anyway fails `attributes_absent`. Both directives are live.
+#
+# `event_usage_is_present_record_of_nulls_not_null_record` is live like the rest.
+# It was briefly recorded here as an unfalsifiable negative control, on the
+# grounds that `Response.usage` is typed `Usage` rather than `Usage | None`. That
+# reasoned about the wrong layer: the invariant and the fixture's contains_event
+# block are about the EVENT, and `LlmCompletionEvent.usage` IS nullable
+# (graph/events.py). Setting `usage=None` where the event is built
+# (llm/providers/openai.py) is accepted and fails 149's contains_event, so the
+# nested `usage:` mapping does discriminate a present record of nulls from a
+# null record, exactly as the fixture argues.
+_MALFORMED_COUNTER_INVARIANTS = {
+    # Case 1: one malformed wire counter ("abc").
+    "malformed_wire_prompt_tokens_nulled_before_event_and_span",
+    "event_usage_prompt_tokens_null_not_verbatim_wire_value",
+    "span_omits_both_input_usage_attributes_on_malformed_wire_counter",
+    # Case 2: all three malformed ("abc" / -5 / true).
+    "all_three_malformed_wire_counters_nulled_before_event_and_span",
+    "event_usage_is_present_record_of_nulls_not_null_record",
+    "span_omits_all_usage_attributes_when_all_counters_malformed",
+}
+
+
+async def _run_typed_event_with_span_cases(spec: Mapping[str, Any]) -> None:
+    """Fixture 149: cases asserting the SAME null-counter claim on the typed
+    event AND the OTel span, from one invocation."""
+    # 149 declares `observers.contains_event` and `span_tree` together, and the
+    # nulling it pins happens BEFORE the fan-out to both, so both must come off
+    # ONE invocation. The typed-event driver reads observers only and the
+    # LLM-payload driver reads span_tree only, so routing 149 to either silently
+    # drops the other half, which an attempted re-route did in each direction.
+    #
+    # `_run_token_budget_fixture` DOES read both, so this is not "no driver
+    # serves both". What stops it is mechanical, and worth recording because the
+    # next 0101-family fixture gets triaged against it: it binds the request
+    # model from the mock response, so 149's `gen_ai.request.model: test-model`
+    # cannot match a `test-model-2026-07-15` response, and it validates
+    # invariants against the metrics / token-budget families, which reject 149's
+    # six names. Confirmed by running 149 through it -- it reaches the span-tree
+    # match and fails on the model attribute.
+    for case in cast("list[dict[str, Any]]", spec["cases"]):
+        case_name = cast("str", case["name"])
+        try:
+            await _run_typed_event_with_span_case(case)
+        except AssertionError as e:
+            raise AssertionError(f"case {case_name!r}: {e}") from e
+
+
+async def _run_typed_event_with_span_case(case: Mapping[str, Any]) -> None:
+    from opentelemetry.sdk.trace.export import SimpleSpanProcessor  # noqa: PLC0415
+    from opentelemetry.sdk.trace.export.in_memory_span_exporter import (  # noqa: PLC0415
+        InMemorySpanExporter,
+    )
+
+    collectors, populate_caller_metadata = _parse_typed_observers(case)
+    # 149 mirrors 144's span claims, and both assert that the REQUEST model
+    # differs from the model the provider returned, without declaring a
+    # `calls_llm.model`. Bind the same default 144's driver uses.
+    graph, state_cls, provider = _build_simple_llm_graph(
+        case, populate_caller_metadata=populate_caller_metadata, model="test-model"
+    )
+    exporter = InMemorySpanExporter()
+    observer = OTelObserver(span_processor=SimpleSpanProcessor(exporter))
+    # Teardown BEFORE the assertions, matching `_run_token_budget_case`: the
+    # shutdown ends the `openarmature.invocation` root span, so reading the
+    # exporter first sees only the node and LLM spans and the span_tree root
+    # lookup fails on harness ordering rather than on behaviour. Verified by
+    # moving it after, which fails with "invocation root span missing".
+    try:
+        final, exc = await _invoke_typed_fixture(case, collectors, graph, state_cls, extra_observer=observer)
+    finally:
+        await provider.aclose()
+        observer.shutdown()
+
+    # `_invoke_typed_fixture` CAPTURES a NodeException rather than raising, so
+    # surface it: an implementation that raises on a malformed counter instead
+    # of nulling it is one of the two behaviours 149 discriminates, and
+    # reporting that as a bare state assertion drops the cause.
+    assert final is not None, (
+        f"expected a non-None final state on the success path; the invocation raised {exc!r}"
+    )
+
+    expected = cast("dict[str, Any]", case.get("expected") or {})
+    observer_expectations = cast("dict[str, Any]", expected.get("observers") or {})
+    for name, expectations in observer_expectations.items():
+        collector = collectors.get(name)
+        if collector is None:
+            raise AssertionError(f"fixture references unknown observer {name!r}")
+        _assert_observer_expectations(name, collector, cast("dict[str, Any]", expectations))
+
+    spans = exporter.get_finished_spans()
+    expected_tree = cast("list[dict[str, Any]] | None", expected.get("span_tree"))
+    if expected_tree is not None:
+        inv_root = next(
+            (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_error_span_extras(spans, expected_tree)
+
+    _assert_invariants_recognized(case, _MALFORMED_COUNTER_INVARIANTS, "malformed-counter")
+
+
 async def _run_typed_event_chain_cases(spec: Mapping[str, Any], *, expect_failure: bool = False) -> None:
     """Iterate the multi-node-chain typed-event cases (067 success, 071
     failure)."""
diff --git a/tests/conformance/test_observability_langfuse.py b/tests/conformance/test_observability_langfuse.py
index 270c928..edb9cd2 100644
--- a/tests/conformance/test_observability_langfuse.py
+++ b/tests/conformance/test_observability_langfuse.py
@@ -161,6 +161,14 @@
         "130-langfuse-token-budget-warning-level",
         "155-langfuse-token-budget-exceeded-flag-on-failure",
         "156-langfuse-token-budget-under-budget-flag-false",
+        # 148 (proposal 0101): the Generation's FIXED `usage` record omits a
+        # counter the provider did not report, unlike the Embedding's open
+        # `usageDetails` map (140) where absence is naturally expressible. The
+        # `usage` comparator here iterates the EXPECTED keys, so declining to
+        # declare `input` asserts nothing; the case's two invariants carry the
+        # omission claim, and both are implemented in
+        # `_assert_generation_usage_omission`, which `_assert_trace` calls.
+        "148-langfuse-generation-usage-omits-input-on-null-counter",
         # 134 (proposal 0084): the Langfuse Generation parent resolves by the same
         # chain-aware §5.5 rule as the OTel span parent -- both the nested
         # exact-match (case 1, mirrors OTel 132) and the orphan fallback (case 2,
@@ -2282,13 +2290,22 @@ def _runtime_config_from_spec(config_spec: dict[str, Any] | None) -> RuntimeConf
 # ``correlation_id_consistent_across_traces``, etc.) stay in
 # ``_assert_multi_traces`` as cross-Trace checks.
 #
-# Must list EVERY name ``_assert_trace`` reads: a name it checks but that is
-# missing here is silently discarded on the multi-trace path, so a fixture
-# declaring it there passes without the claim being evaluated.
-# ``test_harness_fidelity.py`` derives the read set from the function body and
-# fails on any omission -- ``no_warning_level_under_budget`` was one.
+# Must list EVERY name ``_assert_trace`` OR A HELPER IT DELEGATES TO reads: a
+# name checked but missing here is silently discarded on the multi-trace path,
+# so a fixture declaring it there passes without the claim being evaluated.
+# The converse also binds: every name here must be EVALUATED by one of those
+# guards, or listed in ``test_harness_fidelity._DOCUMENTARY_PER_TRACE_INVARIANTS``
+# with its reason, so a guard gutted to ``if name: pass`` cannot read as live.
+# ``test_harness_fidelity.py`` derives both sets by walking those bodies and
+# fails either way -- ``no_warning_level_under_budget`` was a missing one.
 _PER_TRACE_INVARIANTS = frozenset(
-    {"trace_id_equals_invocation_id", "correlation_id_consistency", "no_warning_level_under_budget"}
+    {
+        "trace_id_equals_invocation_id",
+        "correlation_id_consistency",
+        "no_warning_level_under_budget",
+        "generation_usage_input_omitted_when_prompt_tokens_null",
+        "generation_usage_output_and_total_present_when_sound",
+    }
 )
 
 
@@ -2404,6 +2421,60 @@ def _assert_multi_traces(
                         )
 
 
+def _assert_generation_usage_omission(trace: LangfuseTrace, expected_invariants: dict[str, Any]) -> None:
+    """Fixture 148: a counter the provider did not report is OMITTED from the
+    Generation's fixed `usage` record, while the sound counters stay present."""
+    # These two are the whole of 148's claim. The `langfuse_trace` block cannot
+    # carry it: `usage` is compared by iterating the EXPECTED keys, so a key the
+    # fixture declines to declare is never looked at, and an implementation
+    # emitting `input: null` passes the block unchanged.
+    #
+    # `None` is how omission is spelled on this side of the boundary. The
+    # in-memory double models `usage` as a dataclass, where every field exists
+    # and an unreported counter reads as None; the SDK adapter builds the wire
+    # `usage_details` dict by skipping exactly the None fields
+    # (langfuse/adapter.py), so None here is the faithful proxy for key-absence
+    # there. Asserting `"input" not in ...` would be checking the double's
+    # dataclass shape rather than the mapping under test.
+    omitted = expected_invariants.get("generation_usage_input_omitted_when_prompt_tokens_null")
+    present = expected_invariants.get("generation_usage_output_and_total_present_when_sound")
+    if not omitted and not present:
+        return
+    generations = [o for o in trace.observations if o.type == "generation"]
+    # Positive anchor: every claim below reads a Generation, and no Generation
+    # at all would satisfy the omission half trivially.
+    assert generations, (
+        f"no Generation observation was recorded, so the usage claims below would pass "
+        f"vacuously; got {[(o.name, o.type) for o in trace.observations]}"
+    )
+    if omitted:
+        # Self-anchoring: skipping a Generation whose `usage` is None would make
+        # this satisfied by an implementation emitting no usage record at all.
+        # The `present` arm below rules that out too, but only when the fixture
+        # happens to declare BOTH names, and declaration lives in the spec repo.
+        carrying: list[tuple[str | None, Any]] = []
+        for gen in generations:
+            assert gen.usage is not None, (
+                f"generation {gen.name!r} carries no usage record at all, so the omission "
+                f"claim would pass vacuously"
+            )
+            if gen.usage.input is not None:
+                carrying.append((gen.name, gen.usage))
+        assert not carrying, (
+            f"an unreported prompt_tokens MUST be omitted from generation.usage rather than "
+            f"rendered as null or zero; present on {carrying}"
+        )
+    if present:
+        # The non-vacuity partner: without it the omission above is satisfied by
+        # an implementation that emitted no usage record at all.
+        for gen in generations:
+            assert gen.usage is not None, f"generation {gen.name!r} carries no usage record at all"
+            assert gen.usage.output is not None and gen.usage.total is not None, (
+                f"generation {gen.name!r} MUST still carry the sound counters; got "
+                f"output={gen.usage.output!r} total={gen.usage.total!r}"
+            )
+
+
 def _assert_trace(
     trace: LangfuseTrace,
     expected: dict[str, Any],
@@ -2421,6 +2492,7 @@ def _assert_trace(
             f"an under-budget call MUST NOT render a WARNING-level observation; got "
             f"{[(o.name, o.level) for o in offending]}"
         )
+    _assert_generation_usage_omission(trace, expected_invariants)
     expected_id = expected.get("id")
     if expected_id is not None and not _is_placeholder(expected_id):
         # Fixtures 035/036: a LITERAL trace.id is the DERIVED Langfuse id; the
diff --git a/tests/unit/test_observability_langfuse_adapter.py b/tests/unit/test_observability_langfuse_adapter.py
index a2f2527..cddbe4b 100644
--- a/tests/unit/test_observability_langfuse_adapter.py
+++ b/tests/unit/test_observability_langfuse_adapter.py
@@ -22,6 +22,7 @@
 from __future__ import annotations
 
 import os
+from types import SimpleNamespace
 from typing import Annotated, Any
 
 import pytest
@@ -36,6 +37,7 @@
     LangfuseClient,
     LangfuseObserver,
     LangfuseSDKAdapter,
+    LangfuseUsage,
 )
 
 
@@ -403,3 +405,45 @@ async def test_adapter_against_real_langfuse_cloud() -> None:
     # The trace_id in the dashboard is the 32-char hex form (no dashes)
     # of OA's UUID4 invocation_id; strip dashes from any logged
     # correlation_id / invocation_id to find it.
+
+
+@pytest.mark.parametrize(
+    ("usage", "expected"),
+    [
+        # 0101: a counter the provider did not report is OMITTED from the wire
+        # record, not rendered as null or zero.
+        (LangfuseUsage(input=None, output=5, total=15), {"output": 5, "total": 15}),
+        (LangfuseUsage(input=7, output=5, total=15), {"input": 7, "output": 5, "total": 15}),
+        (LangfuseUsage(input=0, output=5, total=15), {"input": 0, "output": 5, "total": 15}),
+    ],
+)
+def test_generation_usage_details_omit_unreported_counters(
+    monkeypatch: pytest.MonkeyPatch,
+    usage: LangfuseUsage,
+    expected: dict[str, int],
+) -> None:
+    # The conformance fixture for this claim (148) asserts `usage.input is None`
+    # on the in-memory double, where omission and a rendered null are the SAME
+    # state. This is the other side of that proxy: the adapter is what turns the
+    # record into the wire `usage_details` map, and until this existed the guard
+    # there could be neutralised at all four sites with the whole suite green.
+    #
+    # The zero row is the reason `if usage.input is not None` cannot become a
+    # truthiness check: a genuinely reported 0 MUST still render.
+    adapter = LangfuseSDKAdapter(_dummy_client())
+    captured: dict[str, Any] = {}
+
+    def _capture(**kwargs: Any) -> Any:
+        captured.update(kwargs)
+
+        def _noop(**_kwargs: Any) -> None:
+            return None
+
+        return SimpleNamespace(id="obs-1", update=_noop, end=_noop)
+
+    monkeypatch.setattr(adapter, "_start_observation", _capture)
+    adapter.generation(trace_id="tr-1", name="openarmature.llm.complete", usage=usage)
+
+    assert captured["usage_details"] == expected, (
+        f"usage_details must carry exactly the reported counters; got {captured.get('usage_details')!r}"
+    )