"""Mutation harness: break a control, prove the F4 suite goes red. Every entry in MUTATIONS names one control that F4 depends on. Each one is applied to a working-tree file, the F4 suite is run, and the file is restored. The report records whether the suite could actually fail. Run from the repo root: python3 scripts/f4-mutation-check.py This is a TOOL, not a test. `pytest` does not collect it and it is not part of any CI path - deliberately, because 26 controlled mutations would multiply the suite runtime. It stays in the repo because the evidence it produces is the only thing that shows a passing test can still die. """ from __future__ import annotations import subprocess import sys from pathlib import Path # Resolve the repo root by looking for the marker, not by counting parents. A # fixed `parent` used to be correct only while this file sat in the repo root; # after the move to `scripts/` it would have pointed every mutation at a # non-existent path, so all 26 would SKIP with "anchor not found" and the run # would report zero blind tests while testing nothing. That is a false green, # which is worse than no harness at all. REPO = next( (p for p in Path(__file__).resolve().parents if (p / "src/cyberbrain26").is_dir()), None, ) if REPO is None: raise SystemExit( "cannot locate the repo root: no parent directory contains src/cyberbrain26. " "Refusing to run - every mutation would skip and the report would be a " "false green." ) ROUTER = REPO / "src/cyberbrain26/model_layer/router.py" EGRESS = REPO / "src/cyberbrain26/control_plane/model_gateway/model_egress.py" KILO = REPO / "src/cyberbrain26/model_layer/kilo_config.py" ROLE = REPO / "src/cyberbrain26/model_layer/role_assignment.py" GATEWAY = REPO / "src/cyberbrain26/control_plane/model_gateway/__init__.py" MUTATIONS: list[tuple[str, Path, str, str]] = [ ( "M1 egress: authorize() not called", EGRESS, " self._egress_gate.authorize(request)", " pass # MUTATION: gate not consulted", ), ( "M2 egress: missing-gate fail-closed removed", EGRESS, ' if self._egress_gate is None:\n' ' raise SecurityError("egress gate not configured; refusing to fail open")', " if self._egress_gate is None:\n return # MUTATION: fail open", ), ( "M3 router: _get_candidates free filter removed", ROUTER, " if free_only:\n" " candidates = [m for m in candidates if self._is_free_eligible(m)]\n\n" " if capabilities:\n" " filtered = []\n" " for m in candidates:\n" " match = True\n" " for cap, required in capabilities.items():\n" " if required:\n" " actual = getattr(m.capabilities, cap, False)", " if capabilities:\n" " filtered = []\n" " for m in candidates:\n" " match = True\n" " for cap, required in capabilities.items():\n" " if required:\n" " actual = getattr(m.capabilities, cap, False)", ), ( "M4 router: _get_fallback_candidates free filter removed", ROUTER, " if free_only:\n" " candidates = [m for m in candidates if self._is_free_eligible(m)]\n\n" " if capabilities:\n" " filtered = []\n" " for m in candidates:\n" " match = True\n" " for cap, required in capabilities.items():\n" " if required:\n" " actual = getattr(m.capabilities, cap, False)", " if capabilities:\n" " filtered = []\n" " for m in candidates:\n" " match = True\n" " for cap, required in capabilities.items():\n" " if required:\n" " actual = getattr(m.capabilities, cap, False)", ), ( "M5 router: _route_specific_model free check removed", ROUTER, " if free_only and not self._is_free_eligible(model):", " if False: # MUTATION: paid specific model allowed", ), ( "M6 router: free_only default flipped to True", ROUTER, " strategy: RouteStrategy | None = None,\n free_only: bool = False,", " strategy: RouteStrategy | None = None,\n free_only: bool = True,", ), ( "M7 kilo_config: no-free-model guard removed", KILO, " if not models:\n" " raise NoFreeModelAvailable(", " if False:\n" " raise NoFreeModelAvailable(", ), ( "M8 kilo_config: validation gate removed", KILO, " if problems:\n" " logger.error(\"refusing to apply Kilo config: %s\", \"; \".join(problems))\n" " return {\"applied\": False, \"problems\": problems}", " if False: # MUTATION: apply unvalidated\n" " logger.error(\"refusing to apply Kilo config: %s\", \"; \".join(problems))\n" " return {\"applied\": False, \"problems\": problems}", ), ( "M9 kilo_config: _is_free_eligible always True", KILO, " return bool(model.is_free)", " return True # MUTATION: everything looks free", ), ( "M10 role_assignment: backup free filter removed", ROLE, " if backup.verified and backup.is_free:", " if True: # MUTATION: paid backup accepted", ), ( "M11 router: _is_free_eligible always True", ROUTER, " if model.free_status.value in (\"free\", \"free_for_test\", \"free_for_study\"):\n" " return True", " return True # MUTATION: everything looks free", ), ( "M12 egress: empty-classification check removed", EGRESS, " if not classifications:\n" " raise SecurityError(\n" " \"Model egress requires an explicit classification; refusing to fail open\"\n" " )\n" " # Log the authorization attempt", " # MUTATION: unclassified call allowed\n" " # Log the authorization attempt", ), ( "M13 egress: digest replaced by empty string", EGRESS, " payload_sha256=hashlib.sha256(payload_bytes).hexdigest(),", " payload_sha256=\"\", # MUTATION: digest not computed", ), ( "M14 egress: raw payload written to the log", EGRESS, " classifications,\n )\n if self._egress_gate is None:", " classifications,\n getattr(self, \"_raw_payload\", \"LEAK\"),\n )\n if self._egress_gate is None:", ), ( "M15 gateway: chat skips egress authorization", GATEWAY, " self._authorize_egress(route.provider, model, messages, classifications)", " pass # MUTATION: chat never authorizes", ), ( "M16 gateway: missing classification assumed PUBLIC", GATEWAY, " if not classifications:\n" " raise SecurityError(\n" ' "model egress requires an explicit classification; refusing to "\n' ' "assume PUBLIC"\n' " )", " classifications = frozenset({\"public\"}) # MUTATION: assume public", ), ( "M17 gateway: free_only defaults back to False", GATEWAY, " free_only: bool = True\n", " free_only: bool = False # MUTATION: gateway drops the guarantee\n", ), ( "M18 gateway: route() stops forwarding free_only", GATEWAY, " strategy=strat,\n free_only=self.config.free_only if free_only is None else free_only,\n )", " strategy=strat,\n ) # MUTATION: free_only not forwarded", ), ( "M19 kilo_config: atomic replace replaced by in-place write", KILO, ' with open(tmp_path, "w", encoding="utf-8") as handle:\n' ' json.dump(new_config, handle, indent=2, ensure_ascii=False, sort_keys=True)\n' " handle.write(\"\\n\")\n" " os.replace(tmp_path, path)", ' with open(path, "w", encoding="utf-8") as handle:\n' ' json.dump(new_config, handle, indent=2, ensure_ascii=False, sort_keys=True)\n' " handle.write(\"\\n\")", ), ( "M20 role_assignment: verified flag ignored", ROLE, " if backup.verified and backup.is_free:", " if backup.is_free: # MUTATION: unverified backup accepted", ), ( "M21 gateway: fallback drops the free-only check", GATEWAY, " if self.config.free_only and not self._model_is_free(model):", " if False: # MUTATION: paid fallback reachable", ), ( "M22 gateway: _model_is_free returns True for unknown models", GATEWAY, " record = self.model_registry.get(model_id)\n" " return bool(record is not None and record.is_free)", " return True # MUTATION: unknown counts as free", ), ( "M23 router: round-robin lock removed", ROUTER, " with self._rr_lock:\n" " idx = self._rr_counters.get(provider, 0)\n" " selected = candidates[idx % len(candidates)]\n" " self._rr_counters[provider] = idx + 1", " idx = self._rr_counters.get(provider, 0)\n" " selected = candidates[idx % len(candidates)]\n" " self._rr_counters[provider] = idx + 1 # MUTATION: unsynchronised", ), ( "M24 role_assignment: store lock removed from record_result", ROLE, " with self._lock:\n" " assignment = self._store.get(role)\n" " if assignment is None:\n" " logger.warning(\"No assignment found for role %r\", role)\n" " return\n" " assignment.results.append({", " assignment = self._store.get(role) # MUTATION: unsynchronised\n" " if assignment is None:\n" " logger.warning(\"No assignment found for role %r\", role)\n" " return\n" " if True:\n" " assignment.results.append({", ), ( "M25 router: free_only substitution disclosure removed", ROUTER, " if requested_model is not None and free_only and not self._is_free_eligible(requested_model):\n" " if route.model_id != requested_model.model_id:\n" " route.reason = (\n" " f\"specific model {requested_model.model_id!r} requested with free_only=True \"\n" " f\"but not free-eligible; substituted with {route.model_id!r} ({route.reason})\"\n" " )", " if False: # MUTATION: substitution no longer disclosed\n" " pass", ), ] # Provenly unreachable mutations. These are NOT blind tests - no single-threaded # run could detect them because the mutated branch cannot be reached at all. # They are kept out of the runnable list so the "NEPADÁ — SLEPÝ TEST" report # stays truthful, and documented here so the reasoning is not lost. UNREACHABLE = { "M26 router: substitution claimed even when none happened", } def run_suite() -> tuple[bool, str]: executable = sys.executable or "python3" proc = subprocess.run( [executable, "-m", "pytest", "tests/test_f4_model_intelligence.py", "-q"], cwd=REPO, capture_output=True, text=True, ) return proc.returncode == 0, proc.stdout.strip().splitlines()[-1] STRUCTURAL = { "M23", "M24", } def main() -> None: print("=== BASELINE ===") ok, line = run_suite() print(f"baseline {'PASS' if ok else 'FAIL'}: {line}\n") undetected: list[str] = [] for name, path, old, new in MUTATIONS: original = path.read_text(encoding="utf-8") if old not in original: print(f"SKIP {name} — anchor not found in {path.name}") continue path.write_text(original.replace(old, new, 1), encoding="utf-8") try: ok, line = run_suite() label = name.split()[0] if ok and label in STRUCTURAL: print( f"STRUKTURNÍ {name} -> suite zelená, ale jde o souběžnost, " "kterou jednovláknový test neodhalí (viz STRUCTURAL_NOTE)" ) elif ok: verdict = "NEPADÁ — SLEPÝ TEST" undetected.append(name) print(f"{verdict} {name} -> {line}") else: print(f"PADÁ (test umí umřít) {name} -> {line}") finally: path.write_text(original, encoding="utf-8") print("\n=== PO OBNOVENI ===") ok, line = run_suite() print(f"restored {'PASS' if ok else 'FAIL'}: {line}") print("\n=== SLEPÉ TESTY (kontrola bez testu) ===") print("\n".join(undetected) if undetected else "žádný") STRUCTURAL_NOTE = """ M23 (router round-robin lock) and M24 (role_assignment store lock) are NOT covered by this run. Removing a lock changes nothing observable when only one thread executes, so no single-threaded mutation can catch them. They are kept in the list so the gap stays visible instead of being counted as covered. M26 (claiming a substitution that did not happen) is UNREACHABLE, which is a different thing from a blind test. `route()` returns early at router.py:189 whenever `_route_specific_model` succeeds, so the disclosure block is reached only when specific-model routing failed - and that means a substitution really did occur. The over-claim branch cannot execute, so no test could fail on it. It is listed in UNREACHABLE rather than run, so that "NEPADÁ - SLEPÝ TEST" keeps meaning what it says. Concurrency correctness is a separate class of evidence - a stress test, not a mutation run. The correct response is to leave them here with a note, not to delete them and not to call them verified. """ if __name__ == "__main__": main()