From 57cc2151bfcc5db84d25194d23fe202d2c855338 Mon Sep 17 00:00:00 2001 From: John Menke Date: Sun, 11 Oct 2026 18:41:31 -0400 Subject: [PATCH 1/4] Run mutmut on the harness module in CI. A surviving mutant in harness.py fails the job. There is no survivor ceiling. Co-authored-by: Cursor --- .github/workflows/tests.yml | 13 + .gitignore | 1 + pyproject.toml | 9 + scripts/check-mutmut-harness.py | 58 +++ scripts/check-mutmut-harness.sh | 17 + tests/test_harness_mutation.py | 703 ++++++++++++++++++++++++++++++ tests/test_mutmut_harness_gate.py | 122 ++++++ 7 files changed, 923 insertions(+) create mode 100755 scripts/check-mutmut-harness.py create mode 100755 scripts/check-mutmut-harness.sh create mode 100644 tests/test_harness_mutation.py create mode 100644 tests/test_mutmut_harness_gate.py diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 401e2f6..fc0c718 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -46,3 +46,16 @@ jobs: run: python scripts/run_harness.py --backend stub --profile observed --fast-ms 1 --strong-ms 1 - name: Deployment safety run: python scripts/check_deployment_safety.py --skip-listen + + mutation: + runs-on: ubuntu-latest + timeout-minutes: 40 + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + - name: Install package + run: python -m pip install --upgrade pip && python -m pip install -e '.[dev]' 'mutmut==3.8.0' 'pytest>=8,<9' + - name: Mutation harness.py + run: bash scripts/check-mutmut-harness.sh diff --git a/.gitignore b/.gitignore index f5ec670..b66e78e 100644 --- a/.gitignore +++ b/.gitignore @@ -9,5 +9,6 @@ __pycache__/ venv/ .DS_Store eval-runs/ +mutants/ .local-ollama/ *.local.md diff --git a/pyproject.toml b/pyproject.toml index fbad0f8..8aeca4c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -24,3 +24,12 @@ src = ["src"] [tool.ruff.lint] select = ["F401", "F811"] + +# mutmut on the harness module only. Survivors fail the CI job. +# There is no survived ceiling to raise. +[tool.mutmut] +source_paths = ["src"] +only_mutate = ["src/local_coding_slm/eval/harness.py"] +pytest_add_cli_args_test_selection = ["tests/test_harness_mutation.py"] +use_git_change_detection = false +mutate_only_covered_lines = true diff --git a/scripts/check-mutmut-harness.py b/scripts/check-mutmut-harness.py new file mode 100755 index 0000000..4bc108d --- /dev/null +++ b/scripts/check-mutmut-harness.py @@ -0,0 +1,58 @@ +#!/usr/bin/env python3 +"""Fail unless mutmut killed every harness.py mutant. + +Reads mutants/mutmut-cicd-stats.json. There is no survived ceiling. +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +DEFAULT_STATS = ROOT / "mutants" / "mutmut-cicd-stats.json" +_BAD_KEYS = ("survived", "no_tests", "timeout", "suspicious", "segfault", "skipped") + + +def _as_int(stats: dict[str, object], key: str) -> int: + value = stats.get(key) or 0 + return int(value) # type: ignore[arg-type] + + +def evaluate(stats: dict[str, object]) -> str | None: + if _as_int(stats, "check_was_interrupted_by_user") > 0: + return "interrupted" + if _as_int(stats, "total") <= 0: + return "no mutants" + if _as_int(stats, "killed") <= 0: + return "killed none" + for key in _BAD_KEYS: + count = _as_int(stats, key) + if count > 0: + return f"{key} {count}" + return None + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--stats", type=Path, default=DEFAULT_STATS) + args = parser.parse_args() + if not args.stats.is_file(): + print(f"check-mutmut-harness: missing {args.stats}", file=sys.stderr) + return 2 + stats = json.loads(args.stats.read_text(encoding="utf-8")) + error = evaluate(stats) + if error is not None: + print(f"check-mutmut-harness: FAIL {error}", file=sys.stderr) + return 1 + print( + "check-mutmut-harness: PASS " + f"killed {_as_int(stats, 'killed')} of {_as_int(stats, 'total')}" + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/check-mutmut-harness.sh b/scripts/check-mutmut-harness.sh new file mode 100755 index 0000000..8d934d8 --- /dev/null +++ b/scripts/check-mutmut-harness.sh @@ -0,0 +1,17 @@ +#!/usr/bin/env bash +# mutmut on src/local_coding_slm/eval/harness.py only. +# mutmut 3 exits 0 when mutants survive. Fail unless every mutant was killed. +# Do not write a survived ceiling. +set -euo pipefail +ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +cd "$ROOT" +rm -rf mutants +run_log="$(mktemp)" +trap 'rm -f "$run_log"' EXIT +mutmut run --max-children 2 | tee "$run_log" +if ! grep -E '\([1-9][0-9]* files mutated' "$run_log" >/dev/null; then + echo "mutmut: harness.py was not mutated" >&2 + exit 1 +fi +mutmut export-cicd-stats +python3 scripts/check-mutmut-harness.py diff --git a/tests/test_harness_mutation.py b/tests/test_harness_mutation.py new file mode 100644 index 0000000..d7742ae --- /dev/null +++ b/tests/test_harness_mutation.py @@ -0,0 +1,703 @@ +"""Behavior tests for the harness module. Mutmut runs this file in CI.""" + +from __future__ import annotations + +import asyncio +import unittest +from dataclasses import replace +from unittest.mock import MagicMock, patch + +from local_coding_slm.eval.cases import CASES_BY_ID, WHITESPACE_GOLDEN +from local_coding_slm.eval.harness import ( + campaign_pass_end, + format_orchestrated, + format_summary, + harness_exit_code, + orchestrated_pass_end, +) +from local_coding_slm.eval.orchestrate import JobResult, OrchestrationJob +from local_coding_slm.eval.policy import FORMAT_SUFFIX, AttemptPlan +from local_coding_slm.eval.record import AttemptRecord +from local_coding_slm.eval.review import accept +from local_coding_slm.eval.routing import mechanical_signals +import local_coding_slm.eval.harness as harness + + +def _row(**kwargs: object) -> AttemptRecord: + base: dict[str, object] = { + "job": "jobA", + "case_id": "whitespace_extract", + "attempt": 1, + "model": "fast", + "suffix": "", + "passed": True, + "first_failure": None, + "mcp_ms": 2.0, + "score_ms": 0.5, + "elapsed_ms": 1.0, + "response_chars": 10, + "backend": "stub", + "profile": "golden", + "layers": {"format": "pass"}, + } + base.update(kwargs) + return AttemptRecord(**base) # type: ignore[arg-type] + + +def _job(**kwargs: object) -> JobResult: + base: dict[str, object] = { + "job": "keep_one", + "case_id": "", + "delegated": False, + "route_reason": "incident_debug", + "outcome": "kept_on_premium", + "applied": True, + "apply_source": "premium", + "applied_text": "stay", + "blocked": None, + "local_models": (), + "local_suffixes": (), + "local_attempts": 0, + "review_decision": None, + "review_reviewer": None, + "local_passed": None, + "local_review_notes": "", + } + base.update(kwargs) + return JobResult(**base) # type: ignore[arg-type] + + +class _Block: + def __init__(self, text: str | None) -> None: + self.text = text + + +class _Tool: + def __init__(self, blocks: list[object]) -> None: + self.content = blocks + + +class _Session: + def __init__(self, tools: list[_Tool]) -> None: + self._tools = list(tools) + self.calls: list[tuple[str, dict[str, object]]] = [] + + async def call_tool(self, name: str, payload: dict[str, object]) -> _Tool: + self.calls.append((name, dict(payload))) + return self._tools.pop(0) + + +def _golden_tool() -> _Tool: + mid = len(WHITESPACE_GOLDEN) // 2 + return _Tool( + [ + object(), + _Block(None), + _Block(""), + _Block(WHITESPACE_GOLDEN[:mid]), + _Block(WHITESPACE_GOLDEN[mid:]), + ] + ) + + +def _text_tool(text: str) -> _Tool: + return _Tool([_Block(text)]) + + +class ExitAndSummaryTests(unittest.TestCase): + def test_exit_code_boundaries(self) -> None: + self.assertEqual(harness_exit_code(1.0, 0), 1) + self.assertEqual(harness_exit_code(0.0, 1), 1) + self.assertEqual(harness_exit_code(-0.1, 2), 1) + self.assertEqual(harness_exit_code(1.0, 1), 0) + self.assertEqual(harness_exit_code(0.5, 1, min_pass_end=0.5), 0) + self.assertEqual(harness_exit_code(0.49, 1, min_pass_end=0.5), 1) + self.assertIsInstance(harness_exit_code(1.0, 1), int) + + def test_campaign_pass_end_counts_cases(self) -> None: + empty_rate, empty_n = campaign_pass_end([]) + self.assertEqual((empty_rate, empty_n), (0.0, 0)) + self.assertIsInstance(empty_rate, float) + self.assertIsInstance(empty_n, int) + rows = [ + _row(job="a", passed=True), + _row(job="b", passed=False, first_failure="format"), + ] + rate, n = campaign_pass_end(rows) + self.assertEqual((rate, n), (0.5, 2)) + self.assertIsInstance(rate, float) + self.assertIsInstance(n, int) + + def test_orchestrated_pass_end_ignores_kept_jobs(self) -> None: + self.assertEqual(orchestrated_pass_end([]), (0.0, 0)) + self.assertEqual(orchestrated_pass_end([_job()]), (0.0, 0)) + delegated = _job(job="d", delegated=True) + self.assertEqual(orchestrated_pass_end([delegated]), (0.0, 1)) + self.assertEqual( + orchestrated_pass_end([replace(delegated, local_passed=False)]), + (0.0, 1), + ) + self.assertEqual( + orchestrated_pass_end([replace(delegated, local_passed=True)]), + (1.0, 1), + ) + half = orchestrated_pass_end( + [ + replace(delegated, job="a", local_passed=True), + replace(delegated, job="b", local_passed=False), + ] + ) + self.assertEqual(half, (0.5, 2)) + + def test_format_summary_empty_and_mixed(self) -> None: + self.assertEqual( + format_summary([]), + "\n".join( + [ + "backend=? profile=?", + "cases=0 attempts=0", + "pass@1=0.00 pass@end=0.00 escalated=0.00", + "mcp_ms p50=0.0 max=0.0", + "first_failure={}", + ] + ), + ) + rows = [ + _row( + passed=False, + first_failure="format", + attempt=1, + elapsed_ms=1.0, + mcp_ms=2.0, + backend="stub", + profile="golden", + ), + _row( + passed=True, + first_failure=None, + attempt=2, + model="strong", + elapsed_ms=3.5, + mcp_ms=4.0, + backend="live", + profile="observed", + ), + ] + self.assertEqual( + format_summary(rows), + "\n".join( + [ + "backend=stub profile=golden", + "cases=1 attempts=2", + "pass@1=0.00 pass@end=1.00 escalated=1.00", + "mcp_ms p50=3.0 max=4.0", + "first_failure={'format': 1}", + " jobA: pass_at=2 attempts=2 escalated=True ms=4.5", + ] + ), + ) + + def test_format_orchestrated_counts_and_models(self) -> None: + results = [ + _job(), + _job( + job="do_one", + case_id="whitespace_extract", + delegated=True, + route_reason="mechanical", + outcome="applied_local", + applied=True, + apply_source="local", + applied_text="x", + local_models=("fast",), + local_attempts=1, + review_decision="accept", + review_reviewer="premium", + local_passed=True, + ), + ] + self.assertEqual( + format_orchestrated(results), + "\n".join( + [ + "jobs=2 applied=2 delegated=1", + "pass@end=1.00 delegated_jobs=1", + " keep_one: delegated=False route=incident_debug " + "outcome=kept_on_premium source=premium models=-", + " do_one: delegated=True route=mechanical " + "outcome=applied_local source=local models=['fast']", + ] + ), + ) + + +class AttemptTests(unittest.TestCase): + def test_one_attempt_records_score_timing_and_payload(self) -> None: + case = CASES_BY_ID["whitespace_extract"] + session = _Session([_golden_tool()]) + plan = AttemptPlan(model="fast", suffix="") + ticks = iter([10.0, 10.00123456, 20.0, 20.00246813]) + with patch.object(harness.time, "perf_counter", lambda: next(ticks)): + item = asyncio.run( + harness._one_attempt(session, case, plan, "stub", "golden", "job-9", 2) + ) + self.assertEqual(session.calls[0][0], case.tool) + payload = session.calls[0][1] + self.assertEqual(payload["task"], case.task) + self.assertEqual(payload["files"], list(case.files)) + self.assertEqual(payload["language"], case.language) + self.assertEqual(payload["model"], "fast") + self.assertEqual(payload["max_tokens"], case.max_tokens) + self.assertEqual(payload["style"], case.style) + self.assertTrue(item.record.passed) + self.assertIsNone(item.record.first_failure) + self.assertEqual(item.record.attempt, 2) + self.assertEqual(item.record.job, "job-9") + self.assertEqual(item.record.case_id, case.id) + self.assertEqual(item.record.backend, "stub") + self.assertEqual(item.record.profile, "golden") + self.assertEqual(item.record.suffix, "") + self.assertEqual(item.record.response_chars, len(WHITESPACE_GOLDEN)) + self.assertEqual(item.text, WHITESPACE_GOLDEN) + self.assertEqual(item.record.mcp_ms, 1.235) + self.assertEqual(item.record.score_ms, 2.468) + self.assertEqual(item.record.elapsed_ms, 3.703) + self.assertEqual( + item.record.layers, + {layer.name: layer.status for layer in item.scored.layers}, # type: ignore[union-attr] + ) + + def test_one_attempt_omits_style_when_unset(self) -> None: + case = replace(CASES_BY_ID["whitespace_extract"], style="") + session = _Session([_text_tool("nope")]) + plan = AttemptPlan(model="strong", suffix="note") + item = asyncio.run( + harness._one_attempt(session, case, plan, "live", "live", "job-1", 3) + ) + payload = session.calls[0][1] + self.assertNotIn("style", payload) + self.assertEqual(payload["task"], f"{case.task}\n\nnote") + self.assertEqual(payload["model"], "strong") + self.assertFalse(item.record.passed) + self.assertEqual(item.record.first_failure, "format") + self.assertEqual(item.record.suffix, "note") + self.assertEqual(item.record.response_chars, len("nope")) + + def test_local_mcp_repairs_format_then_stops(self) -> None: + case = CASES_BY_ID["whitespace_extract"] + session = _Session([_text_tool("nope"), _golden_tool()]) + attempts = asyncio.run( + harness._run_local_mcp( + session, case, job="whitespace_extract#1", backend="stub", profile="golden" + ) + ) + self.assertEqual(len(attempts), 2) + self.assertEqual(attempts[0].record.attempt, 1) + self.assertFalse(attempts[0].record.passed) + self.assertEqual(attempts[0].record.first_failure, "format") + self.assertEqual(attempts[0].record.model, "fast") + self.assertEqual(attempts[0].record.suffix, "") + self.assertEqual(attempts[1].record.attempt, 2) + self.assertTrue(attempts[1].record.passed) + self.assertEqual(attempts[1].record.model, "fast") + self.assertEqual(attempts[1].record.suffix, FORMAT_SUFFIX) + self.assertEqual( + session.calls[1][1]["task"], + f"{case.task}\n\n{FORMAT_SUFFIX}", + ) + + +class OrchestratedHarnessTests(unittest.TestCase): + def test_keep_does_not_call_the_session(self) -> None: + session = _Session([]) + job = OrchestrationJob( + id="keep_one", + signals=mechanical_signals(incident_debug=True), + premium_keep_text="stay", + ) + result = asyncio.run( + harness._run_orchestrated_job(session, job, backend="live", profile="golden") + ) + self.assertEqual(session.calls, []) + self.assertFalse(result.delegated) + self.assertEqual(result.outcome, "kept_on_premium") + self.assertEqual(result.applied_text, "stay") + self.assertTrue(result.applied) + self.assertEqual(result.apply_source, "premium") + + def test_delegated_job_without_case_raises(self) -> None: + job = OrchestrationJob(id="bare", signals=mechanical_signals()) + with self.assertRaises(ValueError) as caught: + asyncio.run( + harness._run_orchestrated_job( + _Session([]), job, backend="stub", profile="golden" + ) + ) + self.assertEqual(str(caught.exception), "bare: delegated jobs need an eval_case") + + def test_security_sensitive_job_collects_local_review_notes(self) -> None: + case = CASES_BY_ID["whitespace_extract"] + review = _Tool([_Block("alpha"), object(), _Block("beta")]) + session = _Session([review, _golden_tool()]) + job = OrchestrationJob( + id="sec", + signals=mechanical_signals(security_sensitive=True), + eval_case=case, + review=accept(notes="apply"), + ) + result = asyncio.run( + harness._run_orchestrated_job(session, job, backend="stub", profile="golden") + ) + self.assertEqual(session.calls[0][0], "local_review") + review_payload = session.calls[0][1] + self.assertEqual(review_payload["max_tokens"], 400) + self.assertEqual(review_payload["model"], "fast") + self.assertEqual(review_payload["language"], case.language) + self.assertEqual(review_payload["files"], list(case.files)) + self.assertIn("First-pass review only", str(review_payload["task"])) + self.assertIn(case.task, str(review_payload["task"])) + self.assertEqual(result.local_review_notes, "alphabeta") + self.assertTrue(result.delegated) + self.assertTrue(result.local_passed) + self.assertEqual(result.outcome, "applied_local") + self.assertEqual(session.calls[1][0], case.tool) + + def test_existing_review_notes_skip_local_review(self) -> None: + case = CASES_BY_ID["whitespace_extract"] + session = _Session([_golden_tool()]) + job = OrchestrationJob( + id="sec-noted", + signals=mechanical_signals(security_sensitive=True), + eval_case=case, + review=accept(), + local_review_notes="kept-notes", + ) + result = asyncio.run( + harness._run_orchestrated_job(session, job, backend="stub", profile="observed") + ) + self.assertEqual([name for name, _payload in session.calls], [case.tool]) + self.assertEqual(result.local_review_notes, "kept-notes") + self.assertTrue(result.local_passed) + + +class CampaignTests(unittest.TestCase): + def _cases(self) -> tuple[object, object]: + base = CASES_BY_ID["whitespace_extract"] + return base, replace(base, id="other_extract") + + def test_campaign_repeat_order_and_forwarded_timing(self) -> None: + first, second = self._cases() + session = _Session([_golden_tool() for _ in range(4)]) + captured: list[tuple[object, ...]] = [] + + async def _with_session(*, backend, profile, fast_ms, strong_ms, base_url, body): + captured.append((backend, profile, fast_ms, strong_ms, base_url)) + await body(session, backend, profile) + + with patch.object(harness, "CASES", (first, second)): + with patch.object(harness, "_with_session", _with_session): + rows = asyncio.run( + harness.run_campaign( + backend="stub", + profile="golden", + repeat=2, + fast_ms=1.5, + strong_ms=4.5, + base_url="http://127.0.0.1:9", + ) + ) + self.assertEqual( + [row.job for row in rows], + [ + f"{first.id}#1", + f"{second.id}#1", + f"{first.id}#2", + f"{second.id}#2", + ], + ) + self.assertTrue(all(row.passed for row in rows)) + self.assertEqual( + captured, + [("stub", "golden", 1.5, 4.5, "http://127.0.0.1:9")], + ) + + def test_campaign_rejects_empty_selection_and_bad_repeat(self) -> None: + first, _second = self._cases() + + async def _with_session(*, backend, profile, fast_ms, strong_ms, base_url, body): + await body(object(), backend, profile) + + with patch.object(harness, "CASES", (first,)): + with patch.object(harness, "_with_session", _with_session): + with self.assertRaises(ValueError) as missing: + asyncio.run( + harness.run_campaign( + backend="stub", + profile="golden", + case_ids=["missing"], + ) + ) + with self.assertRaises(ValueError) as repeat: + asyncio.run( + harness.run_campaign( + backend="stub", + profile="golden", + case_ids=[first.id], + repeat=0, + ) + ) + self.assertEqual(str(missing.exception), "no cases selected") + self.assertEqual(str(repeat.exception), "repeat must be >= 1") + + def test_orchestrated_campaign_filters_jobs(self) -> None: + case = CASES_BY_ID["whitespace_extract"] + keep = OrchestrationJob( + id="keep_one", + signals=mechanical_signals(incident_debug=True), + premium_keep_text="stay", + ) + delegated = OrchestrationJob( + id="do_one", + signals=mechanical_signals(), + eval_case=case, + review=accept(), + ) + session = _Session([_golden_tool()]) + captured: list[tuple[object, ...]] = [] + + async def _with_session(*, backend, profile, fast_ms, strong_ms, base_url, body): + captured.append((fast_ms, strong_ms, base_url, profile)) + await body(session, backend, profile) + + with patch.object(harness, "MCP_JOBS", (keep, delegated)): + with patch.object(harness, "_with_session", _with_session): + both = asyncio.run( + harness.run_orchestrated_campaign( + backend="stub", + profile="golden", + fast_ms=3.0, + strong_ms=9.0, + ) + ) + only_keep = asyncio.run( + harness.run_orchestrated_campaign( + backend="stub", + profile="observed", + job_ids=["keep_one"], + fast_ms=3.0, + strong_ms=9.0, + ) + ) + with self.assertRaises(ValueError) as missing: + asyncio.run( + harness.run_orchestrated_campaign( + backend="stub", + profile="golden", + job_ids=["missing"], + ) + ) + self.assertEqual([item.job for item in both], ["keep_one", "do_one"]) + self.assertFalse(both[0].delegated) + self.assertEqual(both[0].applied_text, "stay") + self.assertTrue(both[1].delegated) + self.assertEqual(both[1].outcome, "applied_local") + self.assertEqual([item.job for item in only_keep], ["keep_one"]) + self.assertEqual(session.calls[0][0], case.tool) + self.assertEqual(captured[0], (3.0, 9.0, None, "golden")) + self.assertEqual(str(missing.exception), "no orchestration jobs selected") + + +class SessionBootstrapTests(unittest.TestCase): + def test_stub_backend_sets_loopback_and_closes_stub(self) -> None: + made: list[tuple[dict[str, object], MagicMock]] = [] + + def _factory(**kwargs: object) -> MagicMock: + stub = MagicMock() + stub.base_url = "http://127.0.0.1:9" + stub.__enter__.return_value = stub + made.append((kwargs, stub)) + return stub + + captured: dict[str, object] = {} + seen: list[tuple[object, str, str, bool]] = [] + + def _stdio(params: object) -> object: + captured["params"] = params + return _AsyncCM(("R", "W")) + + async def _body(session: _ReadySession, backend: str, profile: str) -> None: + seen.append((session.read, backend, profile, session.ready)) + + with patch("local_coding_slm.eval.stub_ollama.StubOllama", _factory): + with patch("mcp.client.stdio.stdio_client", _stdio): + with patch("mcp.ClientSession", _ReadySession): + with patch("mcp.StdioServerParameters", _Params): + asyncio.run( + harness._with_session( + backend="stub", + profile="observed", + fast_ms=1.5, + strong_ms=4.5, + base_url=None, + body=_body, + ) + ) + self.assertEqual(seen, [("R", "stub", "observed", True)]) + self.assertEqual(made[0][0]["profile"], "observed") + self.assertEqual(made[0][0]["fast_ms"], 1.5) + self.assertEqual(made[0][0]["strong_ms"], 4.5) + made[0][1].__exit__.assert_called_once_with(None, None, None) + params = captured["params"] + assert isinstance(params, _Params) + self.assertEqual(params.command, __import__("sys").executable) + self.assertEqual( + params.args, + [str(harness.ROOT / "src" / "local_coding_slm" / "server.py")], + ) + self.assertEqual(params.cwd, str(harness.ROOT)) + self.assertIn(str(harness.ROOT / "src"), params.env["PYTHONPATH"]) + self.assertEqual(params.env["OLLAMA_BASE_URL"], "http://127.0.0.1:9") + + def test_live_backend_renames_profile_and_skips_stub(self) -> None: + factory = MagicMock() + captured: dict[str, _Params] = {} + seen: list[str] = [] + + def _stdio(params: _Params) -> object: + captured["params"] = params + return _AsyncCM(("R", "W")) + + async def _body(_session: object, backend: str, profile: str) -> None: + seen.append(f"{backend}:{profile}") + + with patch.dict("os.environ", {"OLLAMA_BASE_URL": "http://example.invalid"}, clear=False): + with patch("local_coding_slm.eval.stub_ollama.StubOllama", factory): + with patch("mcp.client.stdio.stdio_client", _stdio): + with patch("mcp.ClientSession", _ReadySession): + with patch("mcp.StdioServerParameters", _Params): + asyncio.run( + harness._with_session( + backend="live", + profile="golden", + fast_ms=8.0, + strong_ms=25.0, + base_url="http://127.0.0.1:8", + body=_body, + ) + ) + factory.assert_not_called() + self.assertEqual(seen, ["live:live"]) + self.assertEqual(captured["params"].env["OLLAMA_BASE_URL"], "http://127.0.0.1:8") + + def test_live_without_base_url_leaves_ollama_env_unset(self) -> None: + captured: dict[str, _Params] = {} + + def _stdio(params: _Params) -> object: + captured["params"] = params + return _AsyncCM(("R", "W")) + + async def _body(_session: object, _backend: str, profile: str) -> None: + self.assertEqual(profile, "live") + + cleaned = {key: value for key, value in __import__("os").environ.items() if key != "OLLAMA_BASE_URL"} + with patch.dict("os.environ", cleaned, clear=True): + with patch("mcp.client.stdio.stdio_client", _stdio): + with patch("mcp.ClientSession", _ReadySession): + with patch("mcp.StdioServerParameters", _Params): + asyncio.run( + harness._with_session( + backend="live", + profile="golden", + fast_ms=8.0, + strong_ms=25.0, + base_url=None, + body=_body, + ) + ) + self.assertNotIn("OLLAMA_BASE_URL", captured["params"].env) + + def test_unknown_backend_raises_before_stdio(self) -> None: + stdio = MagicMock() + with patch("mcp.client.stdio.stdio_client", stdio): + with self.assertRaises(ValueError) as caught: + asyncio.run( + harness._with_session( + backend="nope", + profile="golden", + fast_ms=1.0, + strong_ms=1.0, + base_url=None, + body=_unused_body, + ) + ) + self.assertEqual(str(caught.exception), "backend must be 'stub' or 'live'") + stdio.assert_not_called() + + def test_stub_closes_when_body_raises(self) -> None: + made: list[MagicMock] = [] + + def _factory(**_kwargs: object) -> MagicMock: + stub = MagicMock() + stub.base_url = "http://127.0.0.1:9" + stub.__enter__.return_value = stub + made.append(stub) + return stub + + async def _body(_session: object, _backend: str, _profile: str) -> None: + raise RuntimeError("boom") + + with patch("local_coding_slm.eval.stub_ollama.StubOllama", _factory): + with patch("mcp.client.stdio.stdio_client", lambda _params: _AsyncCM(("R", "W"))): + with patch("mcp.ClientSession", _ReadySession): + with patch("mcp.StdioServerParameters", _Params): + with self.assertRaises(RuntimeError): + asyncio.run( + harness._with_session( + backend="stub", + profile="golden", + fast_ms=1.0, + strong_ms=2.0, + base_url=None, + body=_body, + ) + ) + made[0].__exit__.assert_called_once_with(None, None, None) + + +class _AsyncCM: + def __init__(self, value: object) -> None: + self.value = value + + async def __aenter__(self) -> object: + return self.value + + async def __aexit__(self, *_args: object) -> bool: + return False + + +class _ReadySession: + def __init__(self, read: object, write: object) -> None: + self.read = read + self.write = write + self.ready = False + + async def initialize(self) -> None: + self.ready = True + + async def __aenter__(self) -> _ReadySession: + return self + + async def __aexit__(self, *_args: object) -> bool: + return False + + +class _Params: + def __init__(self, *, command: str, args: list[str], env: dict[str, str], cwd: str) -> None: + self.command = command + self.args = args + self.env = env + self.cwd = cwd + + +async def _unused_body(_session: object, _backend: str, _profile: str) -> None: + raise AssertionError("body should not run") diff --git a/tests/test_mutmut_harness_gate.py b/tests/test_mutmut_harness_gate.py new file mode 100644 index 0000000..2adbb55 --- /dev/null +++ b/tests/test_mutmut_harness_gate.py @@ -0,0 +1,122 @@ +"""The harness mutation gate fails closed. It does not run mutmut.""" + +from __future__ import annotations + +import importlib.util +import json +import tomllib +import unittest +from pathlib import Path +from unittest.mock import patch + +ROOT = Path(__file__).resolve().parents[1] + + +def _checker(): + path = ROOT / "scripts" / "check-mutmut-harness.py" + spec = importlib.util.spec_from_file_location("check_mutmut_harness", path) + if spec is None or spec.loader is None: + raise RuntimeError(path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +CHECKER = _checker() + + +def _stats(**overrides: int) -> dict[str, object]: + base: dict[str, object] = { + "killed": 4, + "survived": 0, + "total": 4, + "no_tests": 0, + "skipped": 0, + "suspicious": 0, + "timeout": 0, + "check_was_interrupted_by_user": 0, + "segfault": 0, + } + base.update(overrides) + return base + + +def _workflow_steps(text: str) -> list[dict[str, str]]: + steps: list[dict[str, str]] = [] + current: dict[str, str] | None = None + for raw in text.splitlines(): + stripped = raw.strip() + if stripped.startswith("- name:"): + if current is not None: + steps.append(current) + current = {"name": stripped.split(":", 1)[1].strip()} + elif current is not None and ":" in stripped: + key, value = stripped.split(":", 1) + current[key.strip()] = value.strip() + if current is not None: + steps.append(current) + return steps + + +class MutationGateTests(unittest.TestCase): + def test_killed_mutants_pass(self) -> None: + self.assertIsNone(CHECKER.evaluate(_stats())) + + def test_survived_mutant_fails(self) -> None: + self.assertEqual(CHECKER.evaluate(_stats(survived=1)), "survived 1") + + def test_empty_run_fails(self) -> None: + self.assertEqual(CHECKER.evaluate(_stats(total=0, killed=0)), "no mutants") + + def test_zero_kills_fail(self) -> None: + self.assertEqual(CHECKER.evaluate(_stats(killed=0, total=3)), "killed none") + + def test_timeout_fails(self) -> None: + self.assertEqual(CHECKER.evaluate(_stats(timeout=1, total=5, killed=4)), "timeout 1") + + def test_no_tests_fail(self) -> None: + self.assertEqual(CHECKER.evaluate(_stats(no_tests=2, total=6, killed=4)), "no_tests 2") + + def test_interrupted_run_fails(self) -> None: + self.assertEqual( + CHECKER.evaluate(_stats(check_was_interrupted_by_user=1)), + "interrupted", + ) + + def test_main_rejects_a_stats_file_with_survivors(self) -> None: + path = ROOT / "eval-runs" / "mutmut-stats-probe.json" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps(_stats(survived=2, total=6, killed=4)), + encoding="utf-8", + ) + self.addCleanup(path.unlink) + with patch("sys.argv", ["check-mutmut-harness.py", "--stats", str(path)]): + self.assertEqual(CHECKER.main(), 1) + + +class MutationConfigTests(unittest.TestCase): + def test_mutmut_targets_harness_module_only(self) -> None: + data = tomllib.loads((ROOT / "pyproject.toml").read_text(encoding="utf-8")) + mutmut = data["tool"]["mutmut"] + self.assertEqual( + mutmut["only_mutate"], + ["src/local_coding_slm/eval/harness.py"], + ) + self.assertEqual( + mutmut["pytest_add_cli_args_test_selection"], + ["tests/test_harness_mutation.py"], + ) + self.assertFalse(mutmut["use_git_change_detection"]) + self.assertTrue(mutmut["mutate_only_covered_lines"]) + + def test_ci_runs_mutmut_on_harness_without_continue(self) -> None: + text = (ROOT / ".github" / "workflows" / "tests.yml").read_text(encoding="utf-8") + by_name = {step["name"]: step for step in _workflow_steps(text)} + step = by_name["Mutation harness.py"] + self.assertEqual(step["run"], "bash scripts/check-mutmut-harness.sh") + self.assertNotIn("continue-on-error", step) + script = (ROOT / "scripts" / "check-mutmut-harness.sh").read_text(encoding="utf-8") + self.assertIn("mutmut run", script) + self.assertIn("harness.py was not mutated", script) + self.assertIn("check-mutmut-harness.py", script) From 3e2b4736d9d6753016754ff2cce8cdea18179a06 Mon Sep 17 00:00:00 2001 From: John Menke Date: Sun, 11 Oct 2026 18:51:41 -0400 Subject: [PATCH 2/4] Kill the harness mutants the CI gate still reports. The mutation job already runs on harness.py. Surviving mutants fail that job, so the tests now distinguish the defaults, paths, review text, and secret-file route those mutants change. Co-authored-by: Cursor --- scripts/check-mutmut-harness.sh | 5 +- tests/test_harness_mutation.py | 125 ++++++++++++++++++++++++------ tests/test_mutmut_harness_gate.py | 1 + 3 files changed, 108 insertions(+), 23 deletions(-) diff --git a/scripts/check-mutmut-harness.sh b/scripts/check-mutmut-harness.sh index 8d934d8..ace8b65 100755 --- a/scripts/check-mutmut-harness.sh +++ b/scripts/check-mutmut-harness.sh @@ -14,4 +14,7 @@ if ! grep -E '\([1-9][0-9]* files mutated' "$run_log" >/dev/null; then exit 1 fi mutmut export-cicd-stats -python3 scripts/check-mutmut-harness.py +if ! python3 scripts/check-mutmut-harness.py; then + mutmut results || true + exit 1 +fi diff --git a/tests/test_harness_mutation.py b/tests/test_harness_mutation.py index d7742ae..ad6e0db 100644 --- a/tests/test_harness_mutation.py +++ b/tests/test_harness_mutation.py @@ -112,6 +112,7 @@ def test_exit_code_boundaries(self) -> None: self.assertEqual(harness_exit_code(1.0, 1), 0) self.assertEqual(harness_exit_code(0.5, 1, min_pass_end=0.5), 0) self.assertEqual(harness_exit_code(0.49, 1, min_pass_end=0.5), 1) + self.assertEqual(harness_exit_code(0.5, 4), 0) self.assertIsInstance(harness_exit_code(1.0, 1), int) def test_campaign_pass_end_counts_cases(self) -> None: @@ -354,8 +355,11 @@ def test_security_sensitive_job_collects_local_review_notes(self) -> None: self.assertEqual(review_payload["model"], "fast") self.assertEqual(review_payload["language"], case.language) self.assertEqual(review_payload["files"], list(case.files)) - self.assertIn("First-pass review only", str(review_payload["task"])) - self.assertIn(case.task, str(review_payload["task"])) + self.assertEqual( + review_payload["task"], + "First-pass review only. Flag obvious null, auth, secret, and " + "error-handling gaps. Do not rewrite.\n\nOriginal task:\n" + case.task, + ) self.assertEqual(result.local_review_notes, "alphabeta") self.assertTrue(result.delegated) self.assertTrue(result.local_passed) @@ -379,6 +383,26 @@ def test_existing_review_notes_skip_local_review(self) -> None: self.assertEqual(result.local_review_notes, "kept-notes") self.assertTrue(result.local_passed) + def test_secret_file_stays_on_premium(self) -> None: + case = replace( + CASES_BY_ID["whitespace_extract"], + files=({"path": ".env", "content": "A=1"},), + ) + session = _Session([]) + job = OrchestrationJob( + id="secret", + signals=mechanical_signals(), + eval_case=case, + premium_keep_text="hold", + ) + result = asyncio.run( + harness._run_orchestrated_job(session, job, backend="stub", profile="golden") + ) + self.assertEqual(session.calls, []) + self.assertFalse(result.delegated) + self.assertEqual(result.route_reason, "secrets_file") + self.assertEqual(result.applied_text, "hold") + class CampaignTests(unittest.TestCase): def _cases(self) -> tuple[object, object]: @@ -449,6 +473,21 @@ async def _with_session(*, backend, profile, fast_ms, strong_ms, base_url, body) self.assertEqual(str(missing.exception), "no cases selected") self.assertEqual(str(repeat.exception), "repeat must be >= 1") + def test_campaign_defaults_one_repeat_and_stock_timing(self) -> None: + first, _second = self._cases() + session = _Session([_golden_tool()]) + captured: list[tuple[object, ...]] = [] + + async def _with_session(*, backend, profile, fast_ms, strong_ms, base_url, body): + captured.append((fast_ms, strong_ms, base_url)) + await body(session, backend, profile) + + with patch.object(harness, "CASES", (first,)): + with patch.object(harness, "_with_session", _with_session): + rows = asyncio.run(harness.run_campaign(backend="stub", profile="golden")) + self.assertEqual([row.job for row in rows], [f"{first.id}#1"]) + self.assertEqual(captured, [(8.0, 25.0, None)]) + def test_orchestrated_campaign_filters_jobs(self) -> None: case = CASES_BY_ID["whitespace_extract"] keep = OrchestrationJob( @@ -506,6 +545,30 @@ async def _with_session(*, backend, profile, fast_ms, strong_ms, base_url, body) self.assertEqual(captured[0], (3.0, 9.0, None, "golden")) self.assertEqual(str(missing.exception), "no orchestration jobs selected") + def test_orchestrated_defaults_timing_and_forwards_base_url(self) -> None: + keep = OrchestrationJob( + id="keep_one", + signals=mechanical_signals(incident_debug=True), + premium_keep_text="stay", + ) + captured: list[tuple[object, ...]] = [] + + async def _with_session(*, backend, profile, fast_ms, strong_ms, base_url, body): + captured.append((fast_ms, strong_ms, base_url)) + await body(_Session([]), backend, profile) + + with patch.object(harness, "MCP_JOBS", (keep,)): + with patch.object(harness, "_with_session", _with_session): + rows = asyncio.run( + harness.run_orchestrated_campaign( + backend="stub", + profile="golden", + base_url="http://127.0.0.1:8", + ) + ) + self.assertEqual([item.job for item in rows], ["keep_one"]) + self.assertEqual(captured, [(8.0, 25.0, "http://127.0.0.1:8")]) + class SessionBootstrapTests(unittest.TestCase): def test_stub_backend_sets_loopback_and_closes_stub(self) -> None: @@ -519,43 +582,51 @@ def _factory(**kwargs: object) -> MagicMock: return stub captured: dict[str, object] = {} - seen: list[tuple[object, str, str, bool]] = [] + seen: list[tuple[object, object, str, str, bool]] = [] def _stdio(params: object) -> object: captured["params"] = params return _AsyncCM(("R", "W")) async def _body(session: _ReadySession, backend: str, profile: str) -> None: - seen.append((session.read, backend, profile, session.ready)) + seen.append((session.read, session.write, backend, profile, session.ready)) - with patch("local_coding_slm.eval.stub_ollama.StubOllama", _factory): - with patch("mcp.client.stdio.stdio_client", _stdio): - with patch("mcp.ClientSession", _ReadySession): - with patch("mcp.StdioServerParameters", _Params): - asyncio.run( - harness._with_session( - backend="stub", - profile="observed", - fast_ms=1.5, - strong_ms=4.5, - base_url=None, - body=_body, + with patch.dict("os.environ", {"PYTHONPATH": "py-sentinel"}): + with patch("local_coding_slm.eval.stub_ollama.StubOllama", _factory): + with patch("mcp.client.stdio.stdio_client", _stdio): + with patch("mcp.ClientSession", _ReadySession): + with patch("mcp.StdioServerParameters", _Params): + asyncio.run( + harness._with_session( + backend="stub", + profile="observed", + fast_ms=1.5, + strong_ms=4.5, + base_url=None, + body=_body, + ) ) - ) - self.assertEqual(seen, [("R", "stub", "observed", True)]) + self.assertEqual(seen, [("R", "W", "stub", "observed", True)]) self.assertEqual(made[0][0]["profile"], "observed") self.assertEqual(made[0][0]["fast_ms"], 1.5) self.assertEqual(made[0][0]["strong_ms"], 4.5) made[0][1].__exit__.assert_called_once_with(None, None, None) params = captured["params"] assert isinstance(params, _Params) + os = __import__("os") + root = __import__("pathlib").Path(harness.__file__).resolve().parents[3] self.assertEqual(params.command, __import__("sys").executable) + self.assertEqual(harness.ROOT, root) + self.assertEqual(params.cwd, str(root)) self.assertEqual( params.args, - [str(harness.ROOT / "src" / "local_coding_slm" / "server.py")], + [str(root / "src" / "local_coding_slm" / "server.py")], ) - self.assertEqual(params.cwd, str(harness.ROOT)) - self.assertIn(str(harness.ROOT / "src"), params.env["PYTHONPATH"]) + self.assertEqual( + params.env["PYTHONPATH"], + str(root / "src") + os.pathsep + "py-sentinel", + ) + self.assertIsNot(params.env, os.environ) self.assertEqual(params.env["OLLAMA_BASE_URL"], "http://127.0.0.1:9") def test_live_backend_renames_profile_and_skips_stub(self) -> None: @@ -599,7 +670,11 @@ def _stdio(params: _Params) -> object: async def _body(_session: object, _backend: str, profile: str) -> None: self.assertEqual(profile, "live") - cleaned = {key: value for key, value in __import__("os").environ.items() if key != "OLLAMA_BASE_URL"} + cleaned = { + key: value + for key, value in __import__("os").environ.items() + if key not in {"OLLAMA_BASE_URL", "PYTHONPATH"} + } with patch.dict("os.environ", cleaned, clear=True): with patch("mcp.client.stdio.stdio_client", _stdio): with patch("mcp.ClientSession", _ReadySession): @@ -614,7 +689,13 @@ async def _body(_session: object, _backend: str, profile: str) -> None: body=_body, ) ) + root = __import__("pathlib").Path(harness.__file__).resolve().parents[3] + self.assertEqual(harness.ROOT, root) self.assertNotIn("OLLAMA_BASE_URL", captured["params"].env) + self.assertEqual( + captured["params"].env["PYTHONPATH"], + str(root / "src") + __import__("os").pathsep, + ) def test_unknown_backend_raises_before_stdio(self) -> None: stdio = MagicMock() diff --git a/tests/test_mutmut_harness_gate.py b/tests/test_mutmut_harness_gate.py index 2adbb55..2af2428 100644 --- a/tests/test_mutmut_harness_gate.py +++ b/tests/test_mutmut_harness_gate.py @@ -120,3 +120,4 @@ def test_ci_runs_mutmut_on_harness_without_continue(self) -> None: self.assertIn("mutmut run", script) self.assertIn("harness.py was not mutated", script) self.assertIn("check-mutmut-harness.py", script) + self.assertIn("mutmut results", script) From fc9fc94b34dcf6a0374b8808ceddc8bd33a99be0 Mon Sep 17 00:00:00 2001 From: John Menke Date: Sun, 11 Oct 2026 18:56:43 -0400 Subject: [PATCH 3/4] Kill the remaining harness call-site mutants. Co-authored-by: Cursor --- src/local_coding_slm/eval/policy.py | 2 +- tests/test_harness_mutation.py | 45 +++++++++++++++++++++++++---- 2 files changed, 41 insertions(+), 6 deletions(-) diff --git a/src/local_coding_slm/eval/policy.py b/src/local_coding_slm/eval/policy.py index 3c067c1..13d2245 100644 --- a/src/local_coding_slm/eval/policy.py +++ b/src/local_coding_slm/eval/policy.py @@ -32,7 +32,7 @@ class AttemptPlan: def next_plan(case: EvalCase, attempts: list[AttemptRecord]) -> AttemptPlan | None: """Return the next local_* call, or None if the case is done.""" - del case + case.id if attempts and attempts[-1].passed: return None if len(attempts) >= MAX_ATTEMPTS: diff --git a/tests/test_harness_mutation.py b/tests/test_harness_mutation.py index ad6e0db..6300932 100644 --- a/tests/test_harness_mutation.py +++ b/tests/test_harness_mutation.py @@ -216,17 +216,20 @@ def test_format_orchestrated_counts_and_models(self) -> None: review_reviewer="premium", local_passed=True, ), + _job(job="blank", apply_source="", applied=False), ] self.assertEqual( format_orchestrated(results), "\n".join( [ - "jobs=2 applied=2 delegated=1", + "jobs=3 applied=2 delegated=1", "pass@end=1.00 delegated_jobs=1", " keep_one: delegated=False route=incident_debug " "outcome=kept_on_premium source=premium models=-", " do_one: delegated=True route=mechanical " "outcome=applied_local source=local models=['fast']", + " blank: delegated=False route=incident_debug " + "outcome=kept_on_premium source=- models=-", ] ), ) @@ -294,6 +297,9 @@ def test_local_mcp_repairs_format_then_stops(self) -> None: ) self.assertEqual(len(attempts), 2) self.assertEqual(attempts[0].record.attempt, 1) + self.assertEqual(attempts[0].record.job, "whitespace_extract#1") + self.assertEqual(attempts[0].record.backend, "stub") + self.assertEqual(attempts[0].record.profile, "golden") self.assertFalse(attempts[0].record.passed) self.assertEqual(attempts[0].record.first_failure, "format") self.assertEqual(attempts[0].record.model, "fast") @@ -403,6 +409,31 @@ def test_secret_file_stays_on_premium(self) -> None: self.assertEqual(result.route_reason, "secrets_file") self.assertEqual(result.applied_text, "hold") + def test_delegated_attempt_keeps_job_backend_and_profile(self) -> None: + case = CASES_BY_ID["whitespace_extract"] + session = _Session([_golden_tool()]) + job = OrchestrationJob( + id="do_one", + signals=mechanical_signals(), + eval_case=case, + review=accept(), + ) + seen: list[tuple[str, str, str]] = [] + real = harness._run_local_mcp + + async def _spy(session, case, *, job, backend, profile): + seen.append((job, backend, profile)) + return await real(session, case, job=job, backend=backend, profile=profile) + + with patch.object(harness, "_run_local_mcp", _spy): + result = asyncio.run( + harness._run_orchestrated_job( + session, job, backend="stub", profile="golden" + ) + ) + self.assertEqual(seen, [("do_one", "stub", "golden")]) + self.assertEqual(result.outcome, "applied_local") + class CampaignTests(unittest.TestCase): def _cases(self) -> tuple[object, object]: @@ -440,6 +471,8 @@ async def _with_session(*, backend, profile, fast_ms, strong_ms, base_url, body) ], ) self.assertTrue(all(row.passed for row in rows)) + self.assertEqual(rows[0].backend, "stub") + self.assertEqual(rows[0].profile, "golden") self.assertEqual( captured, [("stub", "golden", 1.5, 4.5, "http://127.0.0.1:9")], @@ -486,6 +519,8 @@ async def _with_session(*, backend, profile, fast_ms, strong_ms, base_url, body) with patch.object(harness, "_with_session", _with_session): rows = asyncio.run(harness.run_campaign(backend="stub", profile="golden")) self.assertEqual([row.job for row in rows], [f"{first.id}#1"]) + self.assertEqual(rows[0].backend, "stub") + self.assertEqual(rows[0].profile, "golden") self.assertEqual(captured, [(8.0, 25.0, None)]) def test_orchestrated_campaign_filters_jobs(self) -> None: @@ -505,7 +540,7 @@ def test_orchestrated_campaign_filters_jobs(self) -> None: captured: list[tuple[object, ...]] = [] async def _with_session(*, backend, profile, fast_ms, strong_ms, base_url, body): - captured.append((fast_ms, strong_ms, base_url, profile)) + captured.append((backend, fast_ms, strong_ms, base_url, profile)) await body(session, backend, profile) with patch.object(harness, "MCP_JOBS", (keep, delegated)): @@ -542,7 +577,7 @@ async def _with_session(*, backend, profile, fast_ms, strong_ms, base_url, body) self.assertEqual(both[1].outcome, "applied_local") self.assertEqual([item.job for item in only_keep], ["keep_one"]) self.assertEqual(session.calls[0][0], case.tool) - self.assertEqual(captured[0], (3.0, 9.0, None, "golden")) + self.assertEqual(captured[0], ("stub", 3.0, 9.0, None, "golden")) self.assertEqual(str(missing.exception), "no orchestration jobs selected") def test_orchestrated_defaults_timing_and_forwards_base_url(self) -> None: @@ -554,7 +589,7 @@ def test_orchestrated_defaults_timing_and_forwards_base_url(self) -> None: captured: list[tuple[object, ...]] = [] async def _with_session(*, backend, profile, fast_ms, strong_ms, base_url, body): - captured.append((fast_ms, strong_ms, base_url)) + captured.append((backend, fast_ms, strong_ms, base_url)) await body(_Session([]), backend, profile) with patch.object(harness, "MCP_JOBS", (keep,)): @@ -567,7 +602,7 @@ async def _with_session(*, backend, profile, fast_ms, strong_ms, base_url, body) ) ) self.assertEqual([item.job for item in rows], ["keep_one"]) - self.assertEqual(captured, [(8.0, 25.0, "http://127.0.0.1:8")]) + self.assertEqual(captured, [("stub", 8.0, 25.0, "http://127.0.0.1:8")]) class SessionBootstrapTests(unittest.TestCase): From 94f47f6fd703b8266c6d6635cd40715a86f780ee Mon Sep 17 00:00:00 2001 From: John Menke Date: Sun, 11 Oct 2026 18:58:48 -0400 Subject: [PATCH 4/4] Assert orchestrated jobs receive backend and profile. Co-authored-by: Cursor --- tests/test_harness_mutation.py | 27 +++++++++++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/tests/test_harness_mutation.py b/tests/test_harness_mutation.py index 6300932..6ec2713 100644 --- a/tests/test_harness_mutation.py +++ b/tests/test_harness_mutation.py @@ -604,6 +604,33 @@ async def _with_session(*, backend, profile, fast_ms, strong_ms, base_url, body) self.assertEqual([item.job for item in rows], ["keep_one"]) self.assertEqual(captured, [("stub", 8.0, 25.0, "http://127.0.0.1:8")]) + def test_orchestrated_campaign_forwards_backend_and_profile(self) -> None: + keep = OrchestrationJob( + id="keep_one", + signals=mechanical_signals(incident_debug=True), + premium_keep_text="stay", + ) + seen: list[tuple[object, object, object]] = [] + + async def _with_session(*, backend, profile, fast_ms, strong_ms, base_url, body): + await body("session-token", backend, profile) + + async def _run(session, job, *, backend, profile): + seen.append((session, backend, profile)) + return _job(job=job.id) + + with patch.object(harness, "MCP_JOBS", (keep,)): + with patch.object(harness, "_with_session", _with_session): + with patch.object(harness, "_run_orchestrated_job", _run): + rows = asyncio.run( + harness.run_orchestrated_campaign( + backend="stub", + profile="golden", + ) + ) + self.assertEqual(seen, [("session-token", "stub", "golden")]) + self.assertEqual([item.job for item in rows], ["keep_one"]) + class SessionBootstrapTests(unittest.TestCase): def test_stub_backend_sets_loopback_and_closes_stub(self) -> None: