fix(analyze): a budget-exhausted empty completion retries once at a raised, SDK-safe cap (#569, choice c)

libs/openant-core/core/analysis_core.py

@@ -309,7 +309,8 @@ def analyze_unit(
     use_multifile: bool = False,
     json_corrector: JSONCorrector = None,
     context_reviewer: ContextReviewer = None,
-    app_context: "ApplicationContext" = None
+    app_context: "ApplicationContext" = None,
+    max_tokens: int | None = None
 ) -> dict:
     """
     Analyze a single code unit.
@@ -395,7 +396,12 @@ def analyze_unit(
     # Call the configured analyze-phase model with the threat-model system prompt.
     start_time = datetime.now()
     system_prompt = get_stage1_system_prompt(app_context=app_context)
-    response = simple_text(binding, prompt, system=system_prompt)
+    # #569 (choice c): the budget-retry path passes a raised cap — the
+    # deterministic length-empty class gets one retry that attacks the
+    # cause (the budget) instead of a same-cap coin flip.
+    response = simple_text(binding, prompt, system=system_prompt,
+                           max_tokens=max_tokens) if max_tokens else \
+        simple_text(binding, prompt, system=system_prompt)
     elapsed = (datetime.now() - start_time).total_seconds()
 
     # Parse response

libs/openant-core/core/analyzer.py

@@ -36,7 +36,25 @@
 )
 from utilities.file_io import read_json, write_json
 from utilities.json_corrector import JSONCorrector
-from utilities.rate_limiter import get_rate_limiter, is_retryable_error
+from utilities.llm import DEFAULT_MAX_TOKENS
+from utilities.rate_limiter import (
+    get_rate_limiter,
+    is_budget_exhausted_error,
+    is_retryable_error,
+)
+
+# #569 (choice c): the budget-retry cap — raised above the analyze
+# default but under the Anthropic non-streaming ceiling (the SDK rejects
+# >~21,333 with "Streaming is required" — see helpers.py:29-31; the
+# adapters call non-streaming). The deterministic length-empty class gets
+# ONE retry that attacks the cause (the budget) within that envelope.
+BUDGET_RETRY_MAX_TOKENS = min(DEFAULT_MAX_TOKENS * 2, 21000)
+
+def budget_retry_cap(index: int, budget_set: set) -> int | None:
+    """#569: the per-index retry cap — the raised cap for the budget class,
+    None (the unchanged default) for every other retryable."""
+    return BUDGET_RETRY_MAX_TOKENS if index in budget_set else None
+
 
 # These live in core/ because core is shipped and experiment.py is not: importing
 # them from the research harness made `import core.analyzer` fail in any installed
@@ -119,7 +137,8 @@ def _apply_limit(units, limit):
     return prioritized[:limit]
 
 
-def _process_unit(binding: PhaseBinding, unit, index, json_corrector, app_context):
+def _process_unit(binding: PhaseBinding, unit, index, json_corrector, app_context,
+                  max_tokens=None):
     """Process a single unit for Stage 1 detection.
 
     Returns a dict with all result data. Does not mutate shared state.
@@ -135,6 +154,7 @@ def _process_unit(binding: PhaseBinding, unit, index, json_corrector, app_contex
             use_multifile=True,
             json_corrector=json_corrector,
             app_context=app_context,
+            max_tokens=max_tokens,
         )
 
         # Ensure unit_id is always present
@@ -752,6 +772,16 @@ def _summary_callback(finding, usage=None):
         i for i, r in enumerate(results)
         if r and is_retryable_error(r.get("error"))
     ]
+    # #569 (choice c): the deterministic budget-exhaustion empties (the
+    # length-stop class #561 named) retry ONCE at a RAISED cap — a same-cap
+    # re-roll of a budget exhaustion is a coin flip; the raised cap attacks
+    # the cause. Every other retryable (the filtered/malformed empties, the
+    # transient network class) keeps the #292 same-cap rationale.
+    budget_retry_indices = [
+        i for i in retryable_indices
+        if is_budget_exhausted_error(results[i].get("error"))
+    ]
+    _budget_set = set(budget_retry_indices)
     if retryable_indices:
         rate_limiter = get_rate_limiter()
         backoff = rate_limiter.time_until_ready()
@@ -760,13 +790,18 @@ def _summary_callback(finding, usage=None):
                   f"(waiting {backoff:.0f}s for rate limit to clear)...", file=sys.stderr)
             rate_limiter.wait_if_needed()
         else:
-            print(f"[Analyze] Retrying {len(retryable_indices)} failed units (transient errors)...",
-                  file=sys.stderr)
+            _extra = (f" ({len(budget_retry_indices)} at a raised output "
+                      f"cap {BUDGET_RETRY_MAX_TOKENS} — budget exhaustion)"
+                      if budget_retry_indices else "")
+            print(f"[Analyze] Retrying {len(retryable_indices)} failed units "
+                  f"(transient errors){_extra}...", file=sys.stderr)
 
         # Retry sequentially to avoid re-triggering rate limit
         for i in retryable_indices:
             unit = units[i]
-            out = _process_unit(binding, unit, i, json_corrector, app_context)
+            _cap = budget_retry_cap(i, _budget_set)
+            out = _process_unit(binding, unit, i, json_corrector, app_context,
+                                max_tokens=_cap)
             results[i] = out["result"]
             code_by_route[out["route_key"]] = out["code_for_route"]
 

libs/openant-core/tests/test_issue569_budget_retry.py

@@ -0,0 +1,172 @@
+"""Tests for issue #569 (choice c) — a budget-exhausted empty completion
+retries ONCE at a raised cap; every other retryable keeps the #292
+same-cap re-roll.
+
+The #561 cause-clause split named the deterministic class (finish_reason
+'length' / stop_reason 'max_tokens' → "the output budget was consumed");
+the #292 classifier still retried it at the SAME cap — a coin flip against
+a deterministic cause. Choice (c): the retry attacks the cause (the cap).
+"""
+from __future__ import annotations
+
+from utilities.rate_limiter import (
+    is_budget_exhausted_error,
+    is_retryable_error,
+)
+
+
+MSG_LENGTH = ("OpenAIAdapter returned an empty completion (no text or tool "
+              "calls; finish_reason='length'); the output budget was consumed "
+              "before any visible content (reasoning models spend it on "
+              "hidden reasoning)")
+MSG_STOP = ("OpenAIAdapter returned an empty completion (no text or tool "
+            "calls; finish_reason='stop'); the request may have been "
+            "filtered or the response was malformed")
+
+
+class TestBudgetDiscriminator:
+    def test_length_empty_is_budget_exhausted(self):
+        assert is_budget_exhausted_error(MSG_LENGTH)
+
+    def test_stop_empty_is_not_budget(self):
+        """The filtered/malformed class keeps the #292 same-cap retry."""
+        assert not is_budget_exhausted_error(MSG_STOP)
+
+    def test_both_still_retryable(self):
+        """The split does NOT declassify anything — both empty classes
+        stay retryable (the caller just raises the cap for one)."""
+        assert is_retryable_error(MSG_LENGTH)
+        assert is_retryable_error(MSG_STOP)
+
+    def test_dict_shape_via_message(self):
+        d = {"error": MSG_LENGTH}
+        assert is_budget_exhausted_error(d)
+
+    def test_transient_not_budget(self):
+        assert not is_budget_exhausted_error("connection reset by peer")
+
+
+class TestCapThreaded:
+    def test_analyze_unit_accepts_max_tokens(self):
+        """The cap parameter exists on the full chain (the plumbing)."""
+        import inspect
+        from core.analysis_core import analyze_unit
+        sig = inspect.signature(analyze_unit)
+        assert "max_tokens" in sig.parameters
+
+    def test_budget_retry_cap_under_sdk_ceiling(self):
+        """The refutation round's blocker: the raised cap MUST stay under
+        the Anthropic non-streaming ceiling (~21,333; helpers.py:29-31)
+        or the retry is a guaranteed SDK ValueError on that adapter."""
+        from core.analyzer import BUDGET_RETRY_MAX_TOKENS
+        from utilities.llm.helpers import DEFAULT_MAX_TOKENS
+        assert DEFAULT_MAX_TOKENS < BUDGET_RETRY_MAX_TOKENS <= 21000, (
+            f"the raised cap {BUDGET_RETRY_MAX_TOKENS} must be above the "
+            f"default and at/below the SDK-safe ceiling")
+
+    def test_retry_cap_decision_production_helper(self):
+        """THE production decision, pinned on the real helper: the budget
+        class gets the raised cap, every other retryable gets None."""
+        from core.analyzer import budget_retry_cap, BUDGET_RETRY_MAX_TOKENS
+        assert budget_retry_cap(0, {0, 2}) == BUDGET_RETRY_MAX_TOKENS
+        assert budget_retry_cap(1, {0, 2}) is None  # the #292 same-cap path
+        assert budget_retry_cap(2, {0, 2}) == BUDGET_RETRY_MAX_TOKENS
+
+    def test_retry_loop_drives_the_split(self, tmp_path, monkeypatch):
+        """End-to-end through the REAL run_analysis retry pass: a
+        length-empty unit's retry call carries the raised cap; a
+        stop-empty unit's carries None. Drives the production wiring
+        (analyzer's retry loop) — the #569 review round replaced this
+        test's prior shape, which re-implemented the loop's computation
+        and never invoked it."""
+        import json as _json
+        from core import analyzer
+        from utilities.llm_client import reset_warning_state
+
+        reset_warning_state()
+        dataset_path = tmp_path / "dataset.json"
+        dataset_path.write_text(_json.dumps({"units": [
+            {"id": "a:f1", "code": "x=1"},
+            {"id": "b:f2", "code": "x=1"},
+        ]}))
+        output_dir = tmp_path / "out"
+
+        def fake_run_detection(units, binding, json_corrector, app_context,
+                               workers, checkpoint=None,
+                               summary_callback=None):
+            # Two failed units: one budget-class, one stop-class.
+            return ([{"unit_id": "a:f1", "error": MSG_LENGTH},
+                     {"unit_id": "b:f2", "error": MSG_STOP}],
+                    {u["id"]: "" for u in units})
+
+        calls = []
+
+        def fake_process(binding, unit, i, jc, ac, max_tokens=None):
+            calls.append((unit["id"], max_tokens))
+            return {"result": {"unit_id": unit["id"], "finding": "safe",
+                               "verdict": "SAFE", "confidence": 90,
+                               "vulnerabilities": [], "reasoning": "r"},
+                    "route_key": unit["id"], "code_for_route": "",
+                    "finding": "safe", "usage": {}}
+
+        monkeypatch.setattr(analyzer, "_run_detection", fake_run_detection)
+        monkeypatch.setattr(analyzer, "_analyze_fingerprint",
+                            lambda binding, ctx_sha=None: {
+                                "key_digest": "sha256:test"})
+        monkeypatch.setattr(analyzer, "_process_unit", fake_process)
+
+        from utilities.llm import PhaseBinding
+
+        class _Adapter:
+            name = "anthropic"
+            supports_tools = True
+            pricing = {}
+
+        class _FakeRegistry:
+            def get(self, phase):
+                return PhaseBinding(phase=phase, adapter=_Adapter(),
+                                    model="m", provider_name="anthropic")
+
+        analyzer.run_analysis(
+            str(dataset_path), str(output_dir),
+            registry=_FakeRegistry(), workers=1)
+        reset_warning_state()
+
+        assert calls == [("a:f1", analyzer.BUDGET_RETRY_MAX_TOKENS),
+                         ("b:f2", None)], (
+            "the retry pass must thread the raised cap ONLY into the "
+            "budget-class unit's call")
+
+
+class TestParityMarkers:
+    """The #569 refutation's parity extension: the discriminator reaches
+    EVERY adapter's budget wording — which the #569 review round made
+    DETERMINISTIC-CLASS-ONLY (the truncated wordings carry the marker; the
+    filtered wordings must NOT — gemini/Responses producers branch on the
+    finish signal, mirroring the openai-chat #561 pattern)."""
+
+    def test_gemini_budget_wording(self):
+        assert is_budget_exhausted_error(
+            "Gemini returned a candidate with no usable content (empty "
+            "completion); the response was truncated — a thinking "
+            "model consumed the token budget before emitting output")
+
+    def test_gemini_filtered_not_budget(self):
+        assert not is_budget_exhausted_error(
+            "Gemini returned a candidate with no usable content (empty "
+            "completion); the response may have been filtered or malformed")
+
+    def test_openai_responses_budget_wording(self):
+        assert is_budget_exhausted_error(
+            "OpenAI Responses returned no usable content "
+            "(status='incomplete'); the request was truncated — "
+            "reasoning consumed the budget")
+
+    def test_openai_responses_filtered_not_budget(self):
+        assert not is_budget_exhausted_error(
+            "OpenAI Responses returned no usable content "
+            "(status='completed'); the request may have been filtered")
+
+    def test_filtered_still_not_budget(self):
+        assert not is_budget_exhausted_error(
+            "may have been filtered or the response was malformed")

libs/openant-core/tests/test_llm_google_adapter.py

@@ -101,6 +101,36 @@ def test_present_candidate_with_empty_parts_raises_not_clean_end_turn():
         _response_to_unified(resp)
 
 
+def test_empty_max_tokens_candidate_raises_budget_wording():
+    # #569: the budget marker is DETERMINISTIC-CLASS-ONLY — a MAX_TOKENS
+    # empty candidate carries "consumed the token budget" (the raised-cap
+    # retry class); the other empty shapes must NOT (producer-side pin —
+    # the #569 review round: the classifier must not over-match).
+    from types import SimpleNamespace
+    from utilities.llm import LLMResponseError
+    from utilities.llm.providers.google import _response_to_unified
+    cand = SimpleNamespace(finish_reason="MAX_TOKENS", content=SimpleNamespace(parts=[]))
+    resp = SimpleNamespace(candidates=[cand], usage_metadata=SimpleNamespace(
+        prompt_token_count=1, candidates_token_count=0, total_token_count=1))
+    with pytest.raises(LLMResponseError, match="consumed the token budget"):
+        _response_to_unified(resp)
+
+
+def test_empty_stop_candidate_has_no_budget_wording():
+    # #569 producer-side pin: a filtered/blank empty candidate keeps the
+    # #292 same-cap rationale — its raise must NOT carry the marker.
+    from types import SimpleNamespace
+    from utilities.llm import LLMResponseError
+    from utilities.llm.providers.google import _response_to_unified
+    cand = SimpleNamespace(finish_reason="STOP", content=SimpleNamespace(parts=[]))
+    resp = SimpleNamespace(candidates=[cand], usage_metadata=SimpleNamespace(
+        prompt_token_count=1, candidates_token_count=0, total_token_count=1))
+    with pytest.raises(LLMResponseError) as ei:
+        _response_to_unified(resp)
+    assert "consumed the token budget" not in str(ei.value)
+    assert "filtered" in str(ei.value)
+
+
 def test_tool_use_only_candidate_is_valid_not_empty():
     # Control: a function_call part with no text is a VALID response (content
     # non-empty) and must NOT be caught by the empty-content guard.

libs/openant-core/tests/test_llm_openai_adapter.py

@@ -474,6 +474,25 @@ def test_empty_output_raises_response_error():
         adapter.complete(model=G5, system=None, messages=_hi(), max_tokens=8)
 
 
+def test_empty_output_max_tokens_stop_carries_budget_wording():
+    # #569 producer-side pin: an empty output with a max_tokens stop (the
+    # incomplete/max_output_tokens shape) carries "reasoning consumed the
+    # budget" (the raised-cap retry class).
+    adapter, _ = _stub_resp(lambda **kw: _resp(status="incomplete", incomplete_reason="max_output_tokens",
+                                               output=[_reasoning()]))
+    with pytest.raises(LLMResponseError, match="reasoning consumed the budget"):
+        adapter.complete(model=G5, system=None, messages=_hi(), max_tokens=8)
+
+
+def test_empty_output_filtered_shape_has_no_budget_wording():
+    # #569 producer-side pin: a filtered empty output keeps the #292
+    # same-cap rationale — its raise must NOT carry the budget marker.
+    adapter, _ = _stub_resp(lambda **kw: _resp(output=[_reasoning()]))
+    with pytest.raises(LLMResponseError) as ei:
+        adapter.complete(model=G5, system=None, messages=_hi(), max_tokens=8)
+    assert "consumed the budget" not in str(ei.value)
+
+
 def test_failed_status_raises_response_error():
     adapter, _ = _stub_resp(lambda **kw: _resp(status="failed",
                                                error=SimpleNamespace(message="boom"), output=[]))

libs/openant-core/tests/test_retry_empty_completion.py

@@ -30,12 +30,14 @@
                     "the request may have been filtered or the response was malformed")
 _EMPTY_GEMINI = ("Gemini returned a candidate with no usable content (empty completion); "
                  "the response may have been truncated (a thinking block)")
-# OpenAI Responses-API empty path (openai.py:730): says "no usable content" but
-# NOT "empty completion" — the term must match the shared "no usable content"
-# phrase so this sibling is covered too.
+# OpenAI Responses-API empty path: says "no usable content" but NOT
+# "empty completion" — the term must match the shared "no usable content"
+# phrase so this sibling is covered too. Both post-#569 producer arms (the
+# truncated budget class and the filtered class) stay retryable.
 _EMPTY_OPENAI_RESPONSES = ("OpenAI Responses returned no usable content (status='incomplete'); "
-                           "the request may have been truncated (reasoning consumed the budget) "
-                           "or filtered")
+                           "the request was truncated — reasoning consumed the budget")
+_EMPTY_OPENAI_RESPONSES_FILTERED = ("OpenAI Responses returned no usable content (status='completed'); "
+                                     "the request may have been filtered")
 # OpenAI Chat-Completions empty paths (openai.py:760, :816): say "empty
 # completion" but NOT "no usable content" — so the term set must include the
 # "empty completion" phrase too, or these (and OpenRouter, which reuses the

libs/openant-core/utilities/llm/providers/google.py

@@ -500,11 +500,19 @@ def _response_to_unified(response: Any) -> CompletionResult:
     # VALID and not caught here because content_blocks is non-empty. Refusal is
     # the more specific signal and already raised above.
     if not content_blocks:
+        # #569: the budget wording is DETERMINISTIC-CLASS-ONLY — a
+        # MAX_TOKENS stop is the thinking-budget exhaustion (the raised-cap
+        # retry class); every other empty candidate (filtered/malformed)
+        # keeps the #292 same-cap rationale and must NOT carry the marker.
+        if raw_finish in ("MAX_TOKENS", "FinishReason.MAX_TOKENS"):
+            raise LLMResponseError(
+                "Gemini returned a candidate with no usable content (empty "
+                "completion); the response was truncated — a thinking "
+                "model consumed the token budget before emitting output"
+            )
         raise LLMResponseError(
             "Gemini returned a candidate with no usable content (empty "
-            "completion); the response may have been truncated (a thinking "
-            "model consumed the token budget before emitting output) or "
-            "filtered/malformed"
+            "completion); the response may have been filtered or malformed"
         )
 
     stop_reason: StopReason