From b48edd3b41980fa8c11e804b48a805e6758f0d0c Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 31 Oct 2026 07:22:55 -0400 Subject: [PATCH] mcp/assertions: expose validated advisory outcomes in results Assertion outcomes already reach callers inside SARIF text but are absent from structured immediate and background results. Project the optional metadata through the existing outcome validator and the accepted SARIF boundary so clients can distinguish completion, findings, inapplicability, failure and skipping without interpreting stderr. Keep invalid optional metadata separate from the existing findings and receipt projections, retain tuple compatibility, and preserve raw output. Signed-off-by: Codex --- src/code_forge/mcp_jobs.py | 6 +- src/code_forge/mcp_server.py | 142 ++++++---- tests/test_mcp_advisory_results.py | 47 +++- tests/test_mcp_assertion_projection.py | 347 +++++++++++++++++++++++++ 4 files changed, 482 insertions(+), 60 deletions(-) create mode 100644 tests/test_mcp_assertion_projection.py 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