diff --git a/libs/openant-core/context/repo_explorer.py b/libs/openant-core/context/repo_explorer.py index 156c8a3..d4e5629 100644 --- a/libs/openant-core/context/repo_explorer.py +++ b/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( diff --git a/libs/openant-core/tests/test_agent_degenerate_exit.py b/libs/openant-core/tests/test_agent_degenerate_exit.py index ee992b8..35e06d4 100644 --- a/libs/openant-core/tests/test_agent_degenerate_exit.py +++ b/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([ diff --git a/libs/openant-core/tests/test_llm_anthropic_adapter.py b/libs/openant-core/tests/test_llm_anthropic_adapter.py index dca23ae..3b61205 100644 --- a/libs/openant-core/tests/test_llm_anthropic_adapter.py +++ b/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): diff --git a/libs/openant-core/tests/test_repo_explorer_loop.py b/libs/openant-core/tests/test_repo_explorer_loop.py index f9ac560..4d30f7b 100644 --- a/libs/openant-core/tests/test_repo_explorer_loop.py +++ b/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. diff --git a/libs/openant-core/tests/test_truncation_silent_fn_family.py b/libs/openant-core/tests/test_truncation_silent_fn_family.py new file mode 100644 index 0000000..9c270f8 --- /dev/null +++ b/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" diff --git a/libs/openant-core/tests/test_verifier_max_tokens_finish_incomplete.py b/libs/openant-core/tests/test_verifier_max_tokens_finish_incomplete.py new file mode 100644 index 0000000..23a127e --- /dev/null +++ b/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 diff --git a/libs/openant-core/utilities/agentic_enhancer/agent.py b/libs/openant-core/utilities/agentic_enhancer/agent.py index e61a307..fb8d277 100644 --- a/libs/openant-core/utilities/agentic_enhancer/agent.py +++ b/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 diff --git a/libs/openant-core/utilities/finding_verifier.py b/libs/openant-core/utilities/finding_verifier.py index 4af76c7..7151e68 100644 --- a/libs/openant-core/utilities/finding_verifier.py +++ b/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, diff --git a/libs/openant-core/utilities/llm/providers/anthropic.py b/libs/openant-core/utilities/llm/providers/anthropic.py index 3c0fc3f..2bb25f7 100644 --- a/libs/openant-core/utilities/llm/providers/anthropic.py +++ b/libs/openant-core/utilities/llm/providers/anthropic.py @@ -377,15 +377,19 @@ def _response_to_unified(response: Any) -> CompletionResult: if should_warn: sys.stderr.write( f"warning: AnthropicAdapter received unknown stop_reason " - f"{raw_stop!r}; normalising to 'end_turn'. Add this value " - f"to StopReason in utilities/llm/adapter.py and the " + f"{raw_stop!r}; treating as 'max_tokens' (not a clean finish). " + f"Add this value to StopReason in utilities/llm/adapter.py and the " f"_ANTHROPIC_STOP_REASONS table if it's a new SDK addition.\n" ) return CompletionResult( content=content_blocks, input_tokens=getattr(usage, "input_tokens", 0), output_tokens=getattr(usage, "output_tokens", 0), - stop_reason=_ANTHROPIC_STOP_REASONS.get(raw_stop, "end_turn"), + # R2-C: an unknown/abnormal stop_reason defaults to "max_tokens" (not + # "end_turn") — as the warning above notes, treating a refusal/abnormal + # termination as end_turn masks false negatives. Known values (end_turn/ + # max_tokens/tool_use/stop_sequence) use their explicit mapping. + stop_reason=_ANTHROPIC_STOP_REASONS.get(raw_stop, "max_tokens"), raw=response, ) diff --git a/libs/openant-core/utilities/llm/providers/google.py b/libs/openant-core/utilities/llm/providers/google.py index acc7643..169c744 100644 --- a/libs/openant-core/utilities/llm/providers/google.py +++ b/libs/openant-core/utilities/llm/providers/google.py @@ -441,16 +441,14 @@ def _response_to_unified(response: Any) -> CompletionResult: stop_reason: StopReason has_tool_use = any(isinstance(b, ToolUseBlock) for b in content_blocks) - if has_tool_use: - # Gemini doesn't use a dedicated finish_reason for tool calls; - # the presence of a function_call part IS the signal. - stop_reason = "tool_use" - elif raw_finish in _GEMINI_FINISH_REASONS: - stop_reason = _GEMINI_FINISH_REASONS[raw_finish] - else: - # SAFETY / RECITATION / BLOCKLIST / OTHER — warn once, fall - # back to end_turn so pipeline code keeps moving. A future - # release should widen StopReason if these become common. + mapped = _GEMINI_FINISH_REASONS.get(raw_finish) + if mapped is None: + # SAFETY/RECITATION/BLOCKLIST refusals already raised above; a remaining + # unmapped value is an UNKNOWN/abnormal termination. R2-C + round-5: it is + # not a clean finish AND it wins over tool_use — Gemini emits a function_call + # part even on an abnormal termination, so an unknown reason carrying a tool + # call must NOT be laundered into a clean tool_use. Warn once, treat as + # max_tokens (mirrors the OpenAI adapter; checked BEFORE has_tool_use). should_warn = False with _warned_finish_reasons_lock: if raw_finish not in _warned_finish_reasons: @@ -459,12 +457,26 @@ def _response_to_unified(response: Any) -> CompletionResult: if should_warn: sys.stderr.write( f"warning: GoogleAdapter received unknown finish_reason " - f"{raw_finish!r}; normalising to 'end_turn'. Add this value " - f"to StopReason in utilities/llm/adapter.py and " + f"{raw_finish!r}; treating as 'max_tokens' (not a clean finish). " + f"Add this value to StopReason in utilities/llm/adapter.py and " f"_GEMINI_FINISH_REASONS if Gemini added a new termination " f"reason.\n" ) - stop_reason = "end_turn" + stop_reason = "max_tokens" + elif mapped == "max_tokens": + # R2-A: a TRUNCATED response wins over tool_use. Gemini emits a + # function_call part even when it hit the token cap, so tool_use must + # not mask MAX_TOKENS (mirrors the OpenAI responses path) — otherwise a + # truncated finish call reaches the consumer as a clean tool_use and is + # accepted as a complete verdict/classification (a silent false-negative). + stop_reason = "max_tokens" + elif has_tool_use: + # A KNOWN, non-truncation finish reason with a function_call part: Gemini + # doesn't use a dedicated finish_reason for tool calls, so the part IS the + # signal. + stop_reason = "tool_use" + else: + stop_reason = mapped return CompletionResult( content=content_blocks,