fix(agent): stop re-billing deterministic empty responses (NS-503)

Every empty-response retry re-sends the full conversation input at full
price. On large contexts a single turn that produces no visible output
could bill the user several dollars across the 3-retry + fallback-chain
walk (reported: ~$2.33 for one empty answer on a ~26K-token session).

Signaled refusals (finish_reason=content_filter, Anthropic refusal
stop_reason, guardrail interventions) are already terminal today and
never reach this loop. The uncovered class is *unsignaled* refusals:
the provider returns 200 with zero output tokens and a generic finish
reason. Those are deterministic — resending the identical prompt
reproduces the same empty — so burning the remaining retry budget only
multiplies the charge.

New agent/empty_response_guard.py, two independent guards, both failing
OPEN to today's behaviour:

- Deterministic-empty detection: two consecutive empty attempts with
  usage present, output_tokens == 0 (reasoning tokens count as output),
  and identical (model, provider, finish_reason) skip the remaining
  retries and go straight to the fallback chain — a different model may
  well answer. Missing usage, nonzero output, or any signature change
  keeps the full budget.
- Cost-aware retry budget: when one attempt's estimated input cost
  exceeds HERMES_EMPTY_RETRY_COST_THRESHOLD_USD (default $0.25), the
  empty-retry budget drops 3 -> 1 for that streak. Unknown pricing or
  included/subscription routes are untouched.

At exhaustion the status trace now includes the estimated cost of the
empty attempts so the charge is at least explained in-session.

Streak state lives on the agent and self-clears whenever
_empty_content_retries resets to 0, transparently honouring every
existing reset site (turn start, tool success, compaction, fallback
activation) without touching them.

Set HERMES_DETERMINISTIC_EMPTY_GUARD=0 to disable both guards.

Tests: tests/agent/test_empty_response_guard.py (26 unit tests) plus
two loop-level integration tests in tests/run_agent/test_run_agent.py
proving the api_call reduction and the fail-open path.

Refs NS-503.

agent/conversation_loop.py

@@ -93,6 +93,7 @@ from agent.trajectory import has_incomplete_scratchpad
 # finalizer at turn end.
 from agent.turn_finalizer import finalize_turn
 from agent.usage_pricing import estimate_usage_cost, normalize_usage
+from agent import empty_response_guard as _empty_guard
 from hermes_constants import PARTIAL_STREAM_STUB_ID
 from hermes_logging import set_session_context
 from tools.skill_provenance import set_current_write_origin
@@ -7603,7 +7604,36 @@ def run_conversation(
                         _has_structured
                         and agent._thinking_prefill_retries >= 2
                     )
-                    if _truly_empty and (not _has_structured or _prefill_exhausted) and agent._empty_content_retries < 3:
+                    _empty_candidate = _truly_empty and (
+                        not _has_structured or _prefill_exhausted
+                    )
+                    if _empty_candidate:
+                        # NS-503: every empty attempt re-sends the full
+                        # conversation input at full price. Record the
+                        # attempt (usage/finish_reason signature) so
+                        # deterministic empties — e.g. unsignaled
+                        # provider refusals with zero output tokens —
+                        # stop burning paid retries reproducing the
+                        # same empty. Fails open: missing usage or
+                        # any generated tokens keep the full budget.
+                        _empty_guard.record_empty_attempt(
+                            agent,
+                            finish_reason=finish_reason,
+                            response=response,
+                        )
+                    _empty_retry_budget = (
+                        _empty_guard.empty_retry_budget(agent, response)
+                        if _empty_candidate
+                        else _empty_guard.DEFAULT_EMPTY_RETRY_BUDGET
+                    )
+                    _deterministic_empty = _empty_candidate and (
+                        _empty_guard.deterministic_empty(agent)
+                    )
+                    if (
+                        _empty_candidate
+                        and agent._empty_content_retries < _empty_retry_budget
+                        and not _deterministic_empty
+                    ):
                         agent._empty_content_retries += 1
                         wait_time = jittered_backoff(
                             agent._empty_content_retries,
@@ -7612,12 +7642,19 @@ def run_conversation(
                         )
                         logger.warning(
                             "Empty response (no content or reasoning) — "
-                            "retry %d/3 in %.1fs (model=%s)",
-                            agent._empty_content_retries, wait_time, agent.model,
+                            "retry %d/%d in %.1fs (model=%s)",
+                            agent._empty_content_retries,
+                            _empty_retry_budget, wait_time, agent.model,
+                        )
+                        _budget_note = (
+                            " — high-cost request, reduced retry budget"
+                            if _empty_retry_budget < _empty_guard.DEFAULT_EMPTY_RETRY_BUDGET
+                            else ""
                         )
                         agent._buffer_status(
                             f"⚠️ Empty response from model — retrying "
-                            f"({agent._empty_content_retries}/3) in {wait_time:.0f}s"
+                            f"({agent._empty_content_retries}/{_empty_retry_budget}) "
+                            f"in {wait_time:.0f}s{_budget_note}"
                         )
                         # Sleep in small increments to stay responsive to interrupts
                         sleep_end = time.time() + wait_time
@@ -7627,7 +7664,7 @@ def run_conversation(
                                 agent._vprint(f"{agent.log_prefix}⚡ Interrupt detected during empty-response retry wait, aborting.", force=True)
                                 _interrupt_text = (
                                     f"Operation interrupted: retrying empty response from model "
-                                    f"(retry {agent._empty_content_retries}/3)."
+                                    f"(retry {agent._empty_content_retries}/{_empty_retry_budget})."
                                 )
                                 close_interrupted_tool_sequence(messages, _interrupt_text)
                                 agent._persist_session(messages, conversation_history)
@@ -7643,11 +7680,25 @@ def run_conversation(
                             _backoff_touch_counter += 1
                             if _backoff_touch_counter % 150 == 0:  # 150 × 0.2s = 30s
                                 agent._touch_activity(
-                                    f"empty response retry backoff ({agent._empty_content_retries}/3), "
+                                    f"empty response retry backoff ({agent._empty_content_retries}/{_empty_retry_budget}), "
                                     f"{int(sleep_end - time.time())}s remaining"
                                 )
                         continue
 
+                    if _truly_empty and _deterministic_empty:
+                        logger.warning(
+                            "Deterministic empty response detected "
+                            "(consecutive zero-output completions, "
+                            "model=%s provider=%s finish_reason=%s) — "
+                            "skipping remaining retries",
+                            agent.model, agent.provider, finish_reason,
+                        )
+                        agent._buffer_status(
+                            "⚠️ Model is deterministically returning empty "
+                            "(zero output tokens) — skipping further retries "
+                            "to avoid repeat charges"
+                        )
+
                     # ── Exhausted retries — try fallback provider ──
                     # Before giving up with "(empty)", attempt to
                     # switch to the next provider in the fallback
@@ -7694,6 +7745,17 @@ def run_conversation(
                     # "(empty)" terminal.
                     # Surface the buffered retry/fallback trace so the
                     # user can see what was attempted before "(empty)".
+                    # NS-503: if we know roughly what the empty streak
+                    # cost (each attempt re-billed the full input), say
+                    # so — an unexplained charge for "no answer" is the
+                    # core of the complaint.
+                    _streak_cost = _empty_guard.streak_cost_usd(agent)
+                    if _streak_cost is not None:
+                        agent._buffer_status(
+                            f"ℹ️ Estimated cost of these empty attempts: "
+                            f"~${_streak_cost:.2f} (input tokens are billed "
+                            f"per attempt even when no answer is produced)"
+                        )
                     agent._flush_status_buffer()
                     _turn_exit_reason = "empty_response_exhausted"
                     reasoning_text = agent._extract_reasoning(assistant_message)

agent/empty_response_guard.py

@@ -0,0 +1,230 @@
+"""Deterministic-empty detection and cost-aware retry budgets (NS-503).
+
+When a provider returns an empty completion, the agent loop retries up to
+3 times and then walks the fallback chain. Every attempt re-sends the full
+conversation input — at large context on paid routes this bills the user
+repeatedly for a turn that produces no text (the "charged ~$2.33 for an
+empty answer" incident class).
+
+Signaled refusals (``finish_reason="content_filter"``, Anthropic
+``stop_reason="refusal"``, Bedrock guardrails) are already terminal and
+never reach the empty-retry loop. This module addresses the *unsignaled*
+empties: the provider reports a successful completion with zero output
+tokens and a generic finish reason (portal-proxied refusals commonly look
+like this).
+
+Two independent guards, both failing OPEN to today's behaviour:
+
+1. **Deterministic-empty detection** — two consecutive empty attempts,
+   both with usage present and ``output_tokens == 0``, from the same
+   (model, provider, finish_reason), are treated as deterministic: the
+   same prompt will keep producing the same empty. Remaining retries are
+   skipped and the loop proceeds straight to the fallback chain (a
+   different model may behave differently). Attempts with missing usage
+   or ``output_tokens > 0`` (model generated *something* — think-block
+   stripping, whitespace, flaky decoding) never classify as deterministic
+   and keep the full retry budget.
+
+2. **Cost-aware retry budget** — when the estimated input cost of a
+   single empty attempt exceeds ``HERMES_EMPTY_RETRY_COST_THRESHOLD_USD``
+   (default $0.25), the empty-retry budget for this streak drops from 3
+   to 1. Unknown pricing, missing usage, or included/subscription routes
+   leave the budget untouched.
+
+Set ``HERMES_DETERMINISTIC_EMPTY_GUARD=0`` to disable both guards.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from dataclasses import dataclass
+from decimal import Decimal
+from typing import Any, List, Optional
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_EMPTY_RETRY_BUDGET = 3
+REDUCED_EMPTY_RETRY_BUDGET = 1
+DEFAULT_COST_THRESHOLD_USD = Decimal("0.25")
+
+# Attribute names stashed on the agent object. State is scoped to one
+# consecutive empty streak: it is cleared whenever a streak starts
+# (``_empty_content_retries == 0`` at record time), which transparently
+# honours every existing reset site (turn start, compaction, tool
+# success, fallback activation) without touching them.
+_ATTEMPTS_ATTR = "_empty_attempt_history"
+_STREAK_COST_ATTR = "_empty_streak_cost_usd"
+
+
+@dataclass(frozen=True)
+class EmptyAttempt:
+    """One observed empty completion within the current streak."""
+
+    model: str
+    provider: str
+    finish_reason: str
+    usage_present: bool
+    zero_output: bool
+
+    @property
+    def signature(self) -> tuple:
+        return (self.model, self.provider, self.finish_reason)
+
+
+def guard_enabled() -> bool:
+    return os.environ.get("HERMES_DETERMINISTIC_EMPTY_GUARD", "1").strip().lower() not in (
+        "0",
+        "false",
+        "no",
+        "off",
+    )
+
+
+def _cost_threshold_usd() -> Decimal:
+    raw = os.environ.get("HERMES_EMPTY_RETRY_COST_THRESHOLD_USD", "").strip()
+    if not raw:
+        return DEFAULT_COST_THRESHOLD_USD
+    try:
+        value = Decimal(raw)
+    except Exception:
+        return DEFAULT_COST_THRESHOLD_USD
+    if value <= 0:
+        return DEFAULT_COST_THRESHOLD_USD
+    return value
+
+
+def _attempts(agent: Any) -> List[EmptyAttempt]:
+    attempts = getattr(agent, _ATTEMPTS_ATTR, None)
+    if attempts is None:
+        attempts = []
+        setattr(agent, _ATTEMPTS_ATTR, attempts)
+    return attempts
+
+
+def _estimate_attempt_cost(agent: Any, response: Any) -> Optional[Decimal]:
+    """Best-effort USD estimate for one attempt. None when unknown."""
+    raw_usage = getattr(response, "usage", None)
+    if not raw_usage:
+        return None
+    try:
+        from agent.usage_pricing import estimate_usage_cost, normalize_usage
+
+        canonical = normalize_usage(
+            raw_usage,
+            provider=getattr(agent, "provider", None),
+            api_mode=getattr(agent, "api_mode", None),
+        )
+        result = estimate_usage_cost(
+            getattr(agent, "model", "") or "",
+            canonical,
+            provider=getattr(agent, "provider", None),
+            base_url=getattr(agent, "base_url", None),
+            api_key=getattr(agent, "api_key", None),
+        )
+    except Exception:  # noqa: BLE001 — pricing must never break the loop
+        logger.debug("empty-guard: cost estimation failed", exc_info=True)
+        return None
+    return getattr(result, "amount_usd", None)
+
+
+def _zero_output(agent: Any, response: Any) -> tuple:
+    """Return (usage_present, zero_output) for a response, failing open."""
+    raw_usage = getattr(response, "usage", None)
+    if not raw_usage:
+        return (False, False)
+    try:
+        from agent.usage_pricing import normalize_usage
+
+        canonical = normalize_usage(
+            raw_usage,
+            provider=getattr(agent, "provider", None),
+            api_mode=getattr(agent, "api_mode", None),
+        )
+    except Exception:  # noqa: BLE001
+        logger.debug("empty-guard: usage normalization failed", exc_info=True)
+        return (False, False)
+    output = getattr(canonical, "output_tokens", None)
+    if output is None:
+        return (False, False)
+    # A present-but-empty usage object (some proxies emit usage with no
+    # fields) normalizes to all zeros. A genuine completion always has
+    # input tokens — without them the usage is not evidence, fail open.
+    if getattr(canonical, "prompt_tokens", 0) <= 0:
+        return (False, False)
+    # Reasoning tokens count as real generation — a reasoning-only
+    # response is NOT a deterministic empty (the prefill-continuation
+    # path upstream owns that case).
+    reasoning = getattr(canonical, "reasoning_tokens", 0) or 0
+    return (True, (output + reasoning) == 0)
+
+
+def record_empty_attempt(agent: Any, *, finish_reason: str, response: Any) -> None:
+    """Record one empty completion in the current streak.
+
+    Must be called before ``_empty_content_retries`` is incremented for
+    this attempt: a counter of 0 marks the start of a new streak and
+    clears prior history (this transparently follows every existing
+    counter-reset site).
+    """
+    attempts = _attempts(agent)
+    if getattr(agent, "_empty_content_retries", 0) == 0:
+        attempts.clear()
+        setattr(agent, _STREAK_COST_ATTR, Decimal("0"))
+
+    usage_present, zero_output = _zero_output(agent, response)
+    attempts.append(
+        EmptyAttempt(
+            model=str(getattr(agent, "model", "") or ""),
+            provider=str(getattr(agent, "provider", "") or ""),
+            finish_reason=str(finish_reason or ""),
+            usage_present=usage_present,
+            zero_output=zero_output,
+        )
+    )
+
+    cost = _estimate_attempt_cost(agent, response)
+    if cost is not None and cost > 0:
+        prior = getattr(agent, _STREAK_COST_ATTR, Decimal("0")) or Decimal("0")
+        setattr(agent, _STREAK_COST_ATTR, prior + cost)
+
+
+def deterministic_empty(agent: Any) -> bool:
+    """True when the current streak looks deterministic.
+
+    Requires >= 2 consecutive attempts, ALL with usage present, zero
+    output tokens, and an identical (model, provider, finish_reason)
+    signature. Any attempt with missing usage or non-zero output keeps
+    this False (fail open — transients deserve their retries).
+    """
+    if not guard_enabled():
+        return False
+    attempts = getattr(agent, _ATTEMPTS_ATTR, None) or []
+    if len(attempts) < 2:
+        return False
+    first = attempts[0]
+    return all(
+        a.usage_present and a.zero_output and a.signature == first.signature
+        for a in attempts
+    )
+
+
+def empty_retry_budget(agent: Any, response: Any) -> int:
+    """Empty-retry budget for the current streak (3, or 1 when a single
+    attempt is estimated to cost more than the configured threshold)."""
+    if not guard_enabled():
+        return DEFAULT_EMPTY_RETRY_BUDGET
+    cost = _estimate_attempt_cost(agent, response)
+    if cost is None:
+        return DEFAULT_EMPTY_RETRY_BUDGET
+    if cost >= _cost_threshold_usd():
+        return REDUCED_EMPTY_RETRY_BUDGET
+    return DEFAULT_EMPTY_RETRY_BUDGET
+
+
+def streak_cost_usd(agent: Any) -> Optional[Decimal]:
+    """Accumulated estimated cost of the current empty streak, if known."""
+    cost = getattr(agent, _STREAK_COST_ATTR, None)
+    if cost is None or cost <= 0:
+        return None
+    return cost

tests/agent/test_empty_response_guard.py

@@ -0,0 +1,280 @@
+"""Tests for agent/empty_response_guard.py (NS-503).
+
+The guard exists to stop the empty-retry loop re-billing large inputs for
+deterministic empties (unsignaled provider refusals with zero output
+tokens) while never tightening behaviour on ambiguous evidence.
+
+Fail-open contract under test:
+- Missing usage -> never deterministic, default budget.
+- Any generated tokens (output or reasoning) -> never deterministic.
+- Different model/provider/finish_reason across attempts -> not deterministic.
+- Guard disabled via env -> everything falls back to defaults.
+"""
+
+from decimal import Decimal
+from types import SimpleNamespace
+
+import pytest
+
+from agent import empty_response_guard as guard
+
+
+def _agent(**overrides):
+    base = dict(
+        model="anthropic/claude-fable-5",
+        provider="nous",
+        api_mode="chat_completions",
+        base_url=None,
+        api_key=None,
+        _empty_content_retries=0,
+    )
+    base.update(overrides)
+    return SimpleNamespace(**base)
+
+
+def _response(prompt_tokens=25_900, completion_tokens=0, usage_present=True):
+    if not usage_present:
+        return SimpleNamespace(usage=None)
+    usage = SimpleNamespace(
+        prompt_tokens=prompt_tokens,
+        completion_tokens=completion_tokens,
+        total_tokens=prompt_tokens + completion_tokens,
+    )
+    return SimpleNamespace(usage=usage)
+
+
+def _record_streak(agent, responses, finish_reasons=None):
+    """Record attempts the way the loop does: record, then increment."""
+    finish_reasons = finish_reasons or ["stop"] * len(responses)
+    for resp, reason in zip(responses, finish_reasons):
+        guard.record_empty_attempt(agent, finish_reason=reason, response=resp)
+        agent._empty_content_retries += 1
+
+
+class TestDeterministicEmpty:
+    def test_two_zero_output_attempts_same_signature_is_deterministic(self):
+        agent = _agent()
+        _record_streak(agent, [_response(), _response()])
+        assert guard.deterministic_empty(agent) is True
+
+    def test_single_attempt_is_never_deterministic(self):
+        """One empty could be a transient blip — retry #1 must always run."""
+        agent = _agent()
+        _record_streak(agent, [_response()])
+        assert guard.deterministic_empty(agent) is False
+
+    def test_missing_usage_fails_open(self):
+        agent = _agent()
+        _record_streak(
+            agent,
+            [_response(usage_present=False), _response(usage_present=False)],
+        )
+        assert guard.deterministic_empty(agent) is False
+
+    def test_mixed_usage_presence_fails_open(self):
+        agent = _agent()
+        _record_streak(agent, [_response(), _response(usage_present=False)])
+        assert guard.deterministic_empty(agent) is False
+
+    def test_nonzero_output_tokens_fails_open(self):
+        """Model generated something (whitespace, stripped think-blocks) —
+        that's the flaky-but-recoverable class, keep full retries."""
+        agent = _agent()
+        _record_streak(
+            agent,
+            [_response(completion_tokens=42), _response(completion_tokens=42)],
+        )
+        assert guard.deterministic_empty(agent) is False
+
+    def test_zero_then_nonzero_fails_open(self):
+        agent = _agent()
+        _record_streak(
+            agent,
+            [_response(completion_tokens=0), _response(completion_tokens=7)],
+        )
+        assert guard.deterministic_empty(agent) is False
+
+    def test_signature_change_resets_determinism(self):
+        """Fallback switched model mid-streak — new model deserves retries."""
+        agent = _agent()
+        guard.record_empty_attempt(agent, finish_reason="stop", response=_response())
+        agent._empty_content_retries += 1
+        agent.model = "other/model"
+        guard.record_empty_attempt(agent, finish_reason="stop", response=_response())
+        agent._empty_content_retries += 1
+        assert guard.deterministic_empty(agent) is False
+
+    def test_finish_reason_change_fails_open(self):
+        agent = _agent()
+        _record_streak(
+            agent,
+            [_response(), _response()],
+            finish_reasons=["stop", "length"],
+        )
+        assert guard.deterministic_empty(agent) is False
+
+    def test_new_streak_clears_history(self):
+        """Counter reset to 0 (turn start / tool success / compaction /
+        fallback) starts a fresh streak — prior attempts must not leak."""
+        agent = _agent()
+        _record_streak(agent, [_response(), _response()])
+        assert guard.deterministic_empty(agent) is True
+
+        agent._empty_content_retries = 0  # any existing reset site
+        guard.record_empty_attempt(agent, finish_reason="stop", response=_response())
+        agent._empty_content_retries += 1
+        assert guard.deterministic_empty(agent) is False
+
+    def test_guard_disabled_via_env(self, monkeypatch):
+        monkeypatch.setenv("HERMES_DETERMINISTIC_EMPTY_GUARD", "0")
+        agent = _agent()
+        _record_streak(agent, [_response(), _response()])
+        assert guard.deterministic_empty(agent) is False
+
+    def test_reasoning_tokens_count_as_generation(self, monkeypatch):
+        """Reasoning-only responses are owned by the prefill path; the
+        guard must not classify them as deterministic empties."""
+        canonical = SimpleNamespace(output_tokens=0, reasoning_tokens=128)
+        monkeypatch.setattr(
+            guard, "_zero_output", lambda a, r: (True, (0 + 128) == 0)
+        )
+        agent = _agent()
+        _record_streak(agent, [_response(), _response()])
+        assert guard.deterministic_empty(agent) is False
+
+
+class TestEmptyRetryBudget:
+    def test_default_budget_when_cost_unknown(self, monkeypatch):
+        monkeypatch.setattr(guard, "_estimate_attempt_cost", lambda a, r: None)
+        assert (
+            guard.empty_retry_budget(_agent(), _response())
+            == guard.DEFAULT_EMPTY_RETRY_BUDGET
+        )
+
+    def test_reduced_budget_above_threshold(self, monkeypatch):
+        monkeypatch.setattr(
+            guard, "_estimate_attempt_cost", lambda a, r: Decimal("0.80")
+        )
+        assert (
+            guard.empty_retry_budget(_agent(), _response())
+            == guard.REDUCED_EMPTY_RETRY_BUDGET
+        )
+
+    def test_default_budget_below_threshold(self, monkeypatch):
+        monkeypatch.setattr(
+            guard, "_estimate_attempt_cost", lambda a, r: Decimal("0.01")
+        )
+        assert (
+            guard.empty_retry_budget(_agent(), _response())
+            == guard.DEFAULT_EMPTY_RETRY_BUDGET
+        )
+
+    def test_custom_threshold_env(self, monkeypatch):
+        monkeypatch.setenv("HERMES_EMPTY_RETRY_COST_THRESHOLD_USD", "5.00")
+        monkeypatch.setattr(
+            guard, "_estimate_attempt_cost", lambda a, r: Decimal("0.80")
+        )
+        assert (
+            guard.empty_retry_budget(_agent(), _response())
+            == guard.DEFAULT_EMPTY_RETRY_BUDGET
+        )
+
+    def test_bad_threshold_env_falls_back(self, monkeypatch):
+        monkeypatch.setenv("HERMES_EMPTY_RETRY_COST_THRESHOLD_USD", "banana")
+        assert guard._cost_threshold_usd() == guard.DEFAULT_COST_THRESHOLD_USD
+        monkeypatch.setenv("HERMES_EMPTY_RETRY_COST_THRESHOLD_USD", "-1")
+        assert guard._cost_threshold_usd() == guard.DEFAULT_COST_THRESHOLD_USD
+
+    def test_guard_disabled_keeps_default_budget(self, monkeypatch):
+        monkeypatch.setenv("HERMES_DETERMINISTIC_EMPTY_GUARD", "0")
+        monkeypatch.setattr(
+            guard, "_estimate_attempt_cost", lambda a, r: Decimal("9.99")
+        )
+        assert (
+            guard.empty_retry_budget(_agent(), _response())
+            == guard.DEFAULT_EMPTY_RETRY_BUDGET
+        )
+
+    def test_pricing_exception_fails_open(self, monkeypatch):
+        def _boom(a, r):
+            raise RuntimeError("pricing exploded")
+
+        # _estimate_attempt_cost itself catches internally; simulate at
+        # the normalize layer by feeding garbage usage.
+        agent = _agent(model=None, provider=None)
+        resp = SimpleNamespace(usage=object())
+        assert (
+            guard.empty_retry_budget(agent, resp)
+            == guard.DEFAULT_EMPTY_RETRY_BUDGET
+        )
+
+
+class TestStreakCost:
+    def test_streak_cost_accumulates(self, monkeypatch):
+        costs = iter([Decimal("1.10"), Decimal("1.23")])
+        monkeypatch.setattr(
+            guard, "_estimate_attempt_cost", lambda a, r: next(costs)
+        )
+        agent = _agent()
+        _record_streak(agent, [_response(), _response()])
+        assert guard.streak_cost_usd(agent) == Decimal("2.33")
+
+    def test_streak_cost_none_when_unknown(self, monkeypatch):
+        monkeypatch.setattr(guard, "_estimate_attempt_cost", lambda a, r: None)
+        agent = _agent()
+        _record_streak(agent, [_response(), _response()])
+        assert guard.streak_cost_usd(agent) is None
+
+    def test_streak_cost_resets_on_new_streak(self, monkeypatch):
+        monkeypatch.setattr(
+            guard, "_estimate_attempt_cost", lambda a, r: Decimal("1.00")
+        )
+        agent = _agent()
+        _record_streak(agent, [_response(), _response()])
+        assert guard.streak_cost_usd(agent) == Decimal("2.00")
+        agent._empty_content_retries = 0
+        guard.record_empty_attempt(agent, finish_reason="stop", response=_response())
+        assert guard.streak_cost_usd(agent) == Decimal("1.00")
+
+
+class TestZeroOutputExtraction:
+    """_zero_output goes through the real normalize_usage path."""
+
+    def test_openai_shape_zero_completion(self):
+        agent = _agent()
+        present, zero = guard._zero_output(agent, _response(completion_tokens=0))
+        assert present is True
+        assert zero is True
+
+    def test_openai_shape_with_completion(self):
+        agent = _agent()
+        present, zero = guard._zero_output(agent, _response(completion_tokens=9))
+        assert present is True
+        assert zero is False
+
+    def test_no_usage(self):
+        agent = _agent()
+        present, zero = guard._zero_output(agent, _response(usage_present=False))
+        assert present is False
+        assert zero is False
+
+    def test_anthropic_shape_zero_output(self):
+        agent = _agent(api_mode="anthropic_messages")
+        usage = SimpleNamespace(
+            input_tokens=25_900,
+            output_tokens=0,
+            cache_read_input_tokens=0,
+            cache_creation_input_tokens=0,
+        )
+        present, zero = guard._zero_output(agent, SimpleNamespace(usage=usage))
+        assert present is True
+        assert zero is True
+
+    def test_all_zero_usage_object_fails_open(self):
+        """Proxies that emit an empty usage object (all fields absent →
+        normalized to zeros) provide no evidence — must not classify."""
+        agent = _agent()
+        usage = SimpleNamespace()  # no token fields at all
+        present, zero = guard._zero_output(agent, SimpleNamespace(usage=usage))
+        assert present is False
+        assert zero is False

tests/run_agent/test_run_agent.py

@@ -3264,6 +3264,52 @@ class TestRunConversation:
         assert "No reply:" in result["final_response"]
         assert result["api_calls"] == 4  # 1 original + 3 retries
 
+    def test_deterministic_empty_stops_retries_early(self, agent):
+        """NS-503: consecutive zero-output-token empties with identical
+        model/provider/finish_reason are deterministic (unsignaled refusal)
+        — the loop must stop re-billing the full input after the second
+        attempt instead of burning the whole retry budget."""
+        self._setup_agent(agent)
+        agent.base_url = "http://127.0.0.1:1234/v1"
+        zero_usage = {
+            "prompt_tokens": 25_900,
+            "completion_tokens": 0,
+            "total_tokens": 25_900,
+        }
+        empty_resp = _mock_response(
+            content=None, finish_reason="stop", usage=zero_usage
+        )
+        # Provide plenty of responses; guard should stop consuming early.
+        agent.client.chat.completions.create.side_effect = [empty_resp] * 6
+        with (
+            patch.object(agent, "_persist_session"),
+            patch.object(agent, "_save_trajectory"),
+            patch.object(agent, "_cleanup_task_resources"),
+        ):
+            result = agent.run_conversation("answer me")
+        assert result["completed"] is True
+        assert result["final_response"] != "(empty)"
+        # 1 original + 1 retry: the second identical zero-output empty
+        # proves determinism, remaining retries are skipped.
+        assert result["api_calls"] == 2
+
+    def test_empty_without_usage_keeps_full_retry_budget(self, agent):
+        """NS-503 fail-open: no usage data means no evidence of a
+        deterministic empty — legacy 3-retry behaviour must be preserved
+        (this is the flaky-provider case retries exist for)."""
+        self._setup_agent(agent)
+        agent.base_url = "http://127.0.0.1:1234/v1"
+        empty_resp = _mock_response(content=None, finish_reason="stop")
+        agent.client.chat.completions.create.side_effect = [empty_resp] * 4
+        with (
+            patch.object(agent, "_persist_session"),
+            patch.object(agent, "_save_trajectory"),
+            patch.object(agent, "_cleanup_task_resources"),
+        ):
+            result = agent.run_conversation("answer me")
+        assert result["completed"] is True
+        assert result["api_calls"] == 4  # unchanged: 1 original + 3 retries
+
     def test_truly_empty_response_succeeds_on_nudge(self, agent):
         """Model produces content after being nudged for empty response."""
         self._setup_agent(agent)