diff --git a/README.md b/README.md
index 4fe6cb7..52f3d98 100644
--- a/README.md
+++ b/README.md
@@ -65,6 +65,19 @@ The existing `orchestrator:provider_budget` event exposes each preflight's
 attempt, result, estimate, allowance, and requested context budget to mounted
 observability consumers.
 
+The same event also reports the absence of a native count, so a session that
+silently fell back to estimate-based behavior is visible rather than invisible.
+An unavailable report carries `result: "unavailable"`, a `mode` of
+`initial_fallback`, `post_concrete_failure`, or `measured`, and a `reason` of
+`capability_missing`, `no_decision`, or `measurement_absent`; `attempt` is
+included only where the preflight naturally knows it. It carries no count, no
+fit, no provider exception, and no request or context data, and providers may
+separately report their own provider-specific reasons. These reports are
+diagnostic only: the compatibility fallback and the fail-closed capability-loss
+error both keep their existing behavior, a malformed advertised budget stays an
+error rather than becoming a fallback, and a provider that never advertised
+`request_budget` keeps its historical silent path.
+
 ### Provider-count measured compaction
 
 When both sides advertise the optional measured contracts —
diff --git a/amplifier_module_loop_streaming/__init__.py b/amplifier_module_loop_streaming/__init__.py
index df523b3..afbc6ba 100644
--- a/amplifier_module_loop_streaming/__init__.py
+++ b/amplifier_module_loop_streaming/__init__.py
@@ -3552,6 +3552,34 @@ def build_chat_request(
                 **kwargs
             )
 
+        async def emit_budget_unavailable(
+            *, mode: str, reason: str, attempt: int | None = None
+        ) -> None:
+            """Report that an invoked budget path produced no usable count.
+
+            Diagnostic only. This never changes a decision: it is emitted
+            beside the existing compatibility fallback or immediately before
+            the existing fail-closed raise, and a failing observability hook
+            must not convert either into a different outcome. Cancellation
+            still propagates -- ``asyncio.CancelledError`` is a
+            ``BaseException`` and is deliberately not caught here.
+
+            The payload is scalar and provider-agnostic: no count, no fit, no
+            raw provider exception, and no request, prompt, or context data.
+            """
+            payload: dict[str, Any] = {
+                "result": "unavailable",
+                "mode": mode,
+                "reason": reason,
+            }
+            if attempt is not None:
+                payload["attempt"] = attempt
+            try:
+                await hooks.emit("orchestrator:provider_budget", payload)
+            except Exception:
+                # A diagnostic hook may include request data in its exception.
+                logger.debug("Provider budget unavailability event failed to emit")
+
         async def check_request_budget(
             request: ChatRequest,
             base_messages: list[dict[str, Any]],
@@ -3572,6 +3600,28 @@ async def check_request_budget(
             """
             request_budget = getattr(provider, "request_budget", None)
             if not budget_capable or not callable(request_budget):
+                # A provider that never advertised the capability keeps the
+                # historical silent path: this flow never invoked a budget
+                # check, so it gains no event. Only a capability that was
+                # present at provider selection and is missing here is a
+                # reportable unavailability.
+                if budget_capable:
+                    if attempt and not allow_unproven:
+                        await emit_budget_unavailable(
+                            mode="post_concrete_failure",
+                            reason="capability_missing",
+                            attempt=attempt,
+                        )
+                        raise ContextLengthError(
+                            "Provider request_budget capability was unavailable after "
+                            "reporting a concrete budget"
+                        )
+                    await emit_budget_unavailable(
+                        mode="initial_fallback",
+                        reason="capability_missing",
+                        attempt=attempt,
+                    )
+                    return None, None
                 if attempt and not allow_unproven:
                     raise ContextLengthError(
                         "Provider request_budget capability was unavailable after "
@@ -3587,10 +3637,20 @@ async def check_request_budget(
                 decision = await decision
             if decision is None:
                 if attempt and not allow_unproven:
+                    await emit_budget_unavailable(
+                        mode="post_concrete_failure",
+                        reason="no_decision",
+                        attempt=attempt,
+                    )
                     raise ContextLengthError(
                         "Provider request_budget capability was unavailable after "
                         "reporting a concrete budget"
                     )
+                await emit_budget_unavailable(
+                    mode="initial_fallback",
+                    reason="no_decision",
+                    attempt=attempt,
+                )
                 return None, None
             required = (
                 "estimated_input_tokens",
@@ -3650,6 +3710,9 @@ async def count_measured_request(
             """
             request_budget = getattr(provider, "request_budget", None)
             if not callable(request_budget):
+                await emit_budget_unavailable(
+                    mode="measured", reason="capability_missing"
+                )
                 return None
             context_estimate = sum(len(str(message)) // 4 for message in base_view)
             kwargs: dict[str, Any] = {"context_estimate": context_estimate}
@@ -3659,6 +3722,7 @@ async def count_measured_request(
             if inspect.isawaitable(decision):
                 decision = await decision
             if decision is None:
+                await emit_budget_unavailable(mode="measured", reason="no_decision")
                 return None
             required = (
                 "estimated_input_tokens",
@@ -3674,7 +3738,14 @@ async def count_measured_request(
                 raise ContextLengthError(
                     "Provider request_budget returned an invalid budget decision"
                 )
-            validate_measured_budget_decision(decision)
+            # Malformed advertised measurements still raise from here: an
+            # error must never be downgraded into a compatibility fallback.
+            # Only a structurally valid decision that carries no usable
+            # provider count is reported as an unavailable measurement.
+            if validate_measured_budget_decision(decision) is None:
+                await emit_budget_unavailable(
+                    mode="measured", reason="measurement_absent"
+                )
             return decision
 
         def validate_measured_budget_decision(
diff --git a/tests/test_measured_compaction_runtime.py b/tests/test_measured_compaction_runtime.py
index 33e50c7..555ae44 100644
--- a/tests/test_measured_compaction_runtime.py
+++ b/tests/test_measured_compaction_runtime.py
@@ -640,3 +640,232 @@ async def get_measured_request_view(self, **kwargs):
 
     assert provider.requests == []
     assert context.transactions[0].rolled_back == 1
+
+
+# ---------------------------------------------------------------------------
+# Native-count unavailability on the measured path. Every assertion below
+# reads the payloads the REAL orchestrator emitted through the hooks object
+# it was handed -- never a logger-only or helper-level probe.
+# ---------------------------------------------------------------------------
+
+
+def _budget_payloads(hooks) -> list[dict]:
+    return [
+        payload
+        for event, payload in hooks.emitted
+        if event == "orchestrator:provider_budget"
+    ]
+
+
+class _CountTolerantContext(_MeasuredContext):
+    """Measured context double that tolerates a decision with no usable count."""
+
+    async def get_measured_request_view(self, *, provider, retain_contents, count_view):
+        base_view = list(self._messages)
+        attempt = await count_view(base_view)
+        transaction = _Transaction()
+        self.transactions.append(transaction)
+        self.measured_calls.append((provider, list(retain_contents)))
+        decision = attempt["budget_decision"]
+        measurement = (decision or {}).get("measurement") or {}
+        count = measurement.get("input_tokens")
+        return {
+            "base_view": base_view,
+            "final_attempt": attempt,
+            "outcome": "not_needed",
+            "measured_before": count,
+            "measured_after": count,
+            "policy_budget": 100,
+            "trigger": 80.0,
+            "target": 50,
+            "count_calls": 1,
+            "transaction": transaction,
+        }
+
+
+class _NoMeasurementProvider(_MeasuredProvider):
+    """Advertises provider_count but returns a valid decision without one."""
+
+    def request_budget(self, request, *, context_estimate, request_options=None):
+        self.budget_calls.append(request)
+        return {
+            "estimated_input_tokens": 9,
+            "input_limit_tokens": 100,
+            "context_token_budget": 0,
+        }
+
+
+class _NoDecisionProvider(_MeasuredProvider):
+    def request_budget(self, request, *, context_estimate, request_options=None):
+        self.budget_calls.append(request)
+        return None
+
+
+class _MalformedMeasurementProvider(_MeasuredProvider):
+    """Structurally invalid envelope: must stay fail-closed, not fall back."""
+
+    def request_budget(self, request, *, context_estimate, request_options=None):
+        self.budget_calls.append(request)
+        return {
+            **_provider_count(),
+            "max_output_tokens": 0,
+        }
+
+
+def _measured_coordinator(context) -> MockCoordinator:
+    coordinator = MockCoordinator()
+    coordinator.register_capability(
+        "context.measured_request_view", context.get_measured_request_view
+    )
+    return coordinator
+
+
+@pytest.mark.asyncio
+async def test_measured_path_reports_a_decision_without_a_usable_count() -> None:
+    context = _CountTolerantContext()
+    provider = _NoMeasurementProvider()
+    hooks = ScriptedHooks({})
+
+    await StreamingOrchestrator({}).execute(
+        "current request",
+        context,
+        {"main": provider},
+        {},
+        hooks,
+        _measured_coordinator(context),
+    )
+
+    assert len(provider.requests) == 1
+    assert _budget_payloads(hooks) == [
+        {
+            "result": "unavailable",
+            "mode": "measured",
+            "reason": "measurement_absent",
+        }
+    ]
+
+
+@pytest.mark.asyncio
+async def test_measured_path_reports_a_none_budget_decision() -> None:
+    context = _CountTolerantContext()
+    provider = _NoDecisionProvider()
+    hooks = ScriptedHooks({})
+
+    await StreamingOrchestrator({}).execute(
+        "current request",
+        context,
+        {"main": provider},
+        {},
+        hooks,
+        _measured_coordinator(context),
+    )
+
+    assert len(provider.requests) == 1
+    assert _budget_payloads(hooks) == [
+        {
+            "result": "unavailable",
+            "mode": "measured",
+            "reason": "no_decision",
+        }
+    ]
+
+
+@pytest.mark.asyncio
+async def test_measured_valid_count_emits_no_unavailable_event() -> None:
+    context = _MeasuredContext()
+    provider = _MeasuredProvider()
+    hooks = ScriptedHooks({})
+
+    await StreamingOrchestrator({}).execute(
+        "current request",
+        context,
+        {"main": provider},
+        {},
+        hooks,
+        _measured_coordinator(context),
+    )
+
+    payloads = _budget_payloads(hooks)
+    assert [payload["result"] for payload in payloads] == ["fits"]
+    assert all(payload["result"] != "unavailable" for payload in payloads)
+    assert payloads[0]["measurement_kind"] == "provider_count"
+
+
+@pytest.mark.asyncio
+async def test_malformed_measured_envelope_stays_fail_closed_without_fallback() -> None:
+    context = _CountTolerantContext()
+    provider = _MalformedMeasurementProvider()
+    hooks = ScriptedHooks({})
+
+    with pytest.raises(ContextLengthError, match="invalid max_output_tokens"):
+        await StreamingOrchestrator({}).execute(
+            "current request",
+            context,
+            {"main": provider},
+            {},
+            hooks,
+            _measured_coordinator(context),
+        )
+
+    assert provider.requests == []
+    assert _budget_payloads(hooks) == []
+
+
+class _NoMeasurementFinalizingProvider(NRoundToolProvider):
+    """One tool round then finalization; never rejects, so no recovery probe.
+
+    Deliberately NOT derived from ``_MeasuredFinalizingProvider``: that double
+    raises ``ContextLengthError`` on its second completion to exercise overflow
+    recovery, which would add a third budget invocation and make the
+    one-event-per-invocation assertion below ambiguous.
+    """
+
+    def __init__(self) -> None:
+        super().__init__(n_tool_rounds=1)
+        self.budget_calls: list[object] = []
+
+    def get_info(self):
+        return SimpleNamespace(capabilities=["request_budget:provider_count"])
+
+    def request_budget(self, request, *, context_estimate, request_options=None):
+        self.budget_calls.append(request)
+        return {
+            "estimated_input_tokens": 9,
+            "input_limit_tokens": 100,
+            "context_token_budget": 0,
+        }
+
+
+@pytest.mark.asyncio
+async def test_measured_finalization_reports_its_absent_count_once_per_call() -> None:
+    context = _CountTolerantContext()
+    provider = _NoMeasurementFinalizingProvider()
+    hooks = ScriptedHooks({})
+    coordinator = _retaining_coordinator(context)
+    coordinator.register_capability(
+        "context.measured_request_view", context.get_measured_request_view
+    )
+
+    await StreamingOrchestrator({"max_iterations": 1}).execute(
+        "current request",
+        context,
+        {"main": provider},
+        {"mock_tool": OneShotTool()},
+        hooks,
+        coordinator,
+    )
+
+    # One event per budget invocation: ordinary turn, then finalization.
+    assert _budget_payloads(hooks) == [
+        {
+            "result": "unavailable",
+            "mode": "measured",
+            "reason": "measurement_absent",
+        },
+        {
+            "result": "unavailable",
+            "mode": "measured",
+            "reason": "measurement_absent",
+        },
+    ]
+    assert len(provider.budget_calls) == len(_budget_payloads(hooks))
diff --git a/tests/test_provider_budget_guard.py b/tests/test_provider_budget_guard.py
index a09d1d0..64d8c9f 100644
--- a/tests/test_provider_budget_guard.py
+++ b/tests/test_provider_budget_guard.py
@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import asyncio
+
 import pytest
 from amplifier_core import ContextLengthError
 
@@ -56,6 +58,63 @@ async def request_budget(self, request, *, context_estimate: int) -> object:
         return self.decisions.pop(0)
 
 
+class UnavailableDiagnosticFailureHooks(ScriptedHooks):
+    async def emit(self, event_name, payload=None):
+        if (
+            event_name == "orchestrator:provider_budget"
+            and payload.get("result") == "unavailable"
+        ):
+            raise self.failure
+        return await super().emit(event_name, payload)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("post_concrete", [False, True])
+async def test_unavailable_hook_failure_is_safe_and_preserves_outcome(
+    caplog, post_concrete
+):
+    marker = "SYNTHETIC-SENSITIVE-HOOK-ERROR"
+    hooks = UnavailableDiagnosticFailureHooks({})
+    hooks.failure = RuntimeError(marker)
+    context = BudgetContext()
+    provider = AsyncBudgetProvider(
+        [_decision(100, 10, 7), None] if post_concrete else [None]
+    )
+    caplog.set_level("DEBUG")
+    call = StreamingOrchestrator({}).execute(
+        "work", context, {"main": provider}, {}, hooks,
+        _retaining_coordinator(context),
+    )
+    if post_concrete:
+        with pytest.raises(ContextLengthError, match="unavailable after"):
+            await call
+        assert not provider.requests
+    else:
+        await call
+        assert len(provider.requests) == 1
+    assert "Provider budget unavailability event failed to emit" in caplog.text
+    assert marker not in caplog.text
+    assert all(
+        record.exc_info is None
+        for record in caplog.records
+        if "unavailability event failed" in record.message
+    )
+
+
+@pytest.mark.asyncio
+async def test_unavailable_diagnostic_cancellation_still_propagates():
+    hooks = UnavailableDiagnosticFailureHooks({})
+    hooks.failure = asyncio.CancelledError()
+    context = BudgetContext()
+    provider = AsyncBudgetProvider([None])
+    with pytest.raises(asyncio.CancelledError):
+        await StreamingOrchestrator({}).execute(
+            "work", context, {"main": provider}, {}, hooks,
+            _retaining_coordinator(context),
+        )
+    assert not provider.requests
+
+
 class BudgetContext(MockContext):
     """Context double that records ordinary and retention-budget requests."""
 
@@ -184,6 +243,19 @@ def _injection(body: str) -> ScriptedHookResult:
     )
 
 
+def _budget_payloads(hooks) -> list[dict]:
+    """Return every provider-budget payload the real orchestrator emitted."""
+    return [
+        payload
+        for event, payload in hooks.emitted
+        if event == "orchestrator:provider_budget"
+    ]
+
+
+def _budget_results(hooks) -> list[str]:
+    return [payload["result"] for payload in _budget_payloads(hooks)]
+
+
 @pytest.mark.asyncio
 async def test_provider_without_budget_capability_keeps_single_normal_dispatch() -> None:
     context = MockContext()
@@ -283,7 +355,7 @@ async def test_awaitable_fitting_budget_dispatches_the_original_request_once() -
 
 
 @pytest.mark.asyncio
-async def test_initial_unavailable_budget_keeps_normal_dispatch_without_budget_event() -> None:
+async def test_initial_unavailable_budget_keeps_normal_dispatch_and_reports_it() -> None:
     context = BudgetContext()
     provider = BudgetProvider([None])
     hooks = ScriptedHooks({})
@@ -300,7 +372,33 @@ async def test_initial_unavailable_budget_keeps_normal_dispatch_without_budget_e
     assert len(provider.requests) == 1
     assert len(provider.budget_calls) == 1
     assert context.request_calls == [([], None)]
-    assert [name for name, _ in hooks.emitted].count("orchestrator:provider_budget") == 0
+    assert _budget_payloads(hooks) == [
+        {
+            "result": "unavailable",
+            "mode": "initial_fallback",
+            "reason": "no_decision",
+            "attempt": 0,
+        }
+    ]
+
+
+@pytest.mark.asyncio
+async def test_provider_without_budget_capability_emits_no_unavailable_event() -> None:
+    context = BudgetContext()
+    provider = RequestCapturingProvider()
+    hooks = ScriptedHooks({})
+
+    await StreamingOrchestrator({}).execute(
+        "work",
+        context,
+        {"main": provider},
+        {},
+        hooks,
+        _retaining_coordinator(context),
+    )
+
+    assert len(provider.requests) == 1
+    assert _budget_payloads(hooks) == []
 
 
 @pytest.mark.asyncio
@@ -701,11 +799,13 @@ async def test_awaitable_unavailable_budget_after_rebuild_fails_without_sdk_disp
     assert len(provider.budget_calls) == 2
     assert context.hard_fit_calls == [False, True]
     assert provider.requests == []
-    assert [
-        payload["result"]
-        for event, payload in hooks.emitted
-        if event == "orchestrator:provider_budget"
-    ] == ["oversized"]
+    assert _budget_results(hooks) == ["oversized", "unavailable"]
+    assert _budget_payloads(hooks)[-1] == {
+        "result": "unavailable",
+        "mode": "post_concrete_failure",
+        "reason": "no_decision",
+        "attempt": 1,
+    }
 
 
 @pytest.mark.asyncio
@@ -936,19 +1036,35 @@ async def test_finalization_request_is_budget_checked_before_dispatch() -> None:
 async def test_finalization_initial_unavailable_budget_keeps_normal_dispatch() -> None:
     context = BudgetContext()
     provider = FinalizingBudgetProvider([None, None])
+    hooks = ScriptedHooks({})
 
     await StreamingOrchestrator({"max_iterations": 1}).execute(
         "work",
         context,
         {"main": provider},
         {"mock_tool": OneShotTool()},
-        ScriptedHooks({}),
+        hooks,
         _retaining_coordinator(context),
     )
 
     assert len(provider.requests) == 2
     assert len(provider.budget_calls) == 2
     assert provider.requests[-1].tool_choice == "none"
+    # Ordinary turn and finalization report the same unavailability shape.
+    assert _budget_payloads(hooks) == [
+        {
+            "result": "unavailable",
+            "mode": "initial_fallback",
+            "reason": "no_decision",
+            "attempt": 0,
+        },
+        {
+            "result": "unavailable",
+            "mode": "initial_fallback",
+            "reason": "no_decision",
+            "attempt": 0,
+        },
+    ]
 
 
 @pytest.mark.asyncio