"""Tool-agnostic front door for gate B of the self-mock gate — mutation probing.

Gate A (``validators/detect_self_mock.py``) catches a test that stubs its own
subject by syntax. Gate B catches the ones syntax cannot see: if a mutant of a
changed line goes UNDETECTED by the stage's own suite, no test in it constrains
that line. Undetected covers two distinct failures, and both count — the test
ran the code and asserted nothing about it, or the test never reached the code
at all. The second is what a self-mocked test actually looks like from the
outside: stubbing the subject's own method means the real production line never
executes, so it is the signal this gate most needs to keep.

Every external mutation tool (Stryker, cargo-mutants, PIT) answers a different
CLI and a different report format, so each is wrapped in an `Adapter` and this
module owns everything that must NOT differ between them: which language has an
adapter at all, whether that adapter's tool is actually available, which mutants
are in scope, and how survivors become a verdict. Adapters parse; they do not
judge.

Two rules are load-bearing and both are about refusing a quiet pass:

- Nothing degrades silently. A language with no adapter, an adapter whose tool
  is not declared, and a missing diff each answer `unsupported(<reason>)` naming
  the cause. `validate-run.py` folds `unsupported(...)` down to gate A only, so
  a silent "PASS" here would be indistinguishable from a real one.
- Only mutants covering a line the diff added or modified count. A survivor
  elsewhere is pre-existing debt that this stage neither introduced nor is
  blocked by.
- Having nothing to mutate is never a PASS. Production-source selection and the
  empty-set refusal live in `run_probe`, ahead of every adapter, so "the tool
  never ran" cannot be reported in the same words as "the tool ran and found
  nothing undetected". Each adapter would otherwise have to re-earn that
  distinction, and each one is a chance to lose it.
- A sibling language's PASS may stand over a CAPABILITY GAP, never over an
  inspection failure. `--diff` is one shared file, so "incomplete for Rust" also
  indicts the input that scoped the passing TypeScript verdict. The three reason
  classes are defined once here (`classify_reason`) and read by both the merge
  and `validate-run.py`'s gate.
- The two inputs must AGREE, completely. `changed_files` and the diff arrive
  from separate git commands, so `run_probe` refuses to run unless the diff
  names EVERY changed source — one uncovered target is enough to hide a real
  survivor, since its lines never enter the intersection. A diff that names them
  all but adds no line anywhere is refused too: there is nothing to verify, which
  is not the same as verifying and finding nothing.
- An undetected mutant must be PLACEABLE, in BOTH halves of its `(file, line)`
  key. Each adapter screens its own undetected rows — the line through
  `_coerce_line`, the same test `evaluate` uses, and the file by requiring a
  non-empty string — so a survivor that cannot be located fails its report
  instead of being dropped on the way to the intersection. `evaluate`'s own check then stands as defense
  for a future adapter that forgets, not as the live path for a real survivor.
- Only a CONCLUSIVE trial is evidence. A mutant that failed to compile, was
  skipped, or never finished says nothing about the tests, so those outcomes are
  recognised but not counted; a report made entirely of them is
  `unsupported(no-conclusive-mutants)`. Every outcome vocabulary is matched as an
  ALLOWLIST — an unrecognised word fails the report rather than being read as
  "a test caught it".

"Adapters parse, they never judge" is enforced, not merely stated: `run_probe`
refuses any result whose `status` is outside the `PASS`/`FAIL`/`unsupported(...)`
vocabulary, because a fourth value matches neither of `validate-run.py`'s
branches and would pass by falling between them.

`ADAPTERS` is keyed by the `self_mock_signals.EXT_TO_LANG` vocabulary — the same
lang names gate A resolves a changed file to — so both gates answer to one set
of language keys: `ts_js` (Stryker), `python` (Cosmic Ray), `rust`
(cargo-mutants), and `java`/`kotlin` (PIT, whose diff scoping is not wired up
yet — see `PitAdapter`).
"""
from __future__ import annotations

import fnmatch
import json
import os
import re
import shutil
import stat
import subprocess
import sys
import tempfile
import tomllib
from pathlib import Path
from typing import NamedTuple, Protocol

from .self_mock_signals import (
    EXT_TO_LANG,
    partition_waived_entries,
    selfmock_path_key,
)
from .json_boundary import JsonBoundaryError, external_tool_json_source, load_external_json

# `survived` is read by a human in the sidecar, so it is trimmed. The verdict and
# the logged total are taken before the trim — the cap shortens the report, never
# the finding.
SURVIVOR_CAP = 50

ProbeResult = dict[str, object]

_HUNK = re.compile(r"^@@ -\d+(?:,\d+)? \+(\d+)(?:,(\d+))? @@")
_DIFF_GIT_RE = re.compile(r"^diff --git a/(.+) b/(.+)$")


class Adapter(Protocol):
    """One external mutation tool, reduced to what the probe needs from it.

    An adapter parses; it does not judge and it does not select. `run_probe`
    picks the production sources and refuses an empty set before any adapter is
    reached, so `run` is only ever called with at least one real target. That
    ordering is the contract: "selected nothing, therefore PASS" is the vacuous
    pass this gate exists to prevent, and every adapter would otherwise have to
    remember not to reinvent it.
    """

    name: str

    def is_declared(self, worktree: Path | None) -> bool:
        """True when this tool is installed/configured in `worktree`.

        Must answer from files alone — no subprocess, no network. `run_probe`
        consults it before `run`, so shelling out here would be the very failure
        it guards against.
        """

    def run(
        self,
        targets: list[Path],
        diff_path: Path | None,
        worktree: Path | None,
    ) -> ProbeResult:
        """Mutate `targets` (already selected, never empty) and report via `evaluate`.

        Anything that stops the tool producing a usable report — it was never
        installed, it died, the report is unreadable, its scope cannot be
        narrowed to the diff — is `unsupported(<reason>)`, never `PASS`.
        """


STRYKER_REPORT_PATH = Path("reports/mutation/mutation.json")

# Stryker's `--mutate` takes PRODUCTION sources. Handing it a spec file mutates
# the test instead of the code, and handing it nothing at all makes it fall back
# to mutating the whole project. These mirror Stryker's own default test
# excludes; matching is case-insensitive so `Foo.Spec.ts` is caught too.
_TEST_NAME_MARKERS = (".spec.", ".test.")
_TEST_DIR_SEGMENTS = ("test", "tests", "spec", "__tests__")


def _is_test_source(path: Path) -> bool:
    """True for a file a mutation tool must not mutate.

    Mutating test code produces mutants that sit on the stage's changed lines
    and fail it for nothing. A gate that raises false alarms is one people learn
    to route around, which costs more than the mutants it would have caught.
    """
    name = path.name.lower()
    if any(marker in name for marker in _TEST_NAME_MARKERS):
        return True
    # Only directory components — a production file may legitimately be named
    # `spec.ts`, and `parts[:-1]` keeps the filename out of the comparison.
    return any(part.lower() in _TEST_DIR_SEGMENTS for part in path.parts[:-1])


def production_sources(changed_files: list[Path], lang: str) -> list[Path]:
    """The changed files of `lang` that a mutation tool may target."""
    return [
        p
        for p in changed_files
        if EXT_TO_LANG.get(p.suffix) == lang and not _is_test_source(p)
    ]


def _run_stryker_cli(targets: list[Path], worktree: Path | None) -> None:
    """Invoke the real Stryker CLI, scoped to `targets`, writing a json report.

    `--no-install` keeps the promise `is_declared` makes: plain `npx stryker`
    fetches the package from the registry when it is not installed locally, so
    an undeclared worktree would quietly go to the network instead of reporting
    `unsupported(...)`. Without an install this now fails fast and leaves no
    report, which surfaces as `unsupported(report-unavailable)`.

    A surviving mutant makes Stryker exit non-zero, which is a normal outcome
    here rather than an error — the report is what carries the verdict, so the
    exit status is deliberately not checked.
    """
    subprocess.run(
        [
            "npx",
            "--no-install",
            "stryker",
            "run",
            "--reporters",
            "json",
            "--mutate",
            ",".join(str(t) for t in targets),
        ],
        cwd=str(worktree) if worktree is not None else None,
        check=False,
    )


# Stryker's own model is `Undetected = Survived + NoCoverage`, and both accuse
# the tests: `Survived` means the code ran and nothing asserted on it,
# `NoCoverage` means no test reached the code at all. NoCoverage is in fact the
# purest self-mock fingerprint — stubbing the subject's own method stops the
# real production line from ever executing. Every other status (`Killed`,
# `Timeout`, `RuntimeError`, `CompileError`, `Ignored`) either means a test
# caught the mutant or that no usable trial happened, so none of them counts.
#
# Deliberately local to this adapter: cargo-mutants and PIT report their
# outcomes in different vocabularies, so this must not be hoisted into the
# shared layer.
_STRYKER_UNDETECTED = ("Survived", "NoCoverage")


# The FULL vocabulary Stryker emits. Matching against an allowlist rather than
# "anything that is not undetected" is what stops an unrecognised word — a
# renamed status, a newer Stryker, a truncated report — from being read as "a
# test caught it" and quietly clearing the run.
_STRYKER_STATUSES = (
    "Killed",
    "Survived",
    "NoCoverage",
    "Timeout",
    "RuntimeError",
    "CompileError",
    "Ignored",
    "Pending",
)

# The subset that represents a trial that actually finished and therefore says
# something about the tests. `CompileError` and `RuntimeError` mean the mutant
# never produced a usable trial, `Ignored` means it was skipped, and `Pending`
# means the run was cut short before it got there.
_STRYKER_CONCLUSIVE = ("Killed", "Survived", "NoCoverage", "Timeout")


def _stryker_line(mutant: dict) -> int | None:
    """The new-side line of a Stryker mutant, or `None` if it cannot be read.

    Guards each hop rather than trusting the shape: a `location` or `start` that
    is not an object would otherwise raise `AttributeError` out of `run_probe`.
    That crash is worse than it looks — the probe runs before the sidecar is
    written, so it would also destroy gate A's static result and leave the run
    with no sidecar at all. Every malformed shape converges on `report-unparsed`
    instead.
    """
    location = mutant.get("location")
    if not isinstance(location, dict):
        return None
    start = location.get("start")
    if not isinstance(start, dict):
        return None
    return _coerce_line(start.get("line"))


def _survivors_from_report(
    report: dict, worktree: Path | None
) -> ParsedReport | None:
    """Flatten a Stryker json report into rows plus the two trial counts.

    `None` means the report could not be read at all, which is not the same
    answer as "no undetected mutants" and must never become one.

    `status` is carried through because the two undetected outcomes need
    different fixes — write a real assertion (`Survived`) versus cover the code
    at all (`NoCoverage`) — and the sidecar reader cannot tell them apart once
    they are merged. `evaluate` treats the rows opaquely, so the extra key costs
    the shared layer nothing.
    """
    files = report.get("files")
    if files is None:
        files = {}
    if not isinstance(files, dict):
        return None
    rows: list[dict] = []
    conclusive = 0
    observed = 0
    for name, entry in files.items():
        if not isinstance(entry, dict):
            return None
        mutants = entry.get("mutants")
        if mutants is None:
            mutants = []
        if not isinstance(mutants, list):
            return None
        for mutant in mutants:
            if not isinstance(mutant, dict):
                return None
            status = mutant.get("status")
            if status not in _STRYKER_STATUSES:
                return None
            observed += 1
            if status in _STRYKER_CONCLUSIVE:
                conclusive += 1
            if status not in _STRYKER_UNDETECTED:
                # A DETECTED mutant's position is never matched against the diff,
                # so a malformed location cannot change the verdict and is not
                # worth failing a run over. Only undetected ones are read below.
                continue
            line = _stryker_line(mutant)
            if line is None:
                # An undetected mutant we cannot PLACE is unverifiable: `evaluate`
                # could only drop it, turning the most serious finding a report
                # carries into a PASS with a note on stderr.
                return None
            rows.append(
                {
                    "file": _worktree_relative(name, worktree),
                    "line": line,
                    "mutant": mutant.get("mutatorName"),
                    "status": status,
                }
            )
    return ParsedReport(rows, conclusive, observed)


class ParsedReport(NamedTuple):
    """One tool's report, reduced to what the shared guards need from it.

    `conclusive` is deliberately not `len(survivors)` nor `observed`: a mutant
    that failed to compile, was skipped, or never finished says nothing about
    the tests. Counting those as trials is how a run that proved nothing ends up
    reported as clean.
    """

    survivors: list[dict]
    conclusive: int
    observed: int


def _verdict_from_parsed(
    parsed: ParsedReport | None,
    diff_path: Path | None,
    worktree: Path | None,
    tool: str,
) -> ProbeResult:
    """The guards every adapter needs once its own report shape is parsed.

    Report-shape parsing stays local to each adapter — the formats have nothing
    in common — but what an unreadable report, a mutant-free report and a report
    of nothing but inconclusive trials MEAN is identical across tools, and all
    three are answers only `unsupported` can carry. Keeping the decision here
    means a new adapter inherits it instead of re-deriving it.
    """
    if parsed is None:
        return unsupported("report-unparsed", tool=tool)
    if parsed.observed == 0:
        # The tool ran but produced nothing to detect.
        return unsupported("no-mutants-generated", tool=tool)
    if parsed.conclusive == 0:
        # Mutants existed but not one of them completed a real trial — an
        # all-unviable build, or a truncated run. Never evidence of a good suite.
        return unsupported("no-conclusive-mutants", tool=tool)
    return evaluate(parsed.survivors, diff_path, worktree, tool=tool)


class StrykerAdapter:
    """Gate B for TypeScript / JavaScript, wrapping the Stryker CLI.

    The CLI call is injected (`runner`) rather than hard-coded, so a test can
    substitute the external tool — the genuine collaborator — while the
    adapter's own selection, parsing and path handling still run for real.
    """

    name = "stryker"

    def __init__(self, runner=_run_stryker_cli, report_path=STRYKER_REPORT_PATH):
        self._runner = runner
        self._report_path = Path(report_path)

    def is_declared(self, worktree: Path | None) -> bool:
        """True when Stryker is actually INSTALLED here. No subprocess, no network.

        A `package.json` entry is not enough. `node_modules` is gitignored, so a
        fresh stage worktree routinely declares `@stryker-mutator/core` with no
        binary present; answering True there sends the run into
        `npx --no-install`, which fails, writes no report, and BLOCKS the stage on
        `report-unavailable` — failing a run over a tool nobody installed.
        Requiring the binary makes that case `tool-not-declared`, a capability
        gap, which is non-blocking. Symmetric with `CargoMutantsAdapter`, which
        already requires the executable rather than the manifest entry.

        `run_probe` calls this before `run`, so it must answer from files alone —
        shelling out to check would be the very failure it is guarding against.
        """
        if worktree is None:
            return False
        return (Path(worktree) / "node_modules" / ".bin" / "stryker").exists()

    def run(
        self,
        targets: list[Path],
        diff_path: Path | None,
        worktree: Path | None,
    ) -> ProbeResult:
        # Drop any earlier report FIRST. Single-stage final-verification reuses
        # the implementation stage worktree, so a previous run's mutation.json is
        # genuinely on disk; if Stryker then dies before writing, reading it back
        # would report that older run's verdict for code it never saw.
        self._report_file(worktree).unlink(missing_ok=True)
        self._runner(targets, worktree)
        report = self._read_report(worktree)
        if report is None:
            return unsupported("report-unavailable", tool=self.name)
        return _verdict_from_parsed(
            _survivors_from_report(report, worktree), diff_path, worktree, self.name
        )

    def _report_file(self, worktree: Path | None) -> Path:
        if worktree is None:
            return self._report_path
        return Path(worktree) / self._report_path

    def _read_report(self, worktree: Path | None) -> dict | None:
        """The parsed json report, or `None` when the run left none behind."""
        path = self._report_file(worktree)
        if not path.is_file():
            return None
        try:
            # 외부 입력: mutation testing 도구가 생성한 실행 보고서다.
            return load_external_json(
                external_tool_json_source(path, Path(worktree) if worktree else Path.cwd()),
                artifact="mutation test report",
            )
        except JsonBoundaryError:
            return None


COSMIC_RAY_CONFIG_PATH = Path("cosmic-ray.toml")

_COSMIC_WORKER_OUTCOMES = (
    "normal",
    "abnormal",
    "exception",
    "no-test",
    "skipped",
)
_COSMIC_TEST_OUTCOMES = ("killed", "survived", "incompetent", None)
_COSMIC_CONCLUSIVE = ("killed", "survived")
_COSMIC_SESSION_FAILURES = (
    "init-failed",
    "baseline-failed",
    "exec-failed",
    "dump-failed",
)


class SourceState(NamedTuple):
    """Bytes and permissions Cosmic Ray must leave unchanged."""

    data: bytes
    mode: int


class CosmicRayScope(NamedTuple):
    """Validated configuration and every Python source it may mutate."""

    config_path: Path
    sources: tuple[Path, ...]


class CosmicRayConfig(NamedTuple):
    """The source roots and exclusions declared by ``cosmic-ray.toml``."""

    module_paths: tuple[Path, ...]
    excluded_patterns: tuple[str, ...]


class CosmicRaySession(NamedTuple):
    """The external CLI's dump, or the command that prevented one."""

    dump_text: str | None
    failure_reason: str | None


class CosmicRayScopeError(ValueError):
    """A stable unsupported reason for an unsafe or incomplete config scope."""

    def __init__(self, reason: str):
        super().__init__(reason)
        self.reason = reason


def _cosmic_ray_executable(worktree: Path | None) -> Path | None:
    """Find an installed Cosmic Ray executable without starting a process."""
    if worktree is None:
        return None
    root = Path(worktree)
    local_paths = (
        Path(".venv/bin/cosmic-ray"),
        Path("venv/bin/cosmic-ray"),
        Path(".venv/Scripts/cosmic-ray.exe"),
        Path("venv/Scripts/cosmic-ray.exe"),
    )
    for relative in local_paths:
        candidate = root / relative
        if candidate.is_file() and os.access(candidate, os.X_OK):
            return candidate
    found = shutil.which("cosmic-ray")
    return Path(found) if found else None


def _is_within(path: Path, parent: Path) -> bool:
    try:
        path.relative_to(parent)
        return True
    except ValueError:
        return False


def _cosmic_config(config_path: Path, root: Path) -> CosmicRayConfig:
    try:
        data = tomllib.loads(config_path.read_text(encoding="utf-8"))
    except (OSError, tomllib.TOMLDecodeError) as exc:
        raise CosmicRayScopeError("config-unreadable") from exc
    config = data.get("cosmic-ray")
    raw = config.get("module-path") if isinstance(config, dict) else None
    values = [raw] if isinstance(raw, str) else raw
    if not isinstance(values, list) or not values:
        raise CosmicRayScopeError("config-unreadable")
    if any(not isinstance(value, str) or not value.strip() for value in values):
        raise CosmicRayScopeError("config-unreadable")
    paths = tuple(
        (root / value).resolve()
        if not Path(value).is_absolute()
        else Path(value).resolve()
        for value in values
    )
    if any(not _is_within(path, root) for path in paths):
        raise CosmicRayScopeError("config-unreadable")
    exclusions: list[str] = []
    for key in ("excluded-modules", "exclude-modules"):
        raw_exclusions = config.get(key, [])
        if not isinstance(raw_exclusions, list) or any(
            not isinstance(pattern, str) or not pattern.strip()
            for pattern in raw_exclusions
        ):
            raise CosmicRayScopeError("config-unreadable")
        exclusions.extend(raw_exclusions)
    return CosmicRayConfig(paths, tuple(exclusions))


def _cosmic_target_is_excluded(
    target: Path,
    root: Path,
    patterns: tuple[str, ...],
) -> bool:
    relative = target.relative_to(root).as_posix()
    return any(
        fnmatch.fnmatchcase(relative, pattern) or Path(relative).match(pattern)
        for pattern in patterns
    )


def _configured_cosmic_sources(
    module_paths: tuple[Path, ...],
    root: Path,
) -> tuple[Path, ...]:
    sources: set[Path] = set()
    for module_path in module_paths:
        if module_path.is_file() and module_path.suffix == ".py":
            candidates = (module_path,)
        elif module_path.is_dir():
            candidates = tuple(module_path.rglob("*.py"))
        else:
            raise CosmicRayScopeError("config-unreadable")
        for candidate in candidates:
            if candidate.is_symlink():
                raise CosmicRayScopeError("config-unreadable")
            resolved = candidate.resolve()
            if not _is_within(resolved, root) or not resolved.is_file():
                raise CosmicRayScopeError("config-unreadable")
            sources.add(resolved)
    return tuple(sorted(sources))


def _read_cosmic_ray_scope(worktree: Path, targets: list[Path]) -> CosmicRayScope:
    root = Path(worktree).resolve()
    config_path = root / COSMIC_RAY_CONFIG_PATH
    config = _cosmic_config(config_path, root)
    resolved_targets = tuple(
        (root / target).resolve() if not target.is_absolute() else target.resolve()
        for target in targets
    )
    if any(
        not _is_within(target, root)
        or _cosmic_target_is_excluded(target, root, config.excluded_patterns)
        for target in resolved_targets
    ):
        raise CosmicRayScopeError("config-target-mismatch")
    if any(
        not any(
            target == module_path
            or (module_path.is_dir() and _is_within(target, module_path))
            for module_path in config.module_paths
        )
        for target in resolved_targets
    ):
        raise CosmicRayScopeError("config-target-mismatch")
    sources = _configured_cosmic_sources(config.module_paths, root)
    if any(target not in sources for target in resolved_targets):
        raise CosmicRayScopeError("config-target-mismatch")
    return CosmicRayScope(config_path, sources)


def _snapshot_source_state(sources: tuple[Path, ...]) -> dict[Path, SourceState]:
    return {
        path: SourceState(
            path.read_bytes(),
            stat.S_IMODE(path.stat(follow_symlinks=False).st_mode),
        )
        for path in sources
    }


def _restore_source_state(states: dict[Path, SourceState]) -> bool:
    restored = True
    for path, expected in states.items():
        try:
            if path.is_symlink() or (path.exists() and not path.is_file()):
                restored = False
                continue
            if not path.exists() or path.read_bytes() != expected.data:
                path.write_bytes(expected.data)
            path.chmod(expected.mode)
            actual = SourceState(
                path.read_bytes(),
                stat.S_IMODE(path.stat(follow_symlinks=False).st_mode),
            )
            restored = restored and actual == expected
        except OSError:
            restored = False
    return restored


def _run_cosmic_ray_session(
    executable: Path,
    config_path: Path,
    session_path: Path,
    worktree: Path,
) -> CosmicRaySession:
    root = Path(worktree)
    try:
        config_arg = str(config_path.relative_to(root))
    except ValueError:
        config_arg = str(config_path)
    commands = (
        ("init", [str(executable), "init", config_arg, str(session_path)]),
        (
            "baseline",
            [str(executable), "baseline", "--report", config_arg, str(session_path)],
        ),
        ("exec", [str(executable), "exec", config_arg, str(session_path)]),
        ("dump", [str(executable), "dump", str(session_path)]),
    )
    for verb, argv in commands:
        try:
            completed = subprocess.run(
                argv,
                cwd=str(root),
                capture_output=True,
                text=True,
                check=False,
            )
        except OSError:
            return CosmicRaySession(None, f"{verb}-failed")
        if completed.returncode != 0:
            return CosmicRaySession(None, f"{verb}-failed")
        if verb == "dump":
            return CosmicRaySession(completed.stdout, None)
    return CosmicRaySession(None, "dump-failed")


def _cosmic_diff_file(diff: object, worktree: Path) -> str | None:
    if not isinstance(diff, list) or any(not isinstance(line, str) for line in diff):
        return None
    new_paths = [line[4:].split("\t", 1)[0] for line in diff if line.startswith("+++ ")]
    if len(new_paths) != 1 or new_paths[0] == "/dev/null":
        return None
    raw = new_paths[0]
    if raw.startswith("b/"):
        raw = raw[2:]
    candidate = Path(raw)
    root = Path(worktree).resolve()
    if candidate.is_absolute():
        resolved = candidate.resolve()
    else:
        if ".." in candidate.parts:
            return None
        absolute_style = (Path("/") / candidate).resolve()
        resolved = (
            absolute_style if _is_within(absolute_style, root) else root / candidate
        )
    if not _is_within(resolved, root):
        return None
    return str(resolved.relative_to(root))


def _parse_cosmic_ray_dump(
    dump_text: str,
    diff_path: Path | None,
    worktree: Path,
) -> ParsedReport | None:
    scope = read_diff(diff_path)
    if scope is None:
        return None
    survivors: list[dict] = []
    observed = 0
    conclusive = 0
    for raw in dump_text.splitlines():
        if not raw.strip():
            continue
        try:
            row = json.loads(raw)
        except json.JSONDecodeError:
            return None
        if not isinstance(row, dict):
            return None
        worker = row.get("worker_outcome")
        outcome = row.get("test_outcome")
        file = _cosmic_diff_file(row.get("diff"), worktree)
        line = _coerce_line(row.get("line_number"))
        operator = row.get("operator")
        occurrence = row.get("occurrence")
        if (
            worker not in _COSMIC_WORKER_OUTCOMES
            or outcome not in _COSMIC_TEST_OUTCOMES
        ):
            return None
        if worker == "normal" and outcome is None:
            return None
        if file is None or line is None or line < 1:
            return None
        if not isinstance(operator, str) or not operator.strip():
            return None
        if not isinstance(occurrence, int) or isinstance(occurrence, bool):
            return None
        if line not in scope.touched.get(selfmock_path_key(file), ()):
            continue
        observed += 1
        if worker == "normal" and outcome in _COSMIC_CONCLUSIVE:
            conclusive += 1
        if worker == "normal" and outcome == "survived":
            survivors.append(
                {
                    "file": file,
                    "line": line,
                    "mutant": f"{operator}#{occurrence}",
                    "status": outcome,
                }
            )
    return ParsedReport(survivors, conclusive, observed)


def _cosmic_session_verdict(
    session: CosmicRaySession,
    diff_path: Path | None,
    worktree: Path,
    tool: str,
) -> ProbeResult:
    if not isinstance(session, CosmicRaySession):
        return unsupported("report-unparsed", tool=tool)
    if session.failure_reason is not None:
        if session.failure_reason not in _COSMIC_SESSION_FAILURES:
            return unsupported("report-unparsed", tool=tool)
        return unsupported(session.failure_reason, tool=tool)
    if not isinstance(session.dump_text, str):
        return unsupported("report-unparsed", tool=tool)
    parsed = _parse_cosmic_ray_dump(session.dump_text, diff_path, worktree)
    if parsed is None:
        return unsupported("report-unparsed", tool=tool)
    if parsed.observed == 0:
        return unsupported("no-mutable-changed-lines", tool=tool)
    return _verdict_from_parsed(parsed, diff_path, worktree, tool)


class CosmicRayAdapter:
    """Gate B for Python, wrapping an explicitly configured Cosmic Ray CLI."""

    name = "cosmic-ray"

    def __init__(self, runner=_run_cosmic_ray_session):
        self._runner = runner

    def is_declared(self, worktree: Path | None) -> bool:
        if worktree is None:
            return False
        root = Path(worktree)
        return (root / COSMIC_RAY_CONFIG_PATH).is_file() and (
            _cosmic_ray_executable(root) is not None
        )

    def run(
        self,
        targets: list[Path],
        diff_path: Path | None,
        worktree: Path | None,
    ) -> ProbeResult:
        if worktree is None:
            return unsupported("config-unreadable", tool=self.name)
        root = Path(worktree)
        executable = _cosmic_ray_executable(root)
        if executable is None:
            return unsupported("tool-not-declared", tool=self.name)
        try:
            scope = _read_cosmic_ray_scope(root, targets)
            states = _snapshot_source_state(scope.sources)
        except CosmicRayScopeError as exc:
            return unsupported(exc.reason, tool=self.name)
        except OSError:
            return unsupported("source-integrity-failed", tool=self.name)
        result: ProbeResult
        try:
            with tempfile.TemporaryDirectory(prefix="okstra-cosmic-ray-") as temp_dir:
                session_path = Path(temp_dir) / "session.sqlite"
                session = self._runner(
                    executable,
                    scope.config_path,
                    session_path,
                    root,
                )
                result = _cosmic_session_verdict(
                    session,
                    diff_path,
                    root,
                    self.name,
                )
        except OSError:
            result = unsupported("exec-failed", tool=self.name)
        finally:
            sources_restored = _restore_source_state(states)
        if not sources_restored:
            return unsupported("source-integrity-failed", tool=self.name)
        return result


CARGO_OUTCOMES_PATH = Path("mutants.out/outcomes.json")

# cargo-mutants' own vocabulary: `caught` (a test failed, good), `missed` (no
# test failed), `unviable` (the mutant did not compile) and `timeout`. Only
# `missed` accuses the tests. Unlike Stryker and PIT there is no separate
# "not covered" word — cargo-mutants folds uncovered code into `missed`, so this
# tuple is one entry rather than two.
#
# Adapter-local on purpose: Stryker says `Survived`/`NoCoverage`, PIT says
# `SURVIVED`/`NO_COVERAGE`. Hoisting any of them into the shared layer would
# make one tool's vocabulary silently govern another's report.
_CARGO_UNDETECTED = ("missed",)

# The FULL vocabulary, matched as an allowlist. An unrecognised word is exactly
# the unverified case this parser must fail closed on: treating it as "not
# undetected" would let a renamed or suffixed outcome clear the whole run.
#
# UNVERIFIED, and the first thing a real cargo-mutants run must settle: if
# `outcomes.json` also carries a baseline scenario (a `summary` such as
# "Success"), every run here becomes `unsupported(report-unparsed)` and this
# adapter is permanently inert — and `len(entries)` would count that baseline as
# a mutant besides. Confirm the vocabulary AND whether a baseline row exists,
# then fix the allowlist and the `observed` count together. Do not add the word
# on speculation; a wrong guess here is exactly what the allowlist exists to
# catch.
_CARGO_OUTCOMES = ("caught", "missed", "unviable", "timeout")

# `unviable` means the mutant did not compile, so the tests never ran against
# it. Only the rest represent a finished trial.
_CARGO_CONCLUSIVE = ("caught", "missed", "timeout")


def _run_cargo_mutants_cli(diff_path: Path, worktree: Path | None) -> None:
    """Invoke cargo-mutants scoped to `diff_path`, writing `mutants.out/`.

    `--in-diff` restricts testing to mutants overlapping the diff's changed
    regions; the file is expected to carry `b/`-prefixed names, which is exactly
    what the `git diff` output the verifier writes contains. `--no-shuffle`
    keeps the report order stable between runs.

    A missed mutant makes cargo-mutants exit non-zero, which is the normal
    outcome here rather than an error — the report carries the verdict.
    """
    subprocess.run(
        ["cargo", "mutants", "--no-shuffle", "--in-diff", str(diff_path)],
        cwd=str(worktree) if worktree is not None else None,
        check=False,
    )


def _cargo_mutant_location(entry: dict) -> tuple[object, object, object]:
    """`(file, line, description)` for one outcomes.json entry.

    UNVERIFIED SHAPE. The reachable cargo-mutants documentation (mutants.rs)
    pins the outcome words and that `mutants.out/outcomes.json` carries the
    results, but not the key names inside it. This reads the nesting the tool is
    believed to use, and every caller treats an unreadable entry as a reason to
    fail the whole report rather than to skip a row — so if this guess is wrong
    the run reports `unsupported(report-unparsed)` instead of a PASS bought with
    our own parsing error. Confirm against a real run in Task 11.
    """
    scenario = entry.get("scenario")
    mutant = scenario.get("Mutant") if isinstance(scenario, dict) else None
    if not isinstance(mutant, dict):
        return None, None, None
    return (
        mutant.get("file"),
        mutant.get("line"),
        mutant.get("description") or mutant.get("function"),
    )


def _survivors_from_cargo_outcomes(
    report: dict, worktree: Path | None
) -> ParsedReport | None:
    """Undetected rows + trial counts, or `None` if the report cannot be read.

    `None` is deliberately distinct from an empty result: "no missed mutants" is
    evidence, "this report is not the shape we can read" is not, and only the
    first may become a PASS.
    """
    entries = report.get("outcomes")
    if not isinstance(entries, list):
        return None
    rows: list[dict] = []
    conclusive = 0
    for entry in entries:
        if not isinstance(entry, dict):
            return None
        summary = entry.get("summary")
        if not isinstance(summary, str):
            return None
        word = summary.lower()
        if word not in _CARGO_OUTCOMES:
            return None
        if word in _CARGO_CONCLUSIVE:
            conclusive += 1
        if word not in _CARGO_UNDETECTED:
            continue
        file, raw_line, describe = _cargo_mutant_location(entry)
        line = _coerce_line(raw_line)
        # Both halves of the placement key must actually place. A non-string
        # `file` would survive `str()` into a key no diff can ever match — and
        # unlike an unplaceable line, that one is dropped without even a note on
        # stderr. The report shape is unverified, so an object here is plausible.
        if not isinstance(file, str) or not file or line is None:
            return None
        rows.append(
            {
                "file": _worktree_relative(file, worktree),
                "line": line,
                "mutant": describe,
                "status": summary,
            }
        )
    return ParsedReport(rows, conclusive, len(entries))


class CargoMutantsAdapter:
    """Gate B for Rust, wrapping the cargo-mutants CLI."""

    name = "cargo-mutants"

    def __init__(self, runner=_run_cargo_mutants_cli, report_path=CARGO_OUTCOMES_PATH):
        self._runner = runner
        self._report_path = Path(report_path)

    def is_declared(self, worktree: Path | None) -> bool:
        """A crate to mutate AND the subcommand installed. No subprocess.

        `shutil.which` only stats candidate paths, so asking whether the binary
        exists costs nothing and starts nothing.
        """
        if worktree is None:
            return False
        if not (Path(worktree) / "Cargo.toml").is_file():
            return False
        return shutil.which("cargo-mutants") is not None

    def run(
        self,
        targets: list[Path],
        diff_path: Path | None,
        worktree: Path | None,
    ) -> ProbeResult:
        # `targets` is deliberately unused: the shared selection in `run_probe`
        # decides WHETHER there is Rust production code worth running on, while
        # `--in-diff` decides WHAT gets mutated. Passing a file list as well
        # would give cargo-mutants a second, redundant scope to disagree with.
        #
        # Checked before running: `--in-diff` is the only thing keeping this to
        # the stage's own changes, and without it cargo-mutants would mutate the
        # whole crate — slow, and full of findings this stage never caused.
        if diff_path is None or not Path(diff_path).is_file():
            return unsupported("diff-unavailable", tool=self.name)
        self._report_file(worktree).unlink(missing_ok=True)
        self._runner(diff_path, worktree)
        report = self._read_report(worktree)
        if report is None:
            return unsupported("report-unavailable", tool=self.name)
        return _verdict_from_parsed(
            _survivors_from_cargo_outcomes(report, worktree),
            diff_path,
            worktree,
            self.name,
        )

    def _report_file(self, worktree: Path | None) -> Path:
        if worktree is None:
            return self._report_path
        return Path(worktree) / self._report_path

    def _read_report(self, worktree: Path | None) -> dict | None:
        path = self._report_file(worktree)
        if not path.is_file():
            return None
        try:
            # 외부 입력: mutation testing 도구가 생성한 실행 보고서다.
            data = load_external_json(
                external_tool_json_source(path, Path(worktree) if worktree else Path.cwd()),
                artifact="mutation test report",
            )
        except JsonBoundaryError:
            return None
        return data if isinstance(data, dict) else None


_PIT_BUILD_FILES = ("pom.xml", "build.gradle", "build.gradle.kts")


class PitAdapter:
    """Gate B for Java / Kotlin — currently honest about not being able to run.

    PIT's own statuses are SURVIVED, KILLED, NO_COVERAGE, TIMED_OUT, NON_VIABLE,
    MEMORY_ERROR and RUN_ERROR, with undetected = SURVIVED + NO_COVERAGE, the
    same two-failure shape Stryker has. Parsing that is the easy part.

    Scoping is what stops it. PIT is not incapable of SCM scoping — it can
    restrict analysis to files changed in source control — but two concrete
    things block wiring it up here:

    1. That scoping is the `scmMutationCoverage` **Maven goal**. There is no
       equivalent among the build files `is_declared` accepts: a Gradle project
       (`build.gradle` / `build.gradle.kts`) has no such goal to invoke, so the
       adapter cannot offer one story for the languages it is registered under.
    2. Joining PIT's report back to the diff needs a mapping from its
       class-oriented output to repo paths — an unverified report-path↔FQCN
       derivation that depends on each project's source-root layout. Guessing it
       is how a survivor silently misses the changed-line intersection.

    Running unscoped `mutationCoverage` instead would mutate the whole module:
    slow enough that a stage would time out, and it would report mutants on
    lines this stage never touched. So this adapter reports what is true — gate
    B is not wired for the JVM yet — and lets gate A carry the run. `is_declared`
    still distinguishes "PIT is not configured" from "PIT is configured but we
    cannot scope it", which is the difference an operator acts on.

    Wiring this up later means: a Maven-only path invoking `scmMutationCoverage`,
    the XML `outputFormats` report, and a verified path mapping — none of which
    should be guessed.
    """

    name = "pit"

    def is_declared(self, worktree: Path | None) -> bool:
        """True when a build file in this worktree configures PIT."""
        if worktree is None:
            return False
        root = Path(worktree)
        for build_file in _PIT_BUILD_FILES:
            path = root / build_file
            if not path.is_file():
                continue
            try:
                text = path.read_text(encoding="utf-8", errors="replace")
            except OSError:
                continue
            if "pitest" in text.lower():
                return True
        return False

    def run(
        self,
        targets: list[Path],
        diff_path: Path | None,
        worktree: Path | None,
    ) -> ProbeResult:
        return unsupported("diff-scope-unavailable", tool=self.name)


# Keyed by the `self_mock_signals.EXT_TO_LANG` vocabulary, which folds `.ts`,
# `.tsx`, `.js`, `.jsx` and `.mjs` into the single lang `ts_js`. A language
# absent here answers `unsupported(no-adapter:<lang>)`, which is the honest report.
_PIT = PitAdapter()
ADAPTERS: dict[str, Adapter] = {
    "ts_js": StrykerAdapter(),
    "python": CosmicRayAdapter(),
    "rust": CargoMutantsAdapter(),
    "java": _PIT,
    "kotlin": _PIT,
}


def unsupported(reason: str, tool: str | None = None) -> ProbeResult:
    """The one way to say "gate B did not run", always naming why."""
    return {
        "status": f"unsupported({reason})",
        "tool": tool,
        "survived": [],
        "waived": [],
    }


class DiffScope(NamedTuple):
    """What a unified diff says about a stage, in the two forms the gate needs.

    `touched` — file → the new-side lines it added or modified. This is what
    scopes a surviving mutant to work the stage actually did.

    `mentioned` — every file the diff NAMES in a header, whether or not it added
    a line. A renamed or mode-changed file is mentioned but not touched, and
    that distinction is what lets the coverage check be total: every changed
    source must be mentioned, while only the ones with added lines can carry
    findings. Checking `touched` instead would fail a stage for renaming a file.
    """

    touched: dict[str, set[int]]
    mentioned: set[str]


def read_diff(diff_path: Path | None) -> DiffScope | None:
    """Parse a unified diff, or `None` when it cannot be read at all.

    A modification shows up as a `-`/`+` pair, so tracking the `+` side alone
    covers both added and modified lines.

    `None` and an empty scope are different answers and callers must keep them
    apart: `None` means the diff was unreadable (`unsupported(diff-unavailable)`),
    while an empty `touched` means it was read and added nothing anywhere.
    """
    if diff_path is None or not Path(diff_path).is_file():
        return None
    try:
        text = Path(diff_path).read_text(encoding="utf-8", errors="replace")
    except OSError:
        return None
    touched: dict[str, set[int]] = {}
    mentioned: set[str] = set()
    current: str | None = None
    line_no = 0
    for raw in text.splitlines():
        if raw.startswith("diff --git "):
            m = _DIFF_GIT_RE.match(raw)
            if m:
                mentioned.update(_strip_diff_prefix(g) for g in m.groups())
            # A pure rename or mode change emits no `---`/`+++` pair and no
            # hunk, so this header is the ONLY place those files are named.
            # Both sides are recorded: a rename names the old and the new path.
            continue
        if raw.startswith("--- ") or raw.startswith("+++ "):
            target = raw[4:].strip()
            named = None if target == "/dev/null" else _strip_diff_prefix(target)
            if named is not None:
                mentioned.add(named)
            if raw.startswith("+++ "):
                # `/dev/null` on the new side is a deletion: nothing to mutate.
                current = named
            continue
        if raw.startswith("@@"):
            m = _HUNK.match(raw)
            if m is None:
                # A header we cannot read — a combined/merge diff (`@@@ -a -b +c @@@`)
                # or a truncated file. Skipping it would drop every line in the
                # hunk, and in a MIXED diff the coverage check would still be
                # satisfied by the ordinary hunks while these changes went
                # unchecked. The whole diff is unreadable instead.
                return None
            line_no = int(m.group(1))
            continue
        if current is None or not line_no:
            continue
        if raw.startswith("+"):
            touched.setdefault(current, set()).add(line_no)
            line_no += 1
        elif raw.startswith(" ") or raw == "":
            # Allowlist, not "anything but `-`": `\ No newline at end of file`
            # occupies no new-side line, and counting it would slide every later
            # `+` down by one — the intersection would then miss the changed line
            # and report PASS. Only a context line advances the counter.
            line_no += 1
    return DiffScope(touched, mentioned)


def _strip_diff_prefix(target: str) -> str:
    """Drop git's `b/` prefix and any trailing tab-separated timestamp."""
    path = target.split("\t", 1)[0]
    if path.startswith("a/") or path.startswith("b/"):
        path = path[2:]
    return selfmock_path_key(path)


def _coerce_line(value: object) -> int | None:
    """A reported line as the `int` the diff index is keyed on, or `None`.

    A tool that reports `"11"` names the same line as `11`; treating them as
    different would drop the survivor and read as PASS. Anything that will not
    coerce — a word, a `10-12` range, an object — places nowhere.

    The single definition of "placeable": the adapters screen undetected mutants
    with it so an unplaceable one fails its report, and `evaluate` applies it
    again as the shared defense. Two spellings of this test could disagree, and
    the row would fall between them into a PASS.
    """
    try:
        return int(value)
    except (TypeError, ValueError):
        return None


def _new_side_line(survivor: dict) -> int | None:
    """The survivor's placeable line, or `None`.

    Reaching `None` here means an adapter let an unplaceable row through; the
    row is skipped and logged rather than raised on, so one bad row cannot abort
    a whole verdict.
    """
    return _coerce_line(survivor.get("line"))


def _worktree_relative(path: str, worktree: Path | None) -> str:
    """Re-spell an absolute tool path relative to the worktree root.

    Mutation tools report absolute file names; a diff names them relative to the
    worktree. `selfmock_path_key` folds cosmetic drift but not this, so without
    the conversion every survivor misses the intersection and the stage reads
    PASS. A path already relative, or outside the worktree, is left alone.
    """
    if worktree is None:
        return path
    try:
        return str(Path(path).relative_to(Path(worktree)))
    except ValueError:
        return path


def _diff_key(survivor: dict, worktree: Path | None) -> str:
    """The spelling of a survivor's file that `read_diff` keys `touched` on."""
    return selfmock_path_key(
        _worktree_relative(str(survivor.get("file", "")), worktree)
    )


def evaluate(
    survivors: list[dict], diff_path: Path | None, worktree: Path | None, tool: str
) -> ProbeResult:
    """Turn an adapter's raw survivors into a verdict. Adapters must route here.

    `survivors` are `{file, line, mutant}` rows. Two spellings are reconciled
    here so no adapter has to get them right on its own — either mistake would
    drop the survivor silently into a PASS:

    - `file` is folded to worktree-relative and then through `selfmock_path_key`,
      which also absorbs `./` prefixes, duplicated slashes and Windows
      separators. Adapters should still emit worktree-relative paths, because
      that is what lands in the sidecar for a human to read.
    - `line` must be the NEW-side (post-change) line number. A string is coerced,
      but an old-side number simply points somewhere else and cannot be rescued.
    """
    scope = read_diff(diff_path)
    if scope is None:
        return unsupported("diff-unavailable", tool=tool)
    touched = scope.touched
    unplaceable = [s for s in survivors if _new_side_line(s) is None]
    if unplaceable:
        print(
            f"mutation-probe: {tool} reported {len(unplaceable)} mutant(s) with no "
            "usable line number; they cannot be matched against the diff and are "
            "not counted",
            file=sys.stderr,
        )
    covered = [
        s
        for s in survivors
        if _new_side_line(s) in touched.get(_diff_key(s, worktree), ())
    ]
    if len(covered) > SURVIVOR_CAP:
        print(
            f"mutation-probe: {tool} left {len(covered)} surviving mutants on changed "
            f"lines; reporting the first {SURVIVOR_CAP} "
            f"({len(covered) - SURVIVOR_CAP} more not listed)",
            file=sys.stderr,
        )
    return {
        "status": "FAIL" if covered else "PASS",
        "tool": tool,
        "survived": covered[:SURVIVOR_CAP],
        # How many were actually found, before the cap shortened the list. The
        # verdict is decided from this, never from the trimmed list: a reader can
        # only waive what the report shows, so recomputing from `survived` alone
        # would let the visible ones be cleared while the overflow stayed unseen.
        "survivedTotal": len(covered),
        "waived": [],
    }


def _targets_absent_from_diff(
    targets: list[Path], mentioned: set[str], worktree: Path | None
) -> list[str]:
    """The changed sources the diff never names. Empty means the two inputs agree.

    TOTAL coverage, not overlap. Overlap left a real hole: with targets [A, B]
    and a diff covering only A, the check passed on A while a survivor in B fell
    outside the intersection and vanished into a PASS. Asking `mentioned` rather
    than `touched` is what makes totality affordable — a renamed or mode-changed
    target is named by the diff without adding a line, so it satisfies this
    without being expected to carry findings.
    """
    return [
        key
        for target in targets
        if (key := selfmock_path_key(_worktree_relative(str(target), worktree)))
        not in mentioned
    ]


def mutation_waivers(waivers) -> list[dict]:
    """The entries in the shared waiver file that belong to gate B.

    ONE file (`<task_root>/qa/self-mock-waivers.json`) serves both gates so the
    user manages a single place and `waiverSource` stays singular. A static
    waiver names a `signal`; a mutation waiver names a `mutant` — the field is
    the discriminator, so there is no `kind` for anyone to get wrong.

    The filter is explicit rather than relying on the keys failing to line up: a
    survivor whose `mutant` the tool left null would otherwise share the `None`
    third element with a static entry and be cleared by it.
    """
    return [
        w
        for w in (waivers or [])
        if isinstance(w, dict)
        and isinstance(w.get("mutant"), str)
        and w["mutant"].strip()
    ]


def _apply_waivers(result: ProbeResult, waivers) -> ProbeResult:
    """Move user-acknowledged survivors out of `survived` and re-decide.

    Mirrors gate A exactly, including what it refuses to do: this only MATCHES.
    An entry missing `reason` or `acknowledgedBy` is carried into `waived` so
    `validate-run.py` can block on it — rejecting it here would let the run that
    produced the finding excuse itself by writing the file.

    Only a real verdict can be waived. There is no finding to excuse when the
    status is `unsupported(...)`, and letting a waiver rewrite that would turn
    "gate B never ran" into a pass.
    """
    if result["status"] not in ("PASS", "FAIL"):
        return result
    entries = mutation_waivers(waivers)
    if not entries:
        return result
    listed = list(result["survived"])
    remaining, waived = partition_waived_entries(listed, entries, "mutant")
    return {
        **result,
        "status": _waived_verdict(result, listed, remaining),
        "survived": remaining,
        "waived": list(result["waived"]) + waived,
    }


def _waived_verdict(
    result: ProbeResult, listed: list[dict], remaining: list[dict]
) -> str:
    """`PASS` only when EVERY survivor was waived — including any the cap hid.

    A user can only acknowledge what they were shown. Waiving every listed
    survivor of a truncated report must not clear the ones that were trimmed:
    they were never reviewed, and no acknowledgement for them can exist.

    `survivedTotal` is what `evaluate` found before the cap. Each way it can be
    untrustworthy fails closed:

    - smaller than the list it accompanies — the result contradicts itself, so
      nothing in it is reliable enough to clear a finding;
    - absent — a list sitting exactly at the cap may be truncated, so it is
      treated as if it is. Every real adapter routes through `evaluate` and does
      carry the total, so this only catches a hand-built result.
    """
    if remaining:
        return "FAIL"
    total = result.get("survivedTotal")
    if isinstance(total, int) and total >= len(listed):
        return "FAIL" if total > len(listed) else "PASS"
    if total is not None:
        return "FAIL"
    return "FAIL" if len(listed) >= SURVIVOR_CAP else "PASS"


def run_probe(
    lang: str,
    changed_files: list[Path],
    diff_path: Path | None,
    worktree: Path | None,
    waivers=(),
) -> ProbeResult:
    """Probe `lang`'s changed files, or say precisely why it could not be done.

    `changed_files` must be EVERY file the stage changed, not a pre-filtered
    subset — the adapter selects its own production sources from it. Forwarding
    only the changed TEST files (gate A's `scannedFiles`, which is the tempting
    thing to reuse) leaves nothing to mutate, and the run reports a vacuous PASS
    with gate B silently dead behind it.
    """
    adapter = ADAPTERS.get(lang)
    if adapter is None:
        return unsupported(f"no-adapter:{lang}")
    # Checked before `run` so a missing binary surfaces as its own reason rather
    # than as an adapter-specific crash or, worse, an empty survivor list.
    if not adapter.is_declared(worktree):
        return unsupported("tool-not-declared", tool=adapter.name)
    # Selected HERE rather than inside each adapter: three adapters each doing
    # their own selection is three chances to answer PASS for a run that never
    # mutated anything. An adapter is never handed an empty set.
    targets = production_sources(changed_files, lang)
    if not targets:
        return unsupported("no-production-sources", tool=adapter.name)
    # `--changed-file` and `--diff` are built by two separate git commands, so
    # nothing guarantees they describe the same work. If the diff covers none of
    # the files about to be mutated, every survivor falls outside the
    # intersection and the stage is stamped PASS having verified nothing.
    scope = read_diff(diff_path)
    if scope is None:
        return unsupported("diff-unavailable", tool=adapter.name)
    missing = _targets_absent_from_diff(targets, scope.mentioned, worktree)
    if missing:
        print(
            f"mutation-probe: {adapter.name} was handed {len(missing)} changed "
            f"source(s) the diff never names: {missing}",
            file=sys.stderr,
        )
        return unsupported("diff-incomplete", tool=adapter.name)
    if not scope.touched:
        # The diff names every target but adds no line anywhere — a deletion-only
        # or header-only diff. There is nothing for gate B to verify, which is
        # not the same as verifying it and finding nothing.
        return unsupported("diff-adds-no-line", tool=adapter.name)
    result = adapter.run(targets, diff_path, worktree)
    if not _states_a_known_verdict(result):
        return unsupported("adapter-malformed-status", tool=adapter.name)
    return _apply_waivers(result, waivers)


def _states_a_known_verdict(result: object) -> bool:
    """True when `result` speaks the vocabulary the gate downstream understands.

    "Adapters parse, they never judge" is only a rule if something checks it.
    `validate-run.py` blocks on `FAIL` and folds `unsupported(...)` down to gate
    A; a third value matches neither and would sail through both. Enforcing it
    on the interface means each new adapter inherits the check instead of
    re-earning trust.
    """
    if not isinstance(result, dict):
        return False
    status = result.get("status")
    if not isinstance(status, str):
        return False
    if status in ("PASS", "FAIL"):
        return True
    if not (status.startswith("unsupported(") and status.endswith(")")):
        return False
    # An empty reason is the one thing `unsupported` must never be: "gate B did
    # not run" with no way to find out why is indistinguishable from a silent skip.
    return bool(_unsupported_reason(status))


def _unsupported_reason(status: str) -> str:
    """The text inside `unsupported(...)`, or `""` when there is none."""
    return status[len("unsupported(") : -1].strip()


def _probe_langs(
    changed_files: list[Path], worktree: Path | None
) -> dict[str, list[Path]]:
    """Group the stage's changed files by the language key each adapter answers to.

    Paths are folded to worktree-relative FIRST. An adapter decides what to
    exclude from its own path shape — Stryker drops anything under a `test`,
    `tests` or `spec` directory — so an absolute path drags the checkout's own
    ancestors into that decision and can exclude real production code whenever
    the worktree happens to live under such a directory. That failure is silent:
    the gate simply finds nothing to mutate and reports a vacuous PASS.
    """
    grouped: dict[str, list[Path]] = {}
    for path in changed_files:
        relative = Path(_worktree_relative(str(path), worktree))
        lang = EXT_TO_LANG.get(relative.suffix)
        if lang is None:
            continue
        grouped.setdefault(lang, []).append(relative)
    return grouped


# --- reason classification SSOT ---------------------------------------------
#
# Every `unsupported(<reason>)` this module can emit falls into exactly one of
# three classes. BOTH consumers read this one definition — `_merge_probes` below
# decides whether a sibling language's PASS may stand, and
# `validate-run.py::_validate_selfmock` decides whether the run blocks. A second,
# parallel table is how those two drift apart.
CAPABILITY_GAP = "capability-gap"
NOTHING_TO_VERIFY = "nothing-to-verify"
INTEGRITY_INSPECTION = "integrity-inspection"

# Gate B declared this boundary in advance: it never claimed to cover that
# language here. Non-blocking, and a sibling's PASS may stand over it — without
# that, gate B would be unusable in any polyglot repo and would wedge every repo
# with no mutation tooling installed.
CAPABILITY_GAP_REASONS = frozenset(
    {
        "tool-not-declared",
        "diff-scope-unavailable",
        "no-production-sources",
        # No changed file is in a language any adapter covers — a Go, Ruby or C#
        # stage. Gate A's trigger is extension-agnostic, so those repos reach
        # gate B on every run; blocking here would wedge all of them.
        "no-changed-sources",
    }
)

# The tool ran to completion and the changed code genuinely offered nothing to
# check. That is a property of the code, not a failure of the run, so it is also
# non-blocking. `diff-adds-no-line` lives here because deletion-only and
# rename-only stages are legitimate and must not wedge.
NOTHING_TO_VERIFY_REASONS = frozenset(
    {
        "no-mutants-generated",
        "no-mutable-changed-lines",
        "diff-adds-no-line",
    }
)

# We tried to inspect and cannot trust the answer. Never overridden by a sibling
# PASS, and BLOCKING at the gate: each of these is a fixable fault in the run's
# own inputs or output, not a boundary anyone declared.
#
# `no-conclusive-mutants` belongs here rather than with "nothing to verify":
# mutants existed and not one completed a trial, so the same mutant reports
# `Survived` when the run finishes and `Pending` when it is cut short. Classing
# it as a gap would let the verdict turn on whether the run was interrupted.
INTEGRITY_INSPECTION_REASONS = frozenset(
    {
        "diff-incomplete",
        "diff-unavailable",
        "report-unavailable",
        "report-unparsed",
        "adapter-malformed-status",
        "no-conclusive-mutants",
        "config-unreadable",
        "config-target-mismatch",
        "init-failed",
        "baseline-failed",
        "exec-failed",
        "dump-failed",
        "source-integrity-failed",
    }
)


def _strip_tool_qualifier(reason: str) -> str:
    """Drop the `<tool>:` prefix `_merged_reasons` adds, keeping `no-adapter:<lang>`.

    The sidecar records the MERGED reason, so what reaches the gate looks like
    `stryker:diff-incomplete`. `no-adapter:<lang>` carries its colon as part of
    the reason itself and is never tool-qualified (its `tool` is `None`), so it
    is recognised before the split rather than being truncated to a bare lang.
    """
    if reason.startswith("no-adapter:"):
        return reason
    _, sep, rest = reason.partition(":")
    return rest if sep else reason


def _classify_one(reason: str) -> str:
    bare = _strip_tool_qualifier(reason.strip())
    if bare.startswith("no-adapter:") or bare in CAPABILITY_GAP_REASONS:
        return CAPABILITY_GAP
    if bare in NOTHING_TO_VERIFY_REASONS:
        return NOTHING_TO_VERIFY
    # Fail closed: a reason nobody classified is treated as an inspection
    # failure, so forgetting to classify one blocks rather than passes.
    return INTEGRITY_INSPECTION


def classify_reason(status: str) -> str:
    """Classify an `unsupported(<reason>)` status into one of the three classes.

    A merged status can carry several `; `-joined reasons; the most severe class
    wins, so one inspection failure among capability gaps still governs.
    """
    reasons = [r for r in _unsupported_reason(status).split(";") if r.strip()]
    classes = {_classify_one(r) for r in reasons}
    if not classes or INTEGRITY_INSPECTION in classes:
        return INTEGRITY_INSPECTION
    if NOTHING_TO_VERIFY in classes:
        return NOTHING_TO_VERIFY
    return CAPABILITY_GAP


def _merge_probes(results: list[ProbeResult]) -> ProbeResult:
    """Fold one verdict per language into the single one the sidecar records.

    Any `FAIL` fails the stage — that precedence is absolute.

    Otherwise a single real `PASS` carries the stage, but ONLY over capability
    gaps. A probed language's clean result is evidence and an uncoverable
    language is merely silence; a language whose INSPECTION failed is neither,
    and letting a sibling speak for it is how total coverage inside one language
    becomes partial coverage across two.

    When nothing was probed at all, every reason is kept: which language went
    unchecked is the operator's next action.
    """
    statuses = [r["status"] for r in results]
    survived = [row for r in results for row in r["survived"]]
    waived = [row for r in results for row in r["waived"]]
    ran = [r["tool"] for r in results if r["status"] in ("PASS", "FAIL") and r["tool"]]
    unsupported_results = [
        r for r in results if str(r["status"]).startswith("unsupported(")
    ]
    # One classification, shared with the gate: a sibling PASS may stand over a
    # capability gap or a language with nothing to verify, never over an
    # inspection failure.
    blocking = [
        r
        for r in unsupported_results
        if classify_reason(str(r["status"])) == INTEGRITY_INSPECTION
    ]
    if "FAIL" in statuses:
        status = "FAIL"
    elif blocking:
        # Reported from the blocking reasons alone: those are what the operator
        # has to fix, and folding in the capability gaps would bury them.
        status = f"unsupported({'; '.join(_merged_reasons(blocking))})"
    elif "PASS" in statuses:
        status = "PASS"
    else:
        status = f"unsupported({'; '.join(_merged_reasons(unsupported_results))})"
    return {
        "status": status,
        "tool": ",".join(sorted(set(ran))) or None,
        "survived": survived,
        # Summed so the sidecar still shows how many were found even when a
        # language's list was capped; `survived` alone would understate it.
        "survivedTotal": sum(
            r["survivedTotal"]
            if isinstance(r.get("survivedTotal"), int)
            else len(r["survived"])
            for r in results
        ),
        "waived": waived,
    }


def _merged_reasons(results: list[ProbeResult]) -> list[str]:
    """Each language's reason, tool-qualified where the reason alone is ambiguous."""
    reasons: list[str] = []
    for r in results:
        reason = _unsupported_reason(str(r["status"]))
        # `no-adapter:<lang>` already names its language; `tool-not-declared` and
        # friends do not, so the tool that produced them is prepended.
        if r["tool"]:
            reason = f"{r['tool']}:{reason}"
        if reason not in reasons:
            reasons.append(reason)
    return reasons


def probe_changed_files(
    changed_files: list[Path],
    diff_path: Path | None,
    worktree: Path | None,
    waivers=(),
) -> ProbeResult:
    """Probe every language the stage touched and merge the verdicts into one.

    This is the entry point the detector writes its sidecar from. `changed_files`
    is the stage's WHOLE changed set — each adapter selects its own production
    sources out of it (see `run_probe`).
    """
    grouped = _probe_langs(changed_files, worktree)
    if not grouped:
        return unsupported("no-changed-sources")
    return _merge_probes(
        [
            run_probe(lang, files, diff_path, worktree, waivers)
            for lang, files in sorted(grouped.items())
        ]
    )