diff --git a/benchmarks/measurements/2026-09-08-module-audit.json b/benchmarks/measurements/2026-09-08-module-audit.json
index 7cefec7..e2f5a5b 100644
--- a/benchmarks/measurements/2026-09-08-module-audit.json
+++ b/benchmarks/measurements/2026-09-08-module-audit.json
@@ -1,13 +1,13 @@
 {
   "counts": {
-    "load_bearing": 74,
+    "load_bearing": 79,
     "standalone_capability": 7
   },
-  "generated": "2026-09-08",
+  "generated": "2026-09-08 (regenerated after group 2 added the cpg/ seam)",
   "kept_with_a_lever_stripped": {
     "improve/evolve.py + improve/validator.py": "the design's Removed list named these whole, but each is a mix: the mechanism lever (removed) and the deterministic rule-status improve loop that `ousast improve` still runs from benchmark evidence, which has no LLM in it and is not what this feature supersedes. Deleting the files would have taken a retained capability with them."
   },
-  "modules": 81,
+  "modules": 86,
   "note": "Classification of every source module after the group-1 subtraction. load_bearing = reachable from a gate, the CLI, or named by a surviving requirement; standalone_capability = its own entry point or protocol, deliberately off the scan path; orphaned = no importer, no entry point, no surviving requirement -- removed, never carried.",
   "orphaned": [],
   "removed_in_this_pass": {
@@ -124,6 +124,31 @@
       "module": "config",
       "reason": "reachable from gate, map_gate, pair_gate or the CLI"
     },
+    {
+      "classification": "load_bearing",
+      "importers": [
+        "cpg.backend",
+        "model.taint"
+      ],
+      "module": "cpg",
+      "reason": "model-grounded-detection Req 4: the CPG seam package"
+    },
+    {
+      "classification": "load_bearing",
+      "importers": [
+        "model.taint"
+      ],
+      "module": "cpg.backend",
+      "reason": "model-grounded-detection Req 4.1/4.5: the single Joern subprocess boundary"
+    },
+    {
+      "classification": "load_bearing",
+      "importers": [
+        "cpg.backend"
+      ],
+      "module": "cpg.capability",
+      "reason": "model-grounded-detection Req 4.2/4.3: the engine probe that degrades to suspicion"
+    },
     {
       "classification": "load_bearing",
       "importers": [
@@ -295,7 +320,8 @@
       "classification": "load_bearing",
       "importers": [
         "cli",
-        "model.specs"
+        "model.specs",
+        "model.taint"
       ],
       "module": "model",
       "reason": "reachable from gate, map_gate, pair_gate or the CLI"
@@ -322,10 +348,26 @@
     },
     {
       "classification": "load_bearing",
-      "importers": [],
+      "importers": [
+        "model.taint"
+      ],
+      "module": "model.ladder",
+      "reason": "model-grounded-detection Req 5: the evidence ladder every finding carries"
+    },
+    {
+      "classification": "load_bearing",
+      "importers": [
+        "model.taint"
+      ],
       "module": "model.specs",
       "reason": "model-grounded-detection Req 7.2-7.4: the security vocabularies the CPG queries are parameterised by"
     },
+    {
+      "classification": "load_bearing",
+      "importers": [],
+      "module": "model.taint",
+      "reason": "model-grounded-detection Req 6.1: taint reachability over the CPG, the flow-family arbiter"
+    },
     {
       "classification": "load_bearing",
       "importers": [
@@ -598,6 +640,7 @@
       "classification": "load_bearing",
       "importers": [
         "cli",
+        "cpg.capability",
         "hunter_tools",
         "model.candidates",
         "model.specs",
@@ -627,6 +670,7 @@
     {
       "classification": "load_bearing",
       "importers": [
+        "cpg.capability",
         "semantic",
         "semantic.overlay"
       ],
diff --git a/src/openultrasast/cpg/__init__.py b/src/openultrasast/cpg/__init__.py
new file mode 100644
index 0000000..0b73e52
--- /dev/null
+++ b/src/openultrasast/cpg/__init__.py
@@ -0,0 +1,3 @@
+"""The Code Property Graph seam: the adopted engine the model layer arbitrates over."""
+
+from __future__ import annotations
diff --git a/src/openultrasast/cpg/backend.py b/src/openultrasast/cpg/backend.py
new file mode 100644
index 0000000..e6671f5
--- /dev/null
+++ b/src/openultrasast/cpg/backend.py
@@ -0,0 +1,168 @@
+"""The Joern subprocess seam — the single place this codebase touches a CPG engine (Req 4.1, 4.4, 4.5).
+
+Joern is JVM/Scala. Everything crosses one subprocess boundary: ``joern-parse`` builds a ``cpg.bin`` from a
+source tree, then ``joern --script`` runs a CPGQL script that prints a JSON payload. Nothing is imported
+in-process, so the core install keeps ``dependencies = []`` and an absent engine costs a recorded reason
+rather than a traceback.
+
+Two rules hold everywhere below:
+
+* **Fail closed.** A build or query that errors, times out, or prints something unparseable yields *no*
+  verdict. The model layer then reports ``suspicion``. A wrong verdict is worse than no verdict — that is the
+  whole reason this arbiter exists.
+* **Fence the payload.** Joern prints a banner, compilation notices and a REPL prompt on stdout. The script
+  brackets its JSON between markers so parsing can never swallow the noise around it.
+
+We do not build a control-flow or dataflow engine here (Req 4.4): the CFG, PDG and call graph our flat IR
+lacks are exactly what is being adopted.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import shutil
+import subprocess
+import tempfile
+from collections.abc import Callable, Mapping
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Protocol
+
+logger = logging.getLogger(__name__)
+
+QUERIES_DIR = Path(__file__).resolve().parent / "queries"
+
+# The payload fence. Joern's stdout carries a banner and a prompt; the script prints these around its JSON.
+BEGIN = "---OUSAST-CPG-BEGIN---"
+END = "---OUSAST-CPG-END---"
+
+BUILD_TIMEOUT_SECONDS = 900  # a large tree takes minutes; the go/no-go records what it actually cost
+QUERY_TIMEOUT_SECONDS = 300
+
+Runner = Callable[..., subprocess.CompletedProcess[str]]
+
+
+def extract_payload(stdout: str) -> object | None:
+    """The JSON between the fence markers, or ``None`` when it is absent or malformed."""
+    start = stdout.find(BEGIN)
+    end = stdout.find(END, start + 1) if start >= 0 else -1
+    if start < 0 or end < 0:
+        return None
+    body = stdout[start + len(BEGIN) : end].strip()
+    try:
+        payload: object = json.loads(body)
+    except json.JSONDecodeError:
+        logger.debug("cpg query payload was not valid JSON")
+        return None
+    return payload
+
+
+@dataclass(frozen=True)
+class CpgResult:
+    """A built CPG and the way to query it. Opaque: only ``model/`` reads through ``run``."""
+
+    cpg_path: Path
+    run: Callable[[str, Mapping[str, object]], object | None]
+
+
+class CpgBackend(Protocol):
+    def available(self) -> bool: ...
+
+    def build(self, root: Path) -> CpgResult | None: ...
+
+
+@dataclass
+class NullBackend:
+    """No engine. Every verdict degrades to ``suspicion`` and the caller records ``cpg_unavailable``."""
+
+    reason: str = "cpg_unavailable"
+
+    def available(self) -> bool:
+        return False
+
+    def build(self, root: Path) -> CpgResult | None:
+        del root
+        return None
+
+
+@dataclass
+class JoernBackend:
+    """``joern-parse`` then ``joern --script``, both out of process and both timeout-bounded."""
+
+    runner: Runner | None = None
+    build_timeout: int = BUILD_TIMEOUT_SECONDS
+    query_timeout: int = QUERY_TIMEOUT_SECONDS
+    queries_dir: Path = field(default_factory=lambda: QUERIES_DIR)
+
+    def available(self) -> bool:
+        from .capability import has_cpg
+
+        return has_cpg()
+
+    def build(self, root: Path) -> CpgResult | None:
+        """Build a CPG for ``root``. ``None`` on any failure — the caller degrades, never guesses."""
+        parse = shutil.which("joern-parse")
+        if parse is None and os.environ.get("OPENULTRASAST_JOERN_PROBE", "").strip().lower() not in {"1", "on", "true", "yes"}:
+            return None
+        scratch = Path(tempfile.mkdtemp(prefix="ousast-cpg-"))
+        cpg_path = scratch / "cpg.bin"
+        command = [parse or "joern-parse", str(root), "--output", str(cpg_path)]
+        completed = self._run(command, timeout=self.build_timeout)
+        if completed is None or completed.returncode != 0 or not cpg_path.is_file():
+            detail = (completed.stderr or completed.stdout or "")[-400:] if completed is not None else "timeout"
+            logger.warning("cpg build failed for %s: %s", root, detail)
+            return None
+        return CpgResult(cpg_path=cpg_path, run=lambda query, params: self.query(cpg_path, query, params))
+
+    def query(self, cpg_path: Path, query: str, params: Mapping[str, object]) -> object | None:
+        """Run a shipped CPGQL script against a built CPG and return its fenced JSON payload."""
+        script = self.queries_dir / f"{query}.sc"
+        if not script.is_file():
+            logger.warning("no such cpg query: %s", script)
+            return None
+        command = [shutil.which("joern") or "joern", "--script", str(script), "--param", f"cpgFile={cpg_path}"]
+        for key, value in sorted(params.items()):
+            command += ["--param", f"{key}={_render(value)}"]
+        completed = self._run(command, timeout=self.query_timeout)
+        if completed is None or completed.returncode != 0:
+            detail = (completed.stderr or "")[-400:] if completed is not None else "timeout"
+            logger.warning("cpg query %s failed: %s", query, detail)
+            return None
+        return extract_payload(completed.stdout or "")
+
+    def _run(self, command: list[str], *, timeout: int) -> subprocess.CompletedProcess[str] | None:
+        run = self.runner if self.runner is not None else subprocess.run
+        try:
+            return run(command, capture_output=True, text=True, timeout=timeout, check=False)
+        except (OSError, subprocess.SubprocessError):
+            return None
+
+
+def _render(value: object) -> str:
+    """CPGQL parameters are strings; a sequence becomes a comma-separated list the script splits."""
+    if isinstance(value, str):
+        return value
+    if isinstance(value, (list, tuple, set, frozenset)):
+        return ",".join(sorted(str(item) for item in value))
+    return str(value)
+
+
+def resolve_cpg_backend(*, runner: Runner | None = None) -> CpgBackend:
+    """The backend for this machine: Joern when it is installed, else the null one."""
+    from .capability import has_cpg
+
+    return JoernBackend(runner=runner) if has_cpg() else NullBackend()
+
+
+__all__ = [
+    "BEGIN",
+    "END",
+    "CpgBackend",
+    "CpgResult",
+    "JoernBackend",
+    "NullBackend",
+    "extract_payload",
+    "resolve_cpg_backend",
+]
diff --git a/src/openultrasast/cpg/capability.py b/src/openultrasast/cpg/capability.py
new file mode 100644
index 0000000..7637c01
--- /dev/null
+++ b/src/openultrasast/cpg/capability.py
@@ -0,0 +1,18 @@
+"""Is a CPG engine available? (Req 4.2, 4.3)
+
+One probe, not two: ``semantic/engines.joern_available`` already answers this and already honours the
+``OPENULTRASAST_JOERN_PROBE`` override that the test matrix uses to force the absent case. The model layer
+reuses it rather than growing a second, subtly different notion of "installed".
+"""
+
+from __future__ import annotations
+
+from ..semantic.engines import joern_available
+
+
+def has_cpg() -> bool:
+    """True when a CPG can be built. False means every verdict degrades to ``suspicion`` with a reason."""
+    return joern_available()
+
+
+__all__ = ["has_cpg"]
diff --git a/src/openultrasast/cpg/queries/taint.sc b/src/openultrasast/cpg/queries/taint.sc
new file mode 100644
index 0000000..0e1ab41
--- /dev/null
+++ b/src/openultrasast/cpg/queries/taint.sc
@@ -0,0 +1,61 @@
+// Taint reachability for one flow family (model-grounded-detection Req 6.1).
+//
+// The question this answers is the one our flat IR could not: does an untrusted source reach the labeled sink
+// *through resolved dataflow*, and is there a sanitizer on that path. Joern's `reachableByFlows` walks the
+// PDG, so a value bound three statements upstream — `q = "..." + request.args["x"]; execute(q)` — is found,
+// which is exactly the one-hop limit that held our measured entailment at 2.2%.
+//
+// Parameters (all comma-separated where plural):
+//   cpgFile     the built cpg.bin
+//   sources     source matchers, e.g. request.args,request.form
+//   sinks       sink call names, e.g. execute,os.system
+//   sanitizers  sanitizer call names; a flow through one of these is corroborated, not entailed
+//   function    optional: restrict sinks to this enclosing method
+//
+// Output: a fenced JSON array of {sink, sinkLine, sinkMethod, source, sanitized, length}.
+// The fence exists because Joern prints a banner and a prompt around whatever a script emits.
+
+@main def exec(cpgFile: String, sources: String, sinks: String, sanitizers: String = "", function: String = "") = {
+  importCpg(cpgFile)
+
+  def split(raw: String): List[String] =
+    raw.split(",").map(_.trim).filter(_.nonEmpty).toList
+
+  val sourcePatterns    = split(sources)
+  val sinkNames         = split(sinks)
+  val sanitizerNames    = split(sanitizers)
+
+  // A source is any call or identifier whose code matches one of the source patterns. Matching on `code`
+  // rather than on a method name keeps `request.args["x"]` and `req.body.name` both reachable without
+  // per-framework special cases.
+  def sourceNodes = cpg.call.filter(c => sourcePatterns.exists(p => c.code.contains(p))).l ++
+                    cpg.identifier.filter(i => sourcePatterns.exists(p => i.code.contains(p))).l
+
+  // A sink is a call whose name (or full code) names one of the family's sinks, optionally inside `function`.
+  def sinkNodes = {
+    val all = cpg.call.filter(c => sinkNames.exists(n => c.name == n || c.code.startsWith(n) || c.methodFullName.contains(n))).l
+    if (function.isEmpty) all else all.filter(_.method.name == function)
+  }
+
+  def isSanitized(flowCode: List[String]): Boolean =
+    sanitizerNames.nonEmpty && flowCode.exists(code => sanitizerNames.exists(s => code.contains(s)))
+
+  val rows = sinkNodes.flatMap { sink =>
+    val flows = sourceNodes.reachableByFlows(sink).l
+    flows.map { flow =>
+      val elements = flow.elements.map(_.code).l
+      ujson.Obj(
+        "sink"       -> sink.code.take(200),
+        "sinkLine"   -> sink.lineNumber.getOrElse(-1).toString,
+        "sinkMethod" -> sink.method.name,
+        "source"     -> elements.headOption.getOrElse("").take(200),
+        "sanitized"  -> isSanitized(elements),
+        "length"     -> elements.size
+      )
+    }
+  }
+
+  println("---OUSAST-CPG-BEGIN---")
+  println(ujson.write(ujson.Arr(rows: _*)))
+  println("---OUSAST-CPG-END---")
+}
diff --git a/src/openultrasast/model/audit.py b/src/openultrasast/model/audit.py
index 38a32b8..264cd70 100644
--- a/src/openultrasast/model/audit.py
+++ b/src/openultrasast/model/audit.py
@@ -31,6 +31,11 @@ SPEC_OWNED = {
     "model.specs": "model-grounded-detection Req 7.2-7.4: the security vocabularies the CPG queries are parameterised by",
     "model.endpoint": "model-grounded-detection Req 8: the judge client, wired by model/judge.py in group 3",
     "model.audit": "model-grounded-detection Req 3: this classifier; run by the maintainer, not by the scan",
+    "model.ladder": "model-grounded-detection Req 5: the evidence ladder every finding carries",
+    "model.taint": "model-grounded-detection Req 6.1: taint reachability over the CPG, the flow-family arbiter",
+    "cpg": "model-grounded-detection Req 4: the CPG seam package",
+    "cpg.backend": "model-grounded-detection Req 4.1/4.5: the single Joern subprocess boundary",
+    "cpg.capability": "model-grounded-detection Req 4.2/4.3: the engine probe that degrades to suspicion",
 }
 
 STANDALONE = {
diff --git a/src/openultrasast/model/ladder.py b/src/openultrasast/model/ladder.py
new file mode 100644
index 0000000..1e85fb7
--- /dev/null
+++ b/src/openultrasast/model/ladder.py
@@ -0,0 +1,48 @@
+"""The evidence ladder: what established a finding (model-grounded-detection Req 5).
+
+Every finding carries the rung that established it, so a deterministic model result is never confused with an
+LLM suspicion. The two middle rungs are the point of this feature:
+
+    suspicion           a site was enumerated and something flagged it
+    model_corroborated  the LLM's claim is consistent with the CPG -- the source it names reaches the sink it
+                        names, or a guard its siblings carry is absent
+    model_entailed      the CPG establishes it alone, independent of trusting the LLM
+    execution_confirmed a sandbox reproduced it (deferred tier, memory/C first)
+
+`model_corroborated` and above are established with no model call, and deterministically: run the checker
+twice over the same CPG and the verdict sequence is identical. That is what removes the run-to-run noise the
+prior architecture tried to average away.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import StrEnum
+
+
+class Rung(StrEnum):
+    SUSPICION = "suspicion"
+    CORROBORATED = "model_corroborated"
+    ENTAILED = "model_entailed"
+    EXECUTION_CONFIRMED = "execution_confirmed"
+
+
+# Ascending strength; `_ORDER.index` is the comparison, so a new rung cannot be silently mis-ranked.
+_ORDER = (Rung.SUSPICION, Rung.CORROBORATED, Rung.ENTAILED, Rung.EXECUTION_CONFIRMED)
+
+
+def at_or_above(rung: Rung, threshold: Rung) -> bool:
+    return _ORDER.index(rung) >= _ORDER.index(threshold)
+
+
+@dataclass(frozen=True)
+class Verdict:
+    """What the model established, and the evidence it established it from."""
+
+    rung: Rung
+    family: str
+    witness: str = ""  # the taint path or the dominance fact; empty only at `suspicion`
+    contradiction: str = ""  # why an LLM claim was dropped, when it was
+
+
+__all__ = ["Rung", "Verdict", "at_or_above"]
diff --git a/src/openultrasast/model/taint.py b/src/openultrasast/model/taint.py
new file mode 100644
index 0000000..3c1b688
--- /dev/null
+++ b/src/openultrasast/model/taint.py
@@ -0,0 +1,73 @@
+"""Taint reachability over the CPG: the arbiter for the flow families (Req 6.1).
+
+The verdict is the static analogue of a reproduced crash. An unsanitized source→sink path is `ENTAILED` — the
+graph establishes it and the LLM's claim is no longer load-bearing. A path a sanitizer may break is
+`CORROBORATED`: the flow is real, but whether the sanitizer is *sufficient* is a semantic judgement the model
+cannot make, and that residual is the LLM's — checked against this verdict rather than averaged over runs.
+
+No path at all yields `None`, not a negative verdict: the caller reports `suspicion`. And a query that failed
+yields `None` too, because "the engine could not decide" is not "there is no flow" — conflating them is how a
+missing engine would silently become a clean bill of health.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+
+from ..cpg.backend import CpgResult
+from .ladder import Rung, Verdict
+from .specs import TaintSpec
+
+
+def verdict(cpg: CpgResult, spec: TaintSpec, *, function: str = "") -> Verdict | None:
+    """The strongest verdict the taint query supports for ``spec`` in ``function``, or ``None`` to stay at suspicion."""
+    rows = cpg.run(
+        "taint",
+        {"sources": spec.sources, "sinks": spec.sinks, "sanitizers": spec.sanitizers, "function": function},
+    )
+    flows = _flows(rows, function=function)
+    if not flows:
+        return None
+    unsanitized = [flow for flow in flows if not flow["sanitized"]]
+    # Deterministic by construction: sort, then take the first. Nothing here depends on engine ordering.
+    chosen = min(unsanitized or flows, key=_rank)
+    rung = Rung.ENTAILED if not chosen["sanitized"] else Rung.CORROBORATED
+    return Verdict(rung=rung, family=spec.family, witness=_witness(chosen))
+
+
+def _flows(rows: object, *, function: str) -> list[dict[str, object]]:
+    """The query's rows, filtered to the labeled function. A non-list (a failure) yields nothing."""
+    if not isinstance(rows, Sequence) or isinstance(rows, (str, bytes)):
+        return []
+    kept: list[dict[str, object]] = []
+    for row in rows:
+        if not isinstance(row, Mapping):
+            continue
+        if function and str(row.get("sinkMethod", "")) != function:
+            continue
+        kept.append({
+            "sink": str(row.get("sink", "")),
+            "sinkLine": str(row.get("sinkLine", "")),
+            "sinkMethod": str(row.get("sinkMethod", "")),
+            "source": str(row.get("source", "")),
+            "sanitized": bool(row.get("sanitized", False)),
+            "length": _as_int(row.get("length")),
+        })
+    return kept
+
+
+def _as_int(value: object) -> int:
+    return int(value) if isinstance(value, int | float) else 0
+
+
+def _rank(flow: Mapping[str, object]) -> tuple[int, str, str]:
+    """Shortest flow first, then lexicographic — a total order, so two runs cannot disagree."""
+    return (_as_int(flow["length"]), str(flow["source"]), str(flow["sink"]))
+
+
+def _witness(flow: Mapping[str, object]) -> str:
+    line = flow.get("sinkLine") or "?"
+    return f"{flow['source']} -> {flow['sink']} (line {line}, {flow['length']} steps)"
+
+
+__all__ = ["verdict"]
diff --git a/tests/test_cpg_backend.py b/tests/test_cpg_backend.py
new file mode 100644
index 0000000..a5d2d60
--- /dev/null
+++ b/tests/test_cpg_backend.py
@@ -0,0 +1,94 @@
+"""model-grounded-detection task 2.1: the Joern seam, out of process and capability-detected.
+
+Joern is a JVM tool. The whole integration is one subprocess boundary so that the core install stays
+zero-dependency and an absent engine degrades to `suspicion` with a recorded reason rather than a traceback.
+Nothing here imports a JVM binding, and every test runs with Joern absent.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+
+def test_without_joern_the_resolved_backend_is_the_null_one(monkeypatch: pytest.MonkeyPatch) -> None:
+    from openultrasast.cpg.backend import NullBackend, resolve_cpg_backend
+
+    monkeypatch.setenv("OPENULTRASAST_JOERN_PROBE", "off")
+    backend = resolve_cpg_backend()
+    assert isinstance(backend, NullBackend)
+    assert backend.available() is False
+    assert backend.build(Path(".")) is None, "an absent engine yields no CPG, and the caller degrades"
+
+
+def test_the_capability_probe_reuses_the_existing_joern_probe(monkeypatch: pytest.MonkeyPatch) -> None:
+    """`semantic/engines.joern_available` already exists; the model layer must not grow a second probe."""
+    from openultrasast.cpg.capability import has_cpg
+
+    monkeypatch.setenv("OPENULTRASAST_JOERN_PROBE", "off")
+    assert has_cpg() is False
+    monkeypatch.setenv("OPENULTRASAST_JOERN_PROBE", "on")
+    assert has_cpg() is True
+
+
+def test_the_seam_imports_nothing_outside_the_standard_library() -> None:
+    """Req 4.2: the core install has no dependencies, so the seam may not import one at module load."""
+    import ast
+
+    source = Path("src/openultrasast/cpg/backend.py").read_text()
+    imported = set()
+    for node in ast.walk(ast.parse(source)):
+        if isinstance(node, ast.ImportFrom) and node.module and node.level == 0:
+            imported.add(node.module.split(".")[0])
+        elif isinstance(node, ast.Import):
+            imported.update(alias.name.split(".")[0] for alias in node.names)
+    stdlib = {"__future__", "json", "os", "re", "shutil", "subprocess", "tempfile", "dataclasses", "pathlib", "typing",
+              "collections", "logging", "hashlib"}
+    assert imported <= stdlib, f"the seam imports outside the standard library: {sorted(imported - stdlib)}"
+    assert "import jpype" not in source and "py4j" not in source, "Joern is reached by subprocess, never in-process"
+
+
+def test_a_scripted_backend_parses_a_query_result_without_joern(tmp_path: Path) -> None:
+    """The parsing is exercised by canned JSON, so the query contract is testable with no engine installed."""
+    from openultrasast.cpg.backend import CpgResult
+
+    payload = [{"sink": "os.system", "source": "request.args", "path": ["a", "b"], "sanitized": False}]
+
+    def run(query: str, params: dict[str, object]) -> object:
+        assert query == "taint"
+        return payload
+
+    result = CpgResult(cpg_path=tmp_path / "cpg.bin", run=run)
+    assert result.run("taint", {}) == payload
+
+
+def test_joern_output_is_read_from_a_delimited_block_not_from_noisy_stdout() -> None:
+    """Joern prints a banner and a REPL prompt; the payload is fenced so the parse cannot swallow either."""
+    from openultrasast.cpg.backend import BEGIN, END, extract_payload
+
+    noisy = f"Compiling...\nJoern v4.0\n{BEGIN}\n{json.dumps([{'sink': 'eval'}])}\n{END}\njoern> \n"
+    assert extract_payload(noisy) == [{"sink": "eval"}]
+    assert extract_payload("no markers at all") is None
+    assert extract_payload(f"{BEGIN}\nnot json\n{END}") is None
+
+
+def test_a_failing_or_slow_engine_fails_closed(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """Fail closed: a build or query that errors yields no verdict, never a wrong one."""
+    import subprocess
+
+    from openultrasast.cpg.backend import JoernBackend
+
+    def boom(*args: object, **kwargs: object) -> object:
+        raise subprocess.TimeoutExpired(cmd="joern-parse", timeout=1)
+
+    backend = JoernBackend(runner=boom)
+    monkeypatch.setenv("OPENULTRASAST_JOERN_PROBE", "on")
+    assert backend.build(tmp_path) is None
+
+
+def test_the_query_scripts_are_shipped_beside_the_seam() -> None:
+    """The CPGQL is ours and is data: it lives with the package, not inlined in a Python string."""
+    queries = Path("src/openultrasast/cpg/queries")
+    assert (queries / "taint.sc").is_file(), "the injection taint query must ship with the package"
diff --git a/tests/test_model_taint.py b/tests/test_model_taint.py
new file mode 100644
index 0000000..bb0fbb6
--- /dev/null
+++ b/tests/test_model_taint.py
@@ -0,0 +1,113 @@
+"""model-grounded-detection task 2.2: the taint verdict, and what separates entailed from corroborated.
+
+An `ENTAILED` verdict is the static analogue of a reproduced crash: the CPG establishes the flow on its own and
+the LLM's assertion is no longer load-bearing. `CORROBORATED` means a path exists but a sanitizer may break it,
+so the judgement of sufficiency is the LLM's residual — checked against the graph, never averaged over runs.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+
+def _cpg(rows: list[dict[str, object]]):  # type: ignore[no-untyped-def]
+    from openultrasast.cpg.backend import CpgResult
+
+    return CpgResult(cpg_path=Path("cpg.bin"), run=lambda query, params: rows)
+
+
+def _spec():  # type: ignore[no-untyped-def]
+    from openultrasast.model.specs import TaintSpec
+
+    return TaintSpec(
+        family="injection",
+        language="python",
+        sources=("request.args",),
+        sinks=("os.system", "execute"),
+        sanitizers=("escape", "quote"),
+    )
+
+
+def test_an_unsanitized_source_to_sink_path_is_entailed() -> None:
+    from openultrasast.model.ladder import Rung
+    from openultrasast.model.taint import verdict
+
+    rows = [
+        {"sink": "os.system(cmd)", "sinkLine": "4", "sinkMethod": "run", "source": "request.args['x']", "sanitized": False, "length": 3}
+    ]
+    answer = verdict(_cpg(rows), _spec(), function="run")
+    assert answer is not None and answer.rung is Rung.ENTAILED
+    assert answer.family == "injection"
+    assert "request.args" in answer.witness and "os.system" in answer.witness, "the witness names the flow it found"
+
+
+def test_a_path_through_a_sanitizer_is_corroborated_not_entailed() -> None:
+    """The model saw the flow but cannot judge whether the sanitizer is sufficient. That residual is the LLM's."""
+    from openultrasast.model.ladder import Rung
+    from openultrasast.model.taint import verdict
+
+    rows = [{"sink": "execute(q)", "sinkLine": "9", "sinkMethod": "run", "source": "request.args['q']", "sanitized": True, "length": 4}]
+    answer = verdict(_cpg(rows), _spec(), function="run")
+    assert answer is not None and answer.rung is Rung.CORROBORATED
+
+
+def test_no_path_yields_no_verdict_so_the_caller_reports_suspicion() -> None:
+    from openultrasast.model.taint import verdict
+
+    assert verdict(_cpg([]), _spec(), function="run") is None
+
+
+def test_a_query_that_failed_is_not_read_as_an_absence_of_flow() -> None:
+    """`None` from the engine means "could not decide", which is not the same as "no path" — Req 4.3."""
+    from openultrasast.cpg.backend import CpgResult
+    from openultrasast.model.taint import verdict
+
+    broken = CpgResult(cpg_path=Path("cpg.bin"), run=lambda query, params: None)
+    assert verdict(broken, _spec(), function="run") is None
+
+
+def test_the_verdict_is_restricted_to_the_labeled_function() -> None:
+    from openultrasast.model.taint import verdict
+
+    rows = [
+        {
+            "sink": "os.system(x)", "sinkLine": "40", "sinkMethod": "elsewhere",
+            "source": "request.args['x']", "sanitized": False, "length": 2,
+        }
+    ]
+    assert verdict(_cpg(rows), _spec(), function="run") is None, "a flow in another function does not judge this one"
+
+
+def test_the_shortest_unsanitized_flow_wins_and_the_sequence_is_deterministic() -> None:
+    """Req 6.4: two runs over one CPG give an identical verdict sequence, so there is nothing to average."""
+    from openultrasast.model.ladder import Rung
+    from openultrasast.model.taint import verdict
+
+    rows = [
+        {"sink": "os.system(a)", "sinkLine": "7", "sinkMethod": "run", "source": "request.args['b']", "sanitized": True, "length": 9},
+        {"sink": "os.system(a)", "sinkLine": "7", "sinkMethod": "run", "source": "request.args['a']", "sanitized": False, "length": 2},
+    ]
+    first = verdict(_cpg(rows), _spec(), function="run")
+    second = verdict(_cpg(rows), _spec(), function="run")
+    assert first == second
+    assert first is not None and first.rung is Rung.ENTAILED and "request.args['a']" in first.witness
+
+
+def test_the_spec_drives_the_query_parameters() -> None:
+    """The security modelling is ours; the traversal is the engine's. The spec is what crosses the seam."""
+    from openultrasast.cpg.backend import CpgResult
+    from openultrasast.model.taint import verdict
+
+    seen: dict[str, object] = {}
+
+    def run(query: str, params: dict[str, object]) -> object:
+        seen["query"] = query
+        seen.update(params)
+        return []
+
+    verdict(CpgResult(cpg_path=Path("cpg.bin"), run=run), _spec(), function="run")
+    assert seen["query"] == "taint"
+    assert seen["sources"] == ("request.args",)
+    assert seen["sinks"] == ("os.system", "execute")
+    assert seen["sanitizers"] == ("escape", "quote")
+    assert seen["function"] == "run"