diff --git a/src/code_forge/llm_invoke.py b/src/code_forge/llm_invoke.py index ccd30bac..130a6c83 100644 --- a/src/code_forge/llm_invoke.py +++ b/src/code_forge/llm_invoke.py @@ -15,6 +15,7 @@ import http.client import json import logging +import math import os import random import re @@ -1135,6 +1136,36 @@ def _loads_model_json(text: str, reply_keys: ReplyKeys | None = None): return _parse_model_json(text, reply_keys)[0] +def _decode_reply(text: str, reply_keys: ReplyKeys | None, strict_response: bool): + """Decode a complete response without losing root provenance when requested.""" + if not strict_response: + return _loads_model_json(text, reply_keys) + + def finite_float(value): + result = float(value) + if not math.isfinite(result): + raise ValueError("Non-finite JSON number") + return result + + def refuse_constant(value): + raise ValueError("Non-finite JSON constant") + + try: + parsed = json.loads( + text, + object_pairs_hook=_refuse_duplicate_keys, + parse_float=finite_float, + parse_constant=refuse_constant, + ) + if not isinstance(parsed, (dict, list)): + raise json.JSONDecodeError("Response root must be an object or array", text, 0) + return parsed + except json.JSONDecodeError: + raise + except (_DuplicateKey, ValueError, RecursionError) as exc: + raise json.JSONDecodeError("Invalid complete JSON response", text, 0) from exc + + # Bounds on the quote repair and on the embedded-object scan (#264). Each # repair attempt parses the whole text again, so the work is the text # length times the attempts. A review reply is a few kilobytes. @@ -2102,6 +2133,7 @@ def llm_invoke( *, l1_evidence_required: bool = False, reply_keys: ReplyKeys | None = None, + strict_response: bool = False, ) -> LLMResult: """Invoke LLM via backend (cli subprocess or api HTTP). @@ -2129,6 +2161,12 @@ def llm_invoke( l1_evidence_required: Host-computed L1 applicability. Ask once for missing inspection evidence even on a quiet review; callers still validate the returned excerpts against source. + strict_response: Non-L1 response policy: require one complete JSON + object or array, without quote + repair, embedded extraction, fences, or continuation replacement. + Callers still validate their own schema. Default preserves legacy + review/falsifier recovery. Disables L1 excerpt repair and cannot + be combined with l1_evidence_required. reply_keys: The keys the reply may carry, per level. When a reply parses only after quote repair and carries a key not listed, the repair is discarded and the parse fails (#264); api @@ -2141,8 +2179,12 @@ def llm_invoke( Raises: LLMInvokeError: on timeout, nonzero exit, HTTP error, or JSON parse failure """ + if type(strict_response) is not bool: + raise ValueError("strict_response must be a bool") if type(l1_evidence_required) is not bool: raise ValueError("l1_evidence_required must be a bool") + if strict_response and l1_evidence_required: + raise ValueError("strict_response is incompatible with l1_evidence_required") if backend is None: # Fail closed: the old fallthrough to DEFAULT_BACKEND spawned an # implicit claude -p subprocess (session-default cli backend), @@ -2185,7 +2227,9 @@ def llm_invoke( "applied. Remove them, or make this an api backend." % (backend.name, field), retryable=False, ) - result = _invoke_cli(prompt, backend, timeout_s, reply_keys=reply_keys) + result = _invoke_cli( + prompt, backend, timeout_s, reply_keys=reply_keys, strict_response=strict_response + ) elif backend.type == "api": result = _invoke_api( prompt, @@ -2197,11 +2241,14 @@ def llm_invoke( continuation_breaker=continuation_breaker, retry_timeout=retry_timeout, reply_keys=reply_keys, + strict_response=strict_response, ) else: raise LLMInvokeError("unsupported backend type: %r" % backend.type) - if _needs_excerpt_repair(result.content, expected_keys, l1_evidence_required=l1_evidence_required): + if not strict_response and _needs_excerpt_repair( + result.content, expected_keys, l1_evidence_required=l1_evidence_required + ): repaired, extra, extra_s = _repair_missing_excerpts( result.content, prompt, @@ -2246,6 +2293,7 @@ def _invoke_cli( timeout_s: int, *, reply_keys: ReplyKeys | None = None, + strict_response: bool = False, ) -> LLMResult: """Invoke LLM via cli subprocess. Returns LLMResult with Usage(0,0)""" # Resolve binary: use backend.command if set, else default to "claude" @@ -2351,7 +2399,7 @@ def _invoke_cli( duration_s=duration, ) - stdout = _strip_fences(stdout_data) + stdout = stdout_data if strict_response else _strip_fences(stdout_data) def refused(exc: json.JSONDecodeError, text: str) -> LLMInvokeError: # A refused repair or key (#264) fails the call: keeping the raw @@ -2367,7 +2415,10 @@ def refused(exc: json.JSONDecodeError, text: str) -> LLMInvokeError: try: # No reply_keys: an envelope's own keys are not the reply's. A # duplicate key is refused here all the same. - parsed, stdout_repaired = _parse_model_json(stdout) + if strict_response: + parsed, stdout_repaired = _decode_reply(stdout, None, True), False + else: + parsed, stdout_repaired = _parse_model_json(stdout) except RepairRefused as exc: raise refused(exc, stdout) from exc except json.JSONDecodeError as exc: @@ -2416,12 +2467,14 @@ def refused(exc: json.JSONDecodeError, text: str) -> LLMInvokeError: usage_data = parsed.get("usage") or {} raw_result = parsed["result"] if isinstance(raw_result, str): - stripped = _strip_fences(raw_result) + stripped = raw_result if strict_response else _strip_fences(raw_result) try: - content = _loads_model_json(stripped, reply_keys) + content = _decode_reply(stripped, reply_keys, strict_response) except RepairRefused as exc: raise refused(exc, stripped) from exc - except json.JSONDecodeError: + except json.JSONDecodeError as exc: + if strict_response: + raise refused(exc, stripped) from exc # Plain text with no repair tried: the caller's own strict # parse decides what it is. content = raw_result @@ -2637,7 +2690,7 @@ def _repair_missing_excerpts( _WIDER_RETRY_FACTOR = 2 -def _is_forge_envelope(parsed, expected_keys): +def _is_forge_envelope(parsed, expected_keys, *, strict_response=False): """The recovery path's acceptance test for a parsed dict. A forge review envelope carries both findings and code_excerpts; @@ -2645,6 +2698,8 @@ def _is_forge_envelope(parsed, expected_keys): later. Callers that pass explicit expected_keys accept full coverage of their own key set instead. """ + if strict_response and isinstance(parsed, list) and len(parsed) == 1: + parsed = parsed[0] if not isinstance(parsed, dict): return False if expected_keys is not None: @@ -2672,6 +2727,7 @@ def _continue_truncated( budget: int = 2, breaker: "TruncationBreaker | None" = None, reply_keys: ReplyKeys | None = None, + strict_response: bool = False, ): """Recover a truncated reply with a bounded continuation. @@ -2767,7 +2823,9 @@ def _continue_truncated( continue if not isinstance(cont, str): cont = "" - parsed = _continued_reply(truncated.content, cont, expected_keys, reply_keys) + parsed = _continued_reply( + truncated.content, cont, expected_keys, reply_keys, strict_response + ) if isinstance(parsed, str): last_failure = parsed continue @@ -2797,6 +2855,7 @@ def _continued_reply( cont: str, expected_keys: frozenset[str] | None, reply_keys: ReplyKeys | None, + strict_response: bool = False, ): """The reply from a cut partial and its continuation, or why not (a str). @@ -2811,29 +2870,33 @@ def _continued_reply( reply, which is taken only when it parses strictly with no fallback. """ readings = [partial + cont] - if _ECHO_ONLY.fullmatch(cont): + if not strict_response and _ECHO_ONLY.fullmatch(cont): readings.append(partial) failure = "combined output is not valid JSON" for text in readings: - cleaned = _strip_fences(text) + cleaned = text if strict_response else _strip_fences(text) try: - parsed = _loads_model_json(cleaned, reply_keys) + parsed = _decode_reply(cleaned, reply_keys, strict_response) except json.JSONDecodeError as parse_error: - parsed = _fallback_reply(cleaned, parse_error, expected_keys, reply_keys) + parsed = ( + None + if strict_response + else _fallback_reply(cleaned, parse_error, expected_keys, reply_keys) + ) if parsed is None: continue # Completing the JSON is not enough: the combined shape must be # a full forge envelope, or the continuation is a failed attempt. - if _is_forge_envelope(parsed, expected_keys): + if _is_forge_envelope(parsed, expected_keys, strict_response=strict_response): return parsed failure = "combined output is not a forge envelope" restated = _strip_fences(cont) - if restated.startswith("{"): + if not strict_response and restated.startswith("{"): try: parsed = _loads_model_json(restated, reply_keys) except json.JSONDecodeError: return failure - if _is_forge_envelope(parsed, expected_keys): + if _is_forge_envelope(parsed, expected_keys, strict_response=strict_response): return parsed return failure @@ -2845,6 +2908,7 @@ def _retry_with_more_headroom( timeout_s: int, expected_keys: frozenset[str] | None, reply_keys: ReplyKeys | None = None, + strict_response: bool = False, ): """Re-ask for the whole answer with a wider output cap. @@ -2905,12 +2969,14 @@ def _retry_with_more_headroom( if not isinstance(content, str): return None - cleaned = _strip_fences(content) + cleaned = content if strict_response else _strip_fences(content) try: - parsed = _loads_model_json(cleaned, reply_keys) + parsed = _decode_reply(cleaned, reply_keys, strict_response) except json.JSONDecodeError as parse_error: - parsed = _fallback_reply(cleaned, parse_error, expected_keys, reply_keys) - if not _is_forge_envelope(parsed, expected_keys): + parsed = ( + None if strict_response else _fallback_reply(cleaned, parse_error, expected_keys, reply_keys) + ) + if not _is_forge_envelope(parsed, expected_keys, strict_response=strict_response): return None return parsed, _observed_usage(usage_data, backend.format) @@ -2926,6 +2992,7 @@ def _invoke_api( continuation_breaker: "TruncationBreaker | None" = None, retry_timeout: bool = False, reply_keys: ReplyKeys | None = None, + strict_response: bool = False, ) -> LLMResult: """Invoke LLM via HTTP API (openai or anthropic format). Returns LLMResult.""" # Look up API key (not needed for vertex which uses OAuth2) @@ -3057,11 +3124,16 @@ def _invoke_api( # is the model's complete output; replaying it five times # produced the same delimiter error on live agnes-cn. raw_content = content - content = _strip_fences(content) + if not strict_response: + content = _strip_fences(content) try: - parsed_content = _loads_model_json(content, reply_keys) + parsed_content = _decode_reply(content, reply_keys, strict_response) except json.JSONDecodeError as exc: - parsed_content = _fallback_reply(content, exc, expected_keys, reply_keys) + parsed_content = ( + None + if strict_response + else _fallback_reply(content, exc, expected_keys, reply_keys) + ) if parsed_content is None: finish_reason = "" if isinstance(usage_data, dict): @@ -3144,6 +3216,7 @@ def _invoke_api( expected_keys, breaker=breaker, reply_keys=reply_keys, + strict_response=strict_response, ) except TruncationBreakerError: raise @@ -3176,6 +3249,7 @@ def _invoke_api( timeout_s, expected_keys, reply_keys=reply_keys, + strict_response=strict_response, ) if wider is not None: parsed, usage = wider diff --git a/src/code_forge/runtime.py b/src/code_forge/runtime.py index fb410472..3f0f53a3 100644 --- a/src/code_forge/runtime.py +++ b/src/code_forge/runtime.py @@ -379,7 +379,13 @@ def run( # str.replace NOT str.format: diffs can contain literal { or }. prompt = RUNTIME_LIFECYCLE_QUESTION.replace("{diff_text}", prompt_diff) try: - result = llm_invoke(prompt, backend=self._backend, reply_keys=RUNTIME_REPLY_KEYS) + result = llm_invoke( + prompt, + backend=self._backend, + reply_keys=RUNTIME_REPLY_KEYS, + expected_keys=RUNTIME_REPLY_KEYS.top, + strict_response=True, + ) except LLMInvokeError as exc: reason = str(exc) self.infra_errors.append("RUNTIME axis LLM call failed: %s" % reason) diff --git a/tests/test_backend_timeout_s.py b/tests/test_backend_timeout_s.py index 86a08f40..a9bec38e 100644 --- a/tests/test_backend_timeout_s.py +++ b/tests/test_backend_timeout_s.py @@ -72,7 +72,8 @@ def test_cli_timeout_reaches_subprocess_communicate(monkeypatch): monkeypatch.delenv("FORGE_LLM_TIMEOUT_S", raising=False) seen = {} - def fake_cli(prompt, backend, timeout_s, *, reply_keys=None): + def fake_cli(prompt, backend, timeout_s, *, reply_keys=None, strict_response=False): + assert strict_response is False seen["timeout_s"] = timeout_s return invoke.LLMResult(content={"ok": True}, usage=invoke.Usage(), duration_s=0.0) diff --git a/tests/test_invoke_config_contract.py b/tests/test_invoke_config_contract.py index 4bccefc9..0351a21f 100644 --- a/tests/test_invoke_config_contract.py +++ b/tests/test_invoke_config_contract.py @@ -49,7 +49,8 @@ def test_cli_backend_without_http_fields_reaches_the_subprocess(monkeypatch): """The refusal above must not fire on a backend that carries no HTTP fields.""" seen = {} - def fake_cli(prompt, backend, timeout_s, *, reply_keys=None): + def fake_cli(prompt, backend, timeout_s, *, reply_keys=None, strict_response=False): + assert strict_response is False seen["prompt"] = prompt raise LLMInvokeError("sentinel from cli", retryable=True) diff --git a/tests/test_output_budget_retry.py b/tests/test_output_budget_retry.py index 5c85ad91..98ac4110 100644 --- a/tests/test_output_budget_retry.py +++ b/tests/test_output_budget_retry.py @@ -245,11 +245,13 @@ def test_limit_is_protected_and_api_only(transport, monkeypatch): def test_cli_zero_limit_preserves_existing_route_without_http(transport, monkeypatch): calls = [] - monkeypatch.setattr( - invoke, - "_invoke_cli", - lambda *args, reply_keys=None: calls.append(args) or invoke.LLMResult(ENVELOPE, invoke.Usage()), - ) + + def fake_cli(*args, reply_keys=None, strict_response=False): + assert strict_response is False + calls.append(args) + return invoke.LLMResult(ENVELOPE, invoke.Usage()) + + monkeypatch.setattr(invoke, "_invoke_cli", fake_cli) config = BackendConfig(name="cli", type="cli", model="") assert invoke.llm_invoke("prompt", backend=config).content == ENVELOPE assert len(calls) == 1 diff --git a/tests/test_runtime_transport_contract.py b/tests/test_runtime_transport_contract.py new file mode 100644 index 00000000..5b5c5146 --- /dev/null +++ b/tests/test_runtime_transport_contract.py @@ -0,0 +1,258 @@ +"""Runtime response provenance through the real invocation transport (#30).""" + +import dataclasses +import importlib +import json +from unittest.mock import Mock + +import pytest + +from code_forge.backend import BackendConfig +from code_forge.runtime import RuntimeRunner + +invoke = importlib.import_module("code_forge.llm_invoke") +DIFF = "diff --git a/a.py b/a.py\n--- a/a.py\n+++ b/a.py\n@@ -1 +1 @@\n-old\n+new\n" +RISK = {"surfaces": ["network"], "findings": [{"file": "a.py", "line": 1, "description": "risk"}]} +EMPTY = '{"surfaces":[],"findings":[]}' +TRAP = ( + '{"surfaces":["network"],"findings":[{"file":"a.py","line":1,' + '"description":"risk","details":{"surfaces":[],"findings":[]}}' +) + + +@pytest.fixture(params=["openai", "anthropic", "vertex"]) +def api(monkeypatch, request): + monkeypatch.setenv("RUNTIME_TEST_KEY", "local-test") + monkeypatch.setattr(invoke.time, "sleep", Mock()) + format_name = request.param + provider = Mock() + monkeypatch.setattr(invoke, "_invoke_" + format_name, provider) + backend = BackendConfig( + name="runtime-test", + type="api", + format=format_name, + base_url="http://127.0.0.1:1/v1", + api_key_env="RUNTIME_TEST_KEY", + model="test", + max_tokens=1024, + ) + return RuntimeRunner(backend=backend), provider + + +def response(text): + return text, {"prompt_tokens": 1, "completion_tokens": 2, "_forge_finish_reason": "stop"} + + +def test_valid_risk_needs_one_request(api, tmp_path): + runner, request = api + request.return_value = response(json.dumps(RISK)) + findings = runner.run(DIFF, tmp_path) + assert request.call_count == 1 + assert any(f.description == "risk" for f in findings) + assert runner.last_surfaces == ["network"] + assert not runner.infra_errors + + +def test_continuation_cannot_replace_original_risk(api, tmp_path): + runner, request = api + request.side_effect = [response(TRAP), response(EMPTY), response(EMPTY)] + findings = runner.run(DIFF, tmp_path) + assert any(f.id == "runtime-skipped" for f in findings) + assert runner.infra_errors + assert runner.last_surfaces == [] + assert request.call_count == 3 + + +@pytest.mark.parametrize("text", [EMPTY + " explanation", "```json\n" + EMPTY + "\n```"]) +def test_non_root_text_is_not_silently_extracted(api, tmp_path, text): + runner, request = api + request.return_value = response(text) + findings = runner.run(DIFF, tmp_path) + assert any(f.id == "runtime-skipped" for f in findings) + assert runner.infra_errors + assert runner.last_surfaces == [] + + +@pytest.mark.parametrize( + "text", + [ + '{"surfaces":[],"findings":[],"findings":[]}', + '{"surfaces":[],"findings":[],"meta":NaN}', + '{"surfaces":[],"findings":[],"meta":1e999}', + '{"surfaces":[],"findings":[{"file":"a.py","line":true,"description":"risk"}]}', + '{"surfaces":["network"],"findings":[{"file":"a.py","line":1,"description":"say "hi""}]}', + '{"surfaces":[],"findings":[{"file":"a.py","line":1,"description":"ok"},null]}', + ], +) +def test_malformed_api_response_visible_skip(api, tmp_path, text): + runner, request = api + request.return_value = response(text) + assert any(f.id == "runtime-skipped" for f in runner.run(DIFF, tmp_path)) + assert runner.infra_errors + assert runner.last_surfaces == [] + assert request.call_count <= 2 + + +@pytest.mark.parametrize("wrapped", [False, True]) +def test_complete_response_keeps_metadata_and_singleton_compat(api, tmp_path, wrapped): + runner, request = api + payload = {**RISK, "meta": {"unknown": "inert"}} + request.return_value = response(json.dumps([payload] if wrapped else payload)) + assert any(f.description == "risk" for f in runner.run(DIFF, tmp_path)) + assert request.call_count == 1 + assert not runner.infra_errors + + +def test_genuine_continuation_preserves_original_risk(api, tmp_path): + runner, request = api + request.side_effect = [response(TRAP), response("]}")] + findings = runner.run(DIFF, tmp_path) + assert any(f.description == "risk" for f in findings) + assert runner.last_surfaces == ["network"] + assert not runner.infra_errors + assert request.call_count == 2 + + +def test_correction_must_be_another_complete_root(api, tmp_path): + runner, request = api + request.side_effect = [response('{"surfaces":[], "findings":[],}'), response(json.dumps(RISK))] + assert any(f.description == "risk" for f in runner.run(DIFF, tmp_path)) + assert not runner.infra_errors + assert request.call_count == 2 + + +@pytest.mark.parametrize("valid", [False, True]) +def test_fresh_wider_retry_uses_same_boundary(api, tmp_path, valid): + runner, request = api + runner._backend = dataclasses.replace(runner._backend, output_token_limit=4096) + last = json.dumps(RISK) if valid else EMPTY + " unrelated prose" + request.side_effect = [response(TRAP), response(EMPTY), response(EMPTY), response(last)] + findings = runner.run(DIFF, tmp_path) + assert request.call_count == 4 + assert bool(runner.infra_errors) is not valid + assert any(f.description == "risk" for f in findings) is valid + assert any(f.id == "runtime-skipped" for f in findings) is not valid + + +@pytest.mark.parametrize("wrapper", ["raw", "string", "object", "events"]) +@pytest.mark.parametrize( + "payload", ["risk", "duplicate", "nonfinite", "string-root", "fence", "truncated"] +) +def test_cli_root_contract(monkeypatch, tmp_path, wrapper, payload): + from unittest.mock import MagicMock + + # An encoded string inserted as an envelope member is the normal + # string-result wire format, not a double-encoded model response. + valid = payload == "risk" or (payload == "string-root" and wrapper == "object") + risk = json.dumps(RISK) + text = { + "risk": risk, + "duplicate": risk[:-1] + ',"findings":[]}', + "nonfinite": risk[:-1] + ',"meta":NaN}', + "string-root": json.dumps(risk), + "fence": "```json\n" + risk + "\n```", + "truncated": TRAP, + }[payload] + if wrapper in {"string", "events"}: + envelope = {"type": "result", "result": text} + text = json.dumps([{"type": "system"}, envelope] if wrapper == "events" else envelope) + elif wrapper == "object": + text = '{"type":"result","result":' + text + "}" + proc = MagicMock(returncode=0, pid=12345) + proc.communicate.return_value = (text, "") + monkeypatch.setattr(invoke.shutil, "which", Mock(return_value="/usr/bin/claude")) + popen = Mock(return_value=proc) + monkeypatch.setattr(invoke.subprocess, "Popen", popen) + runner = RuntimeRunner(backend=BackendConfig(name="runtime-cli", type="cli", model="test")) + findings = runner.run(DIFF, tmp_path) + assert popen.call_count == 1 + assert bool(runner.infra_errors) is not valid + assert any(f.description == "risk" for f in findings) is valid + assert any(f.id == "runtime-skipped" for f in findings) is not valid + + +@pytest.mark.parametrize("value", [None, 0, 1, "true"]) +def test_strict_response_requires_bool_before_io(monkeypatch, value): + request = Mock(side_effect=AssertionError("unexpected dispatch")) + monkeypatch.setattr(invoke, "_invoke_api", request) + with pytest.raises(ValueError, match="strict_response must be a bool"): + invoke.llm_invoke("prompt", strict_response=value) + request.assert_not_called() + + +@pytest.mark.parametrize("format_name", ["openai", "anthropic"]) +@pytest.mark.parametrize("truncated", [False, True]) +def test_actual_http_adapter_keeps_runtime_root(monkeypatch, tmp_path, format_name, truncated): + """Only the HTTP byte boundary is replaced, not the provider adapter.""" + from unittest.mock import MagicMock + + monkeypatch.setenv("RUNTIME_TEST_KEY", "local-test") + monkeypatch.setattr(invoke.time, "sleep", Mock()) + + def wire(text, cut=False): + if format_name == "openai": + payload = { + "choices": [{"message": {"content": text}, "finish_reason": "length" if cut else "stop"}] + } + else: + payload = { + "content": [{"type": "text", "text": text}], + "stop_reason": "max_tokens" if cut else "end_turn", + } + response = MagicMock() + response.__enter__.return_value = response + response.read.return_value = json.dumps(payload).encode() + return response + + replies = [wire(TRAP, True), wire(EMPTY), wire(EMPTY)] if truncated else [wire(json.dumps(RISK))] + http = Mock(side_effect=replies) + monkeypatch.setattr(invoke.urllib.request, "urlopen", http) + backend = BackendConfig( + name="runtime-http", + type="api", + format=format_name, + base_url="https://example.invalid/v1", + api_key_env="RUNTIME_TEST_KEY", + model="test", + max_tokens=1024, + ) + runner = RuntimeRunner(backend=backend) + findings = runner.run(DIFF, tmp_path) + assert bool(runner.infra_errors) is truncated + assert any(f.id == "runtime-skipped" for f in findings) is truncated + assert any(f.description == "risk" for f in findings) is not truncated + assert http.call_count == (3 if truncated else 1) + + +def test_string_root_cannot_be_decoded_twice(api, tmp_path): + runner, request = api + request.return_value = response(json.dumps(json.dumps(RISK))) + findings = runner.run(DIFF, tmp_path) + assert any(f.id == "runtime-skipped" for f in findings) + assert runner.infra_errors + assert runner.last_surfaces == [] + + +def test_strict_response_cannot_request_l1_evidence(): + with pytest.raises(ValueError, match="incompatible"): + invoke.llm_invoke("prompt", strict_response=True, l1_evidence_required=True) + + +def test_strict_response_without_keys_does_not_repair_excerpts(api): + runner, request = api + request.return_value = response(json.dumps(RISK)) + result = invoke.llm_invoke("prompt", backend=runner._backend, strict_response=True) + assert result.content == RISK + assert request.call_count == 1 + + +def test_runtime_forwards_strict_policy_to_cli(monkeypatch, tmp_path): + def fake_cli(prompt, backend, timeout_s, *, reply_keys=None, strict_response=False): + assert strict_response is True + assert reply_keys.top == {"surfaces", "findings"} + return invoke.LLMResult(content=RISK, usage=invoke.Usage()) + + monkeypatch.setattr(invoke, "_invoke_cli", fake_cli) + runner = RuntimeRunner(backend=BackendConfig(name="runtime-cli", type="cli", model="test")) + assert any(f.description == "risk" for f in runner.run(DIFF, tmp_path)) + assert not runner.infra_errors