fix(llm): close the truncated-response silent-FN family (adapters + enhancer + verifier) (#207)

Co-authored-by: gadievron <gadi@unpromptedcon.org>
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

Author: 3 people

libs/openant-core/context/repo_explorer.py

@@ -283,6 +283,25 @@ def explore_repository(
             if not isinstance(block, ToolUseBlock):
                 continue
             if block.name == finish_tool.name:
+                # R3-A: a finish call on a turn TRUNCATED at the token cap
+                # (stop_reason=="max_tokens") may carry partial arguments — accepting
+                # it writes an under-scoped application-context / threat-model doc that
+                # every later scan trusts (silent coverage loss). Don't accept it.
+                # R4-1: but the Messages API requires a tool_result for every tool_use,
+                # so ANSWER this finish's tool_use with a retry nudge (rather than
+                # skipping it unanswered, which 400s the next turn) — the loop then
+                # asks for a complete finish and, failing that, exhausts MAX_TURNS and
+                # raises (a visible failure, not a silent partial). Mirrors the
+                # verifier + enhancer max_tokens gate.
+                if getattr(response, "stop_reason", None) == "max_tokens":
+                    results.append(ToolResultBlock(
+                        tool_use_id=block.id, name=block.name,
+                        content=json.dumps({
+                            "error": "Your finish call was cut off at the token limit; "
+                                     "reply again with a complete but more concise finish call."
+                        }),
+                    ))
+                    continue
                 return dict(block.input or {}), budget
             outcome = explorer.execute(block.name, block.input or {})
             results.append(ToolResultBlock(

libs/openant-core/tests/test_agent_degenerate_exit.py

@@ -113,6 +113,35 @@ def record_call(self, **kw):
     assert result.input_tokens == 5000 and result.output_tokens == 200
 
 
+def _finish_block(classification):
+    # a COMPLETE, valid finish call (all required fields, valid classification)
+    return ToolUseBlock(id="t1", name="finish", input={
+        "include_functions": ["f"], "usage_context": "ctx",
+        "security_classification": classification,
+        "classification_reasoning": "r", "confidence": 0.5,
+    })
+
+
+def test_truncated_finish_at_max_tokens_is_incomplete_not_a_verdict():
+    """R2-B: a VALID finish call on a turn truncated at max_tokens must be
+    INCOMPLETE, not accepted as a complete classification — a truncated
+    finish(security_classification="neutral") would silently drop a unit."""
+    result = _run(_agent([
+        CompletionResult(content=[_finish_block("neutral")],
+                         input_tokens=1, output_tokens=1, stop_reason="max_tokens"),
+    ]))
+    assert result.security_classification == "incomplete"
+
+
+def test_complete_finish_still_accepted():
+    """Regression guard: a finish on a NORMAL (tool_use) turn is still accepted."""
+    result = _run(_agent([
+        CompletionResult(content=[_finish_block("security_control")],
+                         input_tokens=1, output_tokens=1, stop_reason="tool_use"),
+    ]))
+    assert result.security_classification == "security_control"
+
+
 def test_no_tool_calls_is_not_a_neutral_verdict():
     """Sibling path: model responded but made no tool calls."""
     result = _run(_agent([

libs/openant-core/tests/test_llm_anthropic_adapter.py

@@ -180,9 +180,13 @@ def test_tool_use_and_result_blocks_round_trip(self):
 
 
 class TestResponseTranslation:
-    def test_unknown_stop_reason_normalised_to_end_turn(self):
-        # Future SDK adding a new stop reason must not crash the
-        # pipeline. The adapter falls back to "end_turn" defensively.
+    def test_unknown_stop_reason_treated_as_max_tokens(self):
+        # R2-C: a future/unknown/proxy stop_reason must not read as a clean
+        # end_turn — for a security tool that masks a refusal/abnormal
+        # termination as a finished completion. The adapter defaults it to
+        # "max_tokens" (a not-a-clean-finish signal), mirroring the OpenAI
+        # adapter. Known values (end_turn/max_tokens/tool_use/stop_sequence)
+        # keep their explicit mapping.
         adapter, _ = _stub_adapter(
             lambda **kw: _ok_response(stop_reason="future_invention")
         )
@@ -192,7 +196,7 @@ def test_unknown_stop_reason_normalised_to_end_turn(self):
             messages=[Message(role="user", content=[TextBlock("hi")])],
             max_tokens=8,
         )
-        assert result.stop_reason == "end_turn"
+        assert result.stop_reason == "max_tokens"
 
     def test_tool_use_block_extracted_from_response(self):
         def respond(**kw):

libs/openant-core/tests/test_repo_explorer_loop.py

@@ -16,8 +16,9 @@
 
 
 class _Resp:
-    def __init__(self, content):
+    def __init__(self, content, stop_reason=None):
         self.content = content
+        self.stop_reason = stop_reason
 
 
 class _FakeAdapter:
@@ -32,6 +33,30 @@ def complete(self, *, model, system, messages, max_tokens, tools):
         return self._scripted.pop(0)
 
 
+class _PairingFakeAdapter(_FakeAdapter):
+    """Enforces the Messages-API rule real providers enforce: every assistant
+    tool_use must be answered by a tool_result in the immediately following user
+    turn. Catches an unanswered/dangling tool_use (which real APIs 400 on)."""
+
+    def complete(self, *, model, system, messages, max_tokens, tools):
+        for i, m in enumerate(messages):
+            if m.role != "assistant":
+                continue
+            tu_ids = [b.id for b in m.content if isinstance(b, ToolUseBlock)]
+            if not tu_ids:
+                continue
+            nxt = messages[i + 1] if i + 1 < len(messages) else None
+            tr_ids = [b.tool_use_id for b in (nxt.content if nxt else ())
+                      if isinstance(b, ToolResultBlock)]
+            for tid in tu_ids:
+                if tid not in tr_ids:
+                    raise RuntimeError(
+                        f"400: assistant tool_use {tid!r} has no matching tool_result "
+                        f"in the next user turn")
+        return super().complete(model=model, system=system, messages=messages,
+                                max_tokens=max_tokens, tools=tools)
+
+
 class _FakeBinding:
     def __init__(self, adapter):
         self.adapter = adapter
@@ -60,6 +85,24 @@ def test_finish_returns_payload_and_counts_turns(tmp_path):
     assert budget.turns == 1
 
 
+def test_truncated_finish_at_max_tokens_is_not_accepted(tmp_path):
+    # R3-A: a finish call on a turn truncated at max_tokens must NOT be accepted as
+    # a complete survey (it under-scopes the threat model every later scan trusts).
+    # R4-1: and the skipped finish's tool_use must be ANSWERED (pairing-validating
+    # adapter enforces the real Messages-API rule) so the retry turn isn't a 400.
+    # The loop nudges and uses a later COMPLETE finish instead.
+    adapter = _PairingFakeAdapter([
+        _Resp((ToolUseBlock(id="tu1", name="finish", input={"partial": True}),),
+              stop_reason="max_tokens"),
+        _Resp((ToolUseBlock(id="tu2", name="finish", input={"complete": True}),),
+              stop_reason="tool_use"),
+    ])
+    payload, budget = explore_repository(_repo(tmp_path), _FakeBinding(adapter),
+                                         "sys", "task", _FINISH)
+    assert payload == {"complete": True}   # the truncated finish was skipped
+    assert budget.turns == 2
+
+
 def test_chatty_turn_is_answered_with_plain_text_not_toolresult(tmp_path):
     # Turn 1: model returns prose, calls no tool -> the nudge branch.
     # Turn 2: model calls finish.

libs/openant-core/tests/test_truncation_silent_fn_family.py

@@ -0,0 +1,65 @@
+"""Truncation silent-FN family — cross-provider adapter invariants (R2-A, R2-C).
+
+Family invariant (adapter side): every adapter must emit stop_reason="max_tokens"
+for a TRUNCATED response and must not mask it (as tool_use) or launder an
+unknown/abnormal termination into a clean end_turn — otherwise a downstream
+consumer (verifier / enhancer) accepts a truncated reply as a complete verdict.
+Offline stubs; no network.
+"""
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from utilities.llm.providers.google import _response_to_unified as _google_unify
+
+
+def _gemini_resp(*, finish_reason, with_tool=False, text=None):
+    parts = []
+    if with_tool:
+        parts.append(SimpleNamespace(
+            function_call=SimpleNamespace(name="finish", args={"agree": False}, id=None), text=None))
+    if text is not None:
+        parts.append(SimpleNamespace(function_call=None, text=text))
+    return SimpleNamespace(
+        candidates=[SimpleNamespace(finish_reason=finish_reason,
+                                    content=SimpleNamespace(parts=parts))],
+        usage_metadata=SimpleNamespace(prompt_token_count=1, candidates_token_count=1),
+    )
+
+
+def test_gemini_truncation_wins_over_tool_use():
+    # R2-A: a MAX_TOKENS candidate carrying a function_call must surface as
+    # max_tokens (truncation), NOT tool_use — else the consumer accepts a
+    # truncated finish as a complete verdict.
+    r = _google_unify(_gemini_resp(finish_reason="MAX_TOKENS", with_tool=True))
+    assert r.stop_reason == "max_tokens"
+
+
+def test_gemini_normal_tool_call_still_tool_use():
+    # regression guard: a normal (STOP) tool call is still tool_use.
+    r = _google_unify(_gemini_resp(finish_reason="STOP", with_tool=True, text=None))
+    assert r.stop_reason == "tool_use"
+
+
+def test_gemini_unknown_finish_is_max_tokens_not_end_turn():
+    # R2-C: an unknown/abnormal finish_reason (SAFETY/RECITATION/proxy) is not a
+    # clean end_turn.
+    r = _google_unify(_gemini_resp(finish_reason="ZZ_FUTURE_REASON", text="partial"))
+    assert r.stop_reason == "max_tokens"
+
+
+def test_gemini_unknown_finish_with_tool_call_is_max_tokens_not_tool_use():
+    # round-5: an UNKNOWN/abnormal finish_reason carrying a function_call must surface
+    # as max_tokens, not tool_use — an abnormal termination wins over the tool-call
+    # signal (so a consumer's max_tokens gate can fire), consistent with unknown->max_tokens.
+    r = _google_unify(_gemini_resp(finish_reason="ZZ_FUTURE_REASON", with_tool=True))
+    assert r.stop_reason == "max_tokens"
+
+
+def test_gemini_known_stop_unchanged():
+    r = _google_unify(_gemini_resp(finish_reason="STOP", text="hi"))
+    assert r.stop_reason == "end_turn"

libs/openant-core/tests/test_verifier_max_tokens_finish_incomplete.py

@@ -0,0 +1,82 @@
+"""C(b): the Stage-2 verifier must not accept a `finish` tool call from a TRUNCATED
+turn (stop_reason == "max_tokens") as a completed verdict.
+
+Before this fix the block loop harvested a `finish` ToolUseBlock regardless of
+stop_reason, so a `max_tokens`-truncated reply carrying a well-formed
+`finish(agree=False, correct_finding="safe")` was parsed as a COMPLETE verdict
+(`incomplete=False`) and downgraded a Stage-1 `vulnerable` to `safe` silently — the
+verify-stage tail of the same silent-false-negative family the adapter BUG-2/BUG-7
+fixes address (the adapter now honestly reports truncation as `max_tokens`; the
+verifier must gate on it). Offline stub adapters; no real LLM calls.
+"""
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+
+_CORE_ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(_CORE_ROOT))
+
+from utilities.agentic_enhancer.repository_index import RepositoryIndex
+from utilities.finding_verifier import FindingVerifier, VerificationResult
+from utilities.llm import PhaseBinding, ToolUseBlock
+from utilities.llm.adapter import CompletionResult
+from utilities.llm_client import reset_warning_state
+
+STAGE1_FINDING = "vulnerable"
+
+
+@pytest.fixture(autouse=True)
+def _reset():
+    reset_warning_state()
+    yield
+    reset_warning_state()
+
+
+def _verify(adapter) -> VerificationResult:
+    binding = PhaseBinding(phase="verify", adapter=adapter, model="claude-x", provider_name="anthropic")
+    v = FindingVerifier(index=RepositoryIndex({}, repo_path=None), binding=binding)
+    return v.verify_result(code="x = 1", finding=STAGE1_FINDING, attack_vector="a", reasoning="r")
+
+
+class _TruncatedFinishAdapter:
+    """A finish(agree=False, safe) call on a turn the model truncated at max_tokens."""
+    name = "anthropic"
+    supports_tools = True
+    pricing = {"claude-x": {"input": 1.0, "output": 1.0}}
+
+    def complete(self, *, model, system, messages, max_tokens, tools=None):
+        return CompletionResult(
+            content=[ToolUseBlock(id="t1", name="finish",
+                                  input={"agree": False, "correct_finding": "safe"})],
+            input_tokens=1, output_tokens=1, stop_reason="max_tokens",
+        )
+
+
+class _CompleteFinishAdapter:
+    """Regression guard: a finish on a NORMAL turn (tool_use) is still accepted."""
+    name = "anthropic"
+    supports_tools = True
+    pricing = {"claude-x": {"input": 1.0, "output": 1.0}}
+
+    def complete(self, *, model, system, messages, max_tokens, tools=None):
+        return CompletionResult(
+            content=[ToolUseBlock(id="t1", name="finish",
+                                  input={"agree": True, "correct_finding": "vulnerable"})],
+            input_tokens=1, output_tokens=1, stop_reason="tool_use",
+        )
+
+
+def test_truncated_finish_at_max_tokens_is_incomplete_not_safe():
+    r = _verify(_TruncatedFinishAdapter())
+    assert r.incomplete is True                     # must NOT be a completed verdict
+    assert r.correct_finding == STAGE1_FINDING       # Stage-1 verdict preserved, not "safe"
+    assert r.agree is False
+
+
+def test_complete_finish_still_accepted():
+    r = _verify(_CompleteFinishAdapter())
+    assert r.incomplete is False                     # a normal finish is still a real verdict
+    assert r.agree is True

libs/openant-core/utilities/agentic_enhancer/agent.py

@@ -351,6 +351,35 @@ def analyze_unit(
                             )
                         )
 
+            # R2-B: a finish call on a turn the model TRUNCATED (stop_reason ==
+            # "max_tokens") is not a trustworthy complete classification — a
+            # truncated finish defaulting security_classification to "neutral"
+            # would silently drop a unit from the analysed set (a coverage/recall
+            # loss). Treat it as INCOMPLETE, mirroring this agent's degenerate-exit
+            # handling and the verifier's max_tokens gate.
+            if finish_result and stop_reason == "max_tokens":
+                call_record = self.tracker.record_call(
+                    model=self.binding.model,
+                    input_tokens=total_input_tokens,
+                    output_tokens=total_output_tokens,
+                    pricing=lookup_pricing(self.binding),
+                )
+                return AgentResult(
+                    include_functions=[],
+                    usage_context="Agent finish call truncated at max_tokens",
+                    security_classification=INCOMPLETE_CLASSIFICATION,
+                    classification_reasoning="Analysis incomplete - finish call truncated",
+                    confidence=0.3,
+                    iterations=iterations,
+                    total_tokens=total_input_tokens + total_output_tokens,
+                    is_entry_point=is_entry_point,
+                    reachable_from_entry=reachable_from_entry,
+                    entry_point_path=entry_point_path,
+                    input_tokens=total_input_tokens,
+                    output_tokens=total_output_tokens,
+                    cost_usd=call_record.get("cost_usd", 0.0),
+                )
+
             # If finish was called, return result
             if finish_result:
                 # Record token usage

libs/openant-core/utilities/finding_verifier.py

@@ -422,6 +422,15 @@ def verify_result(
                 # sets result["finding"] = correct_finding, and the report
                 # filters on that field — using "inconclusive" here would drop
                 # a Stage-1 "vulnerable" from the report entirely.
+                # C(a): record spend on this degenerate exit too — the three sibling
+                # degenerate paths (finish, no-tool-calls, max-iterations) all record,
+                # this one alone did not, undercounting the unit's tokens/cost.
+                self.tracker.record_call(
+                    model=self.binding.model,
+                    input_tokens=total_input_tokens,
+                    output_tokens=total_output_tokens,
+                    pricing=lookup_pricing(self.binding),
+                )
                 return VerificationResult(
                     agree=False,
                     correct_finding=finding,
@@ -463,6 +472,29 @@ def verify_result(
                             )
                         )
 
+            # A finish call on a turn the model TRUNCATED (stop_reason == "max_tokens")
+            # is not a trustworthy completed verdict: a well-formed
+            # finish(agree=False, "safe") from a cut-off turn would silently downgrade a
+            # Stage-1 vulnerable. Treat it as verification-incomplete (preserve the
+            # Stage-1 verdict for triage) — honoring the adapter's truncation signal
+            # (the responses/chat paths relabel abnormal terminations to "max_tokens")
+            # rather than reading a truncated reply as a clean verdict.
+            if finish_result and stop_reason == "max_tokens":
+                self.tracker.record_call(
+                    model=self.binding.model,
+                    input_tokens=total_input_tokens,
+                    output_tokens=total_output_tokens,
+                    pricing=lookup_pricing(self.binding),
+                )
+                return VerificationResult(
+                    agree=False,
+                    correct_finding=finding,
+                    explanation="Verification incomplete (finish call truncated at max_tokens)",
+                    iterations=iterations,
+                    total_tokens=total_input_tokens + total_output_tokens,
+                    incomplete=True,
+                )
+
             if finish_result:
                 self.tracker.record_call(
                     model=self.binding.model,