fix: expose unavailable provider budget diagnostics safely

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 —

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(

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))