diff --git a/src/code_forge/mcp_jobs.py b/src/code_forge/mcp_jobs.py index 678c839a..7e897013 100644 --- a/src/code_forge/mcp_jobs.py +++ b/src/code_forge/mcp_jobs.py @@ -99,6 +99,8 @@ class ForgeResult(BaseModel): findings_count_status: str | None = None findings: list[dict] | None = None # compact finding summaries for MCP receipt_audit: list[dict] | None = None # metadata diagnostics, not product findings + # Validated optional advisory outcome; None means no validated outcome. + test_assertion: dict[str, Any] | None = None duration_s: float output: str # Additive #332 report_classification block (IC-03 include routes). @@ -404,9 +406,7 @@ async def _wait_for_job(job_id: str) -> None: "stderr": ("stdout lost: process exited at timeout boundary\n" + stderr_tail), "exit_code": proc.returncode, # Exit 0 with the report lost is not a PASS: nothing was read. - "verdict": ( - "UNRELIABLE" if proc.returncode == 0 else exit_to_verdict(proc.returncode) - ), + "verdict": ("UNRELIABLE" if proc.returncode == 0 else exit_to_verdict(proc.returncode)), "stdout_lost": True, "duration_s": elapsed, } diff --git a/src/code_forge/mcp_server.py b/src/code_forge/mcp_server.py index d9f8da62..2a8c1423 100644 --- a/src/code_forge/mcp_server.py +++ b/src/code_forge/mcp_server.py @@ -58,6 +58,12 @@ ) import code_forge +from code_forge.assertion_outcome import ( + AssertionKind, + AssertionOutcome, + AssertionStage, + AssertionStatus, +) from code_forge.exit_codes import EXIT_CLI_ERROR, EXIT_TIMEOUT from code_forge.llm_invoke import effective_invoke_timeout_s from code_forge.mcp_jobs import ( @@ -130,7 +136,8 @@ async def receive(self) -> SessionMessage | Exception: _validate_timeout(request.params["arguments"].get("timeout_s")) except ToolError as exc: error = JSONRPCError( - jsonrpc="2.0", id=request.id, + jsonrpc="2.0", + id=request.id, error=ErrorData(code=INVALID_PARAMS, message=str(exc)), ) await self._outgoing.send(SessionMessage(message=JSONRPCMessage(error))) @@ -520,16 +527,26 @@ def _trusted_test_block(workspace: Path) -> dict: root = os.environ.get("FORGE_TEST_CONFIG_ROOT") or str(workspace) try: - head = subprocess.run( - ["git", "rev-parse", "--verify", "-q", "HEAD"], cwd=str(workspace), - capture_output=True, text=True, encoding="utf-8", timeout=10, check=False, - stdin=subprocess.DEVNULL, - ).stdout.strip() or None + head = ( + subprocess.run( + ["git", "rev-parse", "--verify", "-q", "HEAD"], + cwd=str(workspace), + capture_output=True, + text=True, + encoding="utf-8", + timeout=10, + check=False, + stdin=subprocess.DEVNULL, + ).stdout.strip() + or None + ) except (OSError, subprocess.TimeoutExpired): head = None try: test = load_trusted_gate_config( - config_root=Path(root), source_root=workspace, base_sha=head, + config_root=Path(root), + source_root=workspace, + base_sha=head, ).get("test") except Exception: # noqa: BLE001 - only sizes a time window return {} @@ -548,7 +565,10 @@ def _silent_step_s(workspace: Path) -> int: from .mutation_config import DEFAULT_MUTATION_TIMEOUT # #183: an unset mutation timeout runs mutmut with its real default. - for key, default in (("timeout_seconds", 120), ("mutation_timeout_seconds", DEFAULT_MUTATION_TIMEOUT)): + for key, default in ( + ("timeout_seconds", 120), + ("mutation_timeout_seconds", DEFAULT_MUTATION_TIMEOUT), + ): value = test.get(key, default) if isinstance(value, bool) or not isinstance(value, int) or value <= 0: value = default @@ -712,7 +732,9 @@ async def forge_server_status() -> CallToolResult: current_digest = None source_state = "unreadable" source_error_type = type(exc).__name__ - reason = "Cannot inspect package source. Restore readable source and reconnect this MCP connection." + reason = ( + "Cannot inspect package source. Restore readable source and reconnect this MCP connection." + ) else: if current_digest == _loaded_digest: source_state = "unchanged" @@ -1051,10 +1073,42 @@ def _sarif_result_projection(stdout: str) -> tuple[int | None, list[dict] | None return _sarif_projection_with_status(stdout)[:3] +def _test_assertion_projection(value: object) -> dict[str, Any] | None: + """Whitelist optional assertion metadata through the producer's validator.""" + if type(value) is not dict or "status" not in value or "finding_count" not in value: + return None + try: + status = value["status"] + stage = value.get("stage") + kind = value.get("kind") + if type(status) is not str: + return None + if stage is not None and type(stage) is not str: + return None + if kind is not None and type(kind) is not str: + return None + return AssertionOutcome( + status=AssertionStatus(status), + finding_count=value["finding_count"], + stage=AssertionStage(stage) if stage is not None else None, + kind=AssertionKind(kind) if kind is not None else None, + ).to_metadata() + except (TypeError, ValueError): + return None + + def _sarif_projection_with_status( stdout: str, exit_code: int | None = None, ) -> tuple[int | None, list[dict] | None, list[dict] | None, str | None]: + """Compatibility projection retaining the original four fields.""" + return _sarif_projection_with_assertion(stdout, exit_code)[:4] + + +def _sarif_projection_with_assertion( + stdout: str, + exit_code: int | None = None, +) -> tuple[int | None, list[dict] | None, list[dict] | None, str | None, dict[str, Any] | None]: """Read presentation from one CLI SARIF document, never review authority. The 1 MiB-character cutoff limits optional parsing and structured copies @@ -1071,22 +1125,27 @@ def _sarif_projection_with_status( Each projected list includes at most 64 dictionary/list levels so the full MCP response envelope retains ample serialization depth headroom. """ - no_report = ( - type(stdout) is str and not stdout.strip() - ) or ( + no_report = (type(stdout) is str and not stdout.strip()) or ( exit_code is not None and (exit_code < 0 or exit_code in _NO_REPORT_EXIT_CODES) ) unknown = ( - None, None, None, + None, + None, + None, _NO_REPORT_COUNT_STATUS if no_report else _UNPARSEABLE_COUNT_STATUS, + None, ) if type(stdout) is not str: - return (None, None, None, None) + return (None, None, None, None, None) if len(stdout) > _MAX_SARIF_REPORT_CHARS: - return (None, None, None, _OVERSIZE_COUNT_STATUS) + return (None, None, None, _OVERSIZE_COUNT_STATUS, None) try: - document = json.loads(stdout, object_pairs_hook=_sarif_object, - parse_float=_sarif_float, parse_constant=_sarif_float) + document = json.loads( + stdout, + object_pairs_hook=_sarif_object, + parse_float=_sarif_float, + parse_constant=_sarif_float, + ) if type(document) is not dict or document.get("version") != "2.1.0": return unknown runs = document.get("runs") @@ -1113,6 +1172,7 @@ def _sarif_projection_with_status( return unknown # Escaped unpaired surrogates can parse but cannot reach MCP UTF-8. json.dumps((findings, audit), ensure_ascii=False, allow_nan=False).encode("utf-8") + test_assertion = _test_assertion_projection(properties.get("testAssertion")) diagnostics = properties.get("providerDiagnostics") if type(diagnostics) is list and any( type(row) is dict and row.get("diagnostic_kind") == "capacity-incomplete" @@ -1120,8 +1180,8 @@ def _sarif_projection_with_status( ): # #221: a capacity-incomplete review did not acquire every pass; # its result list is not a count. Unknown, labelled -- never 0. - return None, findings, audit, _CAPACITY_INCOMPLETE_COUNT_STATUS - return len(findings), findings, audit, None + return None, findings, audit, _CAPACITY_INCOMPLETE_COUNT_STATUS, test_assertion + return len(findings), findings, audit, None, test_assertion except (ValueError, TypeError, UnicodeError, RecursionError): return unknown @@ -1140,27 +1200,17 @@ def _report_classification_from_stdout(stdout: str) -> dict[str, object]: ) if type(stdout) is not str or stdout == "": - return wire_to_json( - parse_report_classification(None, legacy_report_state="empty") - ) + return wire_to_json(parse_report_classification(None, legacy_report_state="empty")) try: document = json.loads(stdout) - except (ValueError, TypeError): - return wire_to_json( - parse_report_classification(None, legacy_report_state="rejected") - ) + except (ValueError, TypeError, RecursionError): + return wire_to_json(parse_report_classification(None, legacy_report_state="rejected")) if type(document) is not dict: - return wire_to_json( - parse_report_classification(None, legacy_report_state="rejected") - ) + return wire_to_json(parse_report_classification(None, legacy_report_state="rejected")) runs = document.get("runs") if type(runs) is not list or not runs or type(runs[0]) is not dict: - return wire_to_json( - parse_report_classification(None, legacy_report_state="rejected") - ) - return wire_to_json( - parse_report_classification(runs[0], legacy_report_state="accepted") - ) + return wire_to_json(parse_report_classification(None, legacy_report_state="rejected")) + return wire_to_json(parse_report_classification(runs[0], legacy_report_state="accepted")) def _make_result( @@ -1181,7 +1231,9 @@ def _make_result( note = signal_exit_note(exit_code) if note: output = output + ("\n" if output else "") + note - findings_count, findings, receipt_audit, count_status = _sarif_projection_with_status(stdout, exit_code) + findings_count, findings, receipt_audit, count_status, test_assertion = ( + _sarif_projection_with_assertion(stdout, exit_code) + ) structured = ForgeResult( verdict=exit_to_verdict(exit_code), exit_code=exit_code, @@ -1189,6 +1241,7 @@ def _make_result( findings_count_status=count_status, findings=findings, receipt_audit=receipt_audit, + test_assertion=test_assertion, duration_s=round(elapsed, 2), output=output, report_classification=_report_classification_from_stdout(stdout), @@ -1289,9 +1342,7 @@ async def _dispatch_cli( if deadline is None: result = await _run_cli_budgeted(*cli_args, workspace=workspace, env=env) else: - result = await _run_cli_budgeted( - *cli_args, workspace=workspace, env=env, deadline=deadline - ) + result = await _run_cli_budgeted(*cli_args, workspace=workspace, env=env, deadline=deadline) except BaseException: _unlink(contract_tmp) _unlink(focus_tmp) @@ -1557,7 +1608,9 @@ async def forge_gate_check( # #165 P3: reading the gate/config can block; keep it off the event loop. idle_s = await asyncio.to_thread(_job_idle_s, workspace, backend) if timeout_s is None else None return await _dispatch_cli( - cli_args, workspace, cap, + cli_args, + workspace, + cap, **({"timeout_s": timeout_s} if timeout_s is not None else {"idle_s": idle_s}), ) @@ -1673,16 +1726,16 @@ async def forge_job_status(job_id: str) -> CallToolResult: r = entry["result"] output = r.get("stdout", "") stderr = r.get("stderr", "") - if ( - status == "failed" or r.get("exit_code", 0) != 0 or r.get("stdout_lost") - ) and stderr.strip(): + if (status == "failed" or r.get("exit_code", 0) != 0 or r.get("stdout_lost")) and stderr.strip(): output = output + "\n--- stderr ---\n" + stderr if str(r.get("verdict", "")).startswith("SIGNALED("): note = signal_exit_note(r.get("exit_code", 0)) if note: output = output + ("\n" if output else "") + note job_stdout = r.get("stdout", "") - findings_count, findings, receipt_audit, count_status = _sarif_projection_with_status(job_stdout, r.get("exit_code")) + findings_count, findings, receipt_audit, count_status, test_assertion = ( + _sarif_projection_with_assertion(job_stdout, r.get("exit_code")) + ) forge_result = ForgeResult( verdict=r.get("verdict", UNKNOWN_VERDICT), exit_code=r.get("exit_code", UNKNOWN_EXIT_CODE), @@ -1690,6 +1743,7 @@ async def forge_job_status(job_id: str) -> CallToolResult: findings_count_status=count_status, findings=findings, receipt_audit=receipt_audit, + test_assertion=test_assertion, duration_s=r.get("duration_s", 0.0), output=output, report_classification=_report_classification_from_stdout(job_stdout), diff --git a/tests/test_mcp_advisory_results.py b/tests/test_mcp_advisory_results.py index 4627e406..53816774 100644 --- a/tests/test_mcp_advisory_results.py +++ b/tests/test_mcp_advisory_results.py @@ -39,7 +39,6 @@ def _without_report_classification(payload): return out - @pytest.fixture(autouse=True) def offline(monkeypatch): """These formatting fixtures must not contact a provider or launch a CLI.""" @@ -114,6 +113,7 @@ async def test_terminal_text_preserves_exact_structured_output( "findings_count_status": UNPARSEABLE, "findings": None, "receipt_audit": None, + "test_assertion": None, "duration_s": 12.345, "output": expected_output, }, @@ -130,10 +130,14 @@ async def test_terminal_text_preserves_exact_structured_output( @pytest.mark.parametrize("status,verdict,exit_code", TERMINAL_CASES) @pytest.mark.parametrize("output", ["", " \t\r\n"], ids=["empty", "whitespace"]) async def test_terminal_empty_and_whitespace_output(monkeypatch, status, verdict, exit_code, output): - monkeypatch.setattr(server, "get_job", lambda job_id: { - "status": status, - "result": {"stdout": output, "stderr": " \t\n", "verdict": verdict, "exit_code": exit_code}, - }) + monkeypatch.setattr( + server, + "get_job", + lambda job_id: { + "status": status, + "result": {"stdout": output, "stderr": " \t\n", "verdict": verdict, "exit_code": exit_code}, + }, + ) result = await poll(True) @@ -144,8 +148,12 @@ async def test_terminal_empty_and_whitespace_output(monkeypatch, status, verdict @pytest.mark.parametrize( "exit_code,verdict,advisory_status", - [(0, "PASS", UNPARSEABLE), (1, "FAIL", UNPARSEABLE), - (2, "CLI_ERROR", NO_REPORT), (6, "TIMEOUT", NO_REPORT)], + [ + (0, "PASS", UNPARSEABLE), + (1, "FAIL", UNPARSEABLE), + (2, "CLI_ERROR", NO_REPORT), + (6, "TIMEOUT", NO_REPORT), + ], ) @pytest.mark.parametrize("stdout", [OUTPUT, "", " \t\n"], ids=["advisory", "empty", "whitespace"]) def test_immediate_result_keeps_current_text_format(exit_code, verdict, advisory_status, stdout): @@ -160,6 +168,7 @@ def test_immediate_result_keeps_current_text_format(exit_code, verdict, advisory "findings_count_status": advisory_status if stdout == OUTPUT else NO_REPORT, "findings": None, "receipt_audit": None, + "test_assertion": None, "duration_s": 12.35, "output": expected_output, } @@ -186,7 +195,10 @@ async def test_running_status_keeps_progress_format(monkeypatch, tmp_path, sdk, expected_text += "\n--- progress ---\n" + progress.strip() assert result.content[0].text == expected_text assert result.structuredContent == { - "job_id": "job", "status": "running", "poll_after_seconds": 10, "result": None, + "job_id": "job", + "status": "running", + "poll_after_seconds": 10, + "result": None, } assert entry == before assert log.read_text(encoding="utf-8") == progress @@ -232,11 +244,20 @@ def read_tail(snapshot): @pytest.mark.asyncio async def test_stdout_lost_note_is_visible_on_exit_zero(monkeypatch): note = "stdout lost: process exited at timeout boundary\n" - monkeypatch.setattr(server, "get_job", lambda job_id: { - "status": "completed", - "result": {"stdout": "", "stderr": note, "verdict": "UNRELIABLE", - "exit_code": 0, "stdout_lost": True}, - }) + monkeypatch.setattr( + server, + "get_job", + lambda job_id: { + "status": "completed", + "result": { + "stdout": "", + "stderr": note, + "verdict": "UNRELIABLE", + "exit_code": 0, + "stdout_lost": True, + }, + }, + ) result = await poll(False) data = result.structuredContent["result"] assert data["verdict"] == "UNRELIABLE" diff --git a/tests/test_mcp_assertion_projection.py b/tests/test_mcp_assertion_projection.py new file mode 100644 index 00000000..38e5d84c --- /dev/null +++ b/tests/test_mcp_assertion_projection.py @@ -0,0 +1,347 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (c) 2026, Minxi Hou +"""Optional assertion presentation shares the accepted SARIF projection boundary.""" + +from __future__ import annotations + +import asyncio +from copy import deepcopy +import json +import socket +import subprocess +import time +from unittest.mock import Mock + +import pytest +from mcp.types import JSONRPCMessage, JSONRPCResponse + +from code_forge import mcp_server as server +from code_forge.mcp_jobs import ForgeResult + + +@pytest.fixture(autouse=True) +def offline(monkeypatch): + network = Mock(side_effect=AssertionError("unexpected network operation")) + spawn = Mock(side_effect=AssertionError("unexpected subprocess")) + for name in ("connect", "connect_ex"): + monkeypatch.setattr(socket.socket, name, network) + for name in ("create_connection", "getaddrinfo"): + monkeypatch.setattr(socket, name, network) + monkeypatch.setattr(asyncio, "create_subprocess_exec", spawn) + monkeypatch.setattr(asyncio, "create_subprocess_shell", spawn) + monkeypatch.setattr(subprocess, "Popen", spawn) + yield + network.assert_not_called() + spawn.assert_not_called() + + +VALID = [ + {"status": "completed_empty", "finding_count": 0}, + {"status": "completed_findings", "finding_count": 9}, + {"status": "not_applicable", "finding_count": 0, "stage": "eligibility", "kind": "no_test_files"}, + {"status": "skipped", "finding_count": None, "stage": "eligibility", "kind": "no_git_diff"}, + *[ + {"status": "failed", "finding_count": None, "stage": stage, "kind": kind} + for stage, kind in [ + ("preparation", "error"), + ("invocation", "error"), + ("invocation", "timeout"), + ("validation", "error"), + ("conversion", "error"), + ] + ], +] +ROUTES = ["inline", "handler", "sdk"] +MISSING = object() + + +def document(assertion=MISSING): + properties = {"receiptAudit": [{"ruleId": "audit", "message": {"text": "audit note"}}]} + if assertion is not MISSING: + properties["testAssertion"] = deepcopy(assertion) + return { + "version": "2.1.0", + "runs": [ + { + "tool": {"driver": {"name": "code-forge"}}, + "results": [{"ruleId": "product", "message": {"text": "product note"}}], + "properties": properties, + } + ], + } + + +def wire_result(result): + response = JSONRPCResponse( + jsonrpc="2.0", + id=1, + result=result.model_dump(by_alias=True, mode="json", exclude_none=True), + ) + wire = JSONRPCMessage(response).model_dump_json(by_alias=True, exclude_none=True) + return json.loads(wire.encode("utf-8"))["result"] + + +async def consume(monkeypatch, route, stdout, *, exit_code=1, status="completed", stderr=""): + if route == "inline": + result = server._make_result(stdout, exit_code, 12.345, stderr) + data = result.structuredContent + else: + entry = { + "status": status, + "result": { + "stdout": stdout, + "stderr": stderr, + "exit_code": exit_code, + "verdict": server.exit_to_verdict(exit_code), + "duration_s": 12.345, + }, + } + before = deepcopy(entry) + monkeypatch.setattr(server, "get_job", lambda job_id: entry) + result = ( + await server.mcp.call_tool("forge_job_status", {"job_id": "job"}) + if route == "sdk" + else await server.forge_job_status("job") + ) + assert entry == before + assert result.structuredContent["status"] == status + assert result.structuredContent["poll_after_seconds"] is None + data = result.structuredContent["result"] + assert result.isError is False + wire = wire_result(result) + assert wire["structuredContent"] == result.structuredContent + return result, data + + +@pytest.mark.asyncio +@pytest.mark.parametrize("route", ROUTES) +@pytest.mark.parametrize("status", ["completed", "failed"]) +@pytest.mark.parametrize("exit_code", [0, 1]) +@pytest.mark.parametrize("assertion", VALID) +async def test_valid_assertions_all_routes(monkeypatch, route, status, exit_code, assertion): + report = document(assertion) + stdout = " \n" + json.dumps(report) + "\n" + result, data = await consume(monkeypatch, route, stdout, status=status, exit_code=exit_code) + assert data["test_assertion"] == assertion + # Advisory finding_count is independent of this report's one product result. + assert data["findings_count"] == 1 + assert data["findings_count_status"] is None + assert data["findings"] == report["runs"][0]["results"] + assert data["receipt_audit"] == report["runs"][0]["properties"]["receiptAudit"] + assert data["verdict"] == ("PASS" if exit_code == 0 else "FAIL") + assert data["exit_code"] == exit_code + assert data["duration_s"] == (12.35 if route == "inline" else 12.345) + assert data["output"] == stdout + assert stdout in result.content[0].text + + +INVALID = [ + None, + [], + "completed_empty", + 0, + True, + {}, + {"finding_count": 0}, + {"status": "completed_empty"}, + {"status": "skipped", "stage": "eligibility", "kind": "no_git_diff"}, + {"status": "failed", "stage": "invocation", "kind": "error"}, + *[ + {"status": "completed_empty", "finding_count": value} + for value in [False, True, -1, 0.0, "0", [], {}] + ], + *[ + {"status": status, "finding_count": 0} + for status in [None, False, 1, [], {}, "COMPLETED_EMPTY", "unknown"] + ], + {"status": "completed_findings", "finding_count": 0}, + {"status": "completed_empty", "finding_count": None}, + {"status": "completed_empty", "finding_count": 1}, + {"status": "completed_empty", "finding_count": 0, "stage": "eligibility"}, + {"status": "completed_empty", "finding_count": 0, "kind": "error"}, + {"status": "failed", "finding_count": 0, "stage": "invocation", "kind": "error"}, + *[ + {"status": "failed", "finding_count": None, "stage": stage, "kind": kind} + for stage, kind in [ + (None, None), + ("eligibility", "error"), + ("preparation", "timeout"), + ("validation", "timeout"), + ("conversion", "timeout"), + ("unknown", "error"), + ("invocation", "unknown"), + ([], "error"), + ("invocation", {}), + (True, "error"), + ("invocation", False), + ] + ], + {"status": "not_applicable", "finding_count": 0, "stage": "eligibility", "kind": "no_git_diff"}, + {"status": "skipped", "finding_count": 0, "stage": "eligibility", "kind": "no_git_diff"}, +] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("route", ROUTES) +@pytest.mark.parametrize("assertion", INVALID) +async def test_invalid_optional_assertion_keeps_legacy_projection(monkeypatch, route, assertion): + report = document(assertion) + stdout = json.dumps(report) + _, data = await consume(monkeypatch, route, stdout) + assert data["test_assertion"] is None + assert data["findings_count"] == 1 + assert data["findings"] == report["runs"][0]["results"] + assert data["receipt_audit"] == report["runs"][0]["properties"]["receiptAudit"] + assert data["findings_count_status"] is None + assert data["output"] == stdout + + +@pytest.mark.asyncio +@pytest.mark.parametrize("route", ROUTES) +async def test_missing_assertion_and_stderr_cannot_supply_metadata(monkeypatch, route): + stdout = json.dumps(document()) + stderr = "testAssertion: " + json.dumps(VALID[0]) + _, data = await consume(monkeypatch, route, stdout, stderr=stderr) + assert data["test_assertion"] is None + assert data["findings_count"] == 1 + assert data["output"] == stdout + "\n--- stderr ---\n" + stderr + + +@pytest.mark.asyncio +@pytest.mark.parametrize("route", ROUTES) +@pytest.mark.parametrize("optional_null", [False, True]) +async def test_canonical_whitelist_ignores_unknown_payload(monkeypatch, route, optional_null): + assertion = {**VALID[0], "reviewer_payload": {"secret_like": ["fixture-private-marker"]}} + if optional_null: + assertion.update(stage=None, kind=None) + stdout = json.dumps(document(assertion)) + _, data = await consume(monkeypatch, route, stdout) + assert data["test_assertion"] == VALID[0] + assert "fixture-private-marker" not in json.dumps(data["test_assertion"]) + assert data["output"] == stdout + assert "fixture-private-marker" in data["output"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("route", ROUTES) +async def test_capacity_incomplete_keeps_assertion(monkeypatch, route): + report = document(VALID[6]) + report["runs"][0]["properties"]["providerDiagnostics"] = [{"diagnostic_kind": "capacity-incomplete"}] + _, data = await consume(monkeypatch, route, json.dumps(report)) + assert data["test_assertion"] == VALID[6] + assert data["findings_count"] is None + assert data["findings_count_status"] == server._CAPACITY_INCOMPLETE_COUNT_STATUS + assert data["findings"] == report["runs"][0]["results"] + assert data["receipt_audit"] == report["runs"][0]["properties"]["receiptAudit"] + + +def damaged_stdout(damage): + report = document(VALID[0]) + run = report["runs"][0] + if damage == "wrong-tool": + run["tool"]["driver"]["name"] = "not-code-forge" + elif damage == "multi-run": + report["runs"].append(deepcopy(run)) + elif damage == "invalid-results": + run["results"] = [None] + elif damage == "invalid-audit": + run["properties"]["receiptAudit"] = [None] + elif damage == "invalid-properties": + run["properties"] = [] + elif damage == "nonfinite": + run["results"][0]["value"] = float("nan") + elif damage == "utf8": + run["results"][0]["message"]["text"] = "\ud800" + elif damage in ("deep-results", "deep-audit"): + value = 0 + for _ in range(70): + value = [value] + rows = run["results"] if damage == "deep-results" else run["properties"]["receiptAudit"] + rows[0]["nested"] = value + stdout = json.dumps(report) + if damage == "duplicate-key": + stdout = stdout.replace( + '"status": "completed_empty"', '"status": "completed_empty", "status": "completed_empty"' + ) + elif damage == "trailing-document": + stdout += " {}" + elif damage == "oversize": + stdout += " " * (server._MAX_SARIF_REPORT_CHARS + 1 - len(stdout)) + elif damage == "parser-recursion": + # Python 3.12 accepts 1200 levels; this bounded 20 KiB fixture reaches + # the decoder limit without changing global recursion settings. + stdout = stdout[:-1] + ', "nested": ' + "[" * 10000 + "0" + "]" * 10000 + "}" + return stdout + + +@pytest.mark.asyncio +@pytest.mark.parametrize("route", ROUTES) +@pytest.mark.parametrize( + "damage", + [ + "wrong-tool", + "multi-run", + "duplicate-key", + "trailing-document", + "invalid-results", + "invalid-audit", + "invalid-properties", + "nonfinite", + "utf8", + "deep-results", + "deep-audit", + "oversize", + "parser-recursion", + ], +) +async def test_rejected_envelope_cannot_project_assertion(monkeypatch, route, damage): + stdout = damaged_stdout(damage) + _, data = await consume(monkeypatch, route, stdout) + assert data["test_assertion"] is None + assert data["findings_count"] is None + assert data["findings"] is None + assert data["receipt_audit"] is None + assert data["findings_count_status"] is not None + assert data["output"] == stdout + + +def test_exact_legacy_projection_tuple_shapes(): + stdout = json.dumps(document(VALID[0])) + count, findings, audit = server._sarif_result_projection(stdout) + count4, findings4, audit4, status = server._sarif_projection_with_status(stdout, 0) + assert (count, findings, audit) == (count4, findings4, audit4) + assert count == 1 and status is None + assert server._sarif_result_projection(None) == (None, None, None) + assert server._sarif_projection_with_status(None) == (None, None, None, None) + + +def test_model_default_assertion_is_none(): + result = ForgeResult(verdict="PASS", exit_code=0, duration_s=0, output="") + assert result.test_assertion is None + assert result.model_dump()["test_assertion"] is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize("route", ["handler", "sdk"]) +async def test_running_result_remains_none_and_entry_immutable(monkeypatch, route): + entry = { + "status": "running", + "created_at": time.monotonic(), + "stderr_log_path": None, + "result": {"stdout": json.dumps(document(VALID[0]))}, + } + before = deepcopy(entry) + monkeypatch.setattr(server, "get_job", lambda job_id: entry) + result = ( + await server.mcp.call_tool("forge_job_status", {"job_id": "job"}) + if route == "sdk" + else await server.forge_job_status("job") + ) + assert result.structuredContent == { + "job_id": "job", + "status": "running", + "poll_after_seconds": 10, + "result": None, + } + assert entry == before + assert wire_result(result)["structuredContent"] == result.structuredContent