diff --git a/src/openultrasast/cpg/backend.py b/src/openultrasast/cpg/backend.py
index e6671f5..00615f6 100644
--- a/src/openultrasast/cpg/backend.py
+++ b/src/openultrasast/cpg/backend.py
@@ -109,7 +109,7 @@ class JoernBackend:
         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)
+        completed = self._run(command, timeout=self.build_timeout, cwd=scratch)
         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)
@@ -125,17 +125,19 @@ class JoernBackend:
         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)
+        # Joern writes a `workspace/` beside the working directory; run it inside the CPG's own scratch dir so
+        # it can never land in the repository being analysed.
+        completed = self._run(command, timeout=self.query_timeout, cwd=cpg_path.parent)
         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:
+    def _run(self, command: list[str], *, timeout: int, cwd: Path | None = None) -> 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)
+            return run(command, capture_output=True, text=True, timeout=timeout, check=False, cwd=str(cwd) if cwd else None)
         except (OSError, subprocess.SubprocessError):
             return None
 
diff --git a/src/openultrasast/cpg/queries/taint.sc b/src/openultrasast/cpg/queries/taint.sc
index 0e1ab41..63fbcfb 100644
--- a/src/openultrasast/cpg/queries/taint.sc
+++ b/src/openultrasast/cpg/queries/taint.sc
@@ -1,55 +1,60 @@
 // 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%.
+// *through resolved dataflow*, and is there a sanitizer on that path. `reachableByFlows` walks the PDG, so a
+// value bound several statements upstream is found:
 //
-// 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
+//     cmd  = request.args["cmd"]        request.args["cmd"] -> cmd -> "echo " + cmd -> full
+//     full = "echo " + cmd
+//     os.system(full)
 //
-// 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.
+// That three-hop chain is exactly the one-hop limit which held our measured entailment at 2.2%.
+//
+// Two things here are easy to get backwards and were, first time round:
+//   * the sink is the call's ARGUMENTS, not the call node -- data flows into arguments;
+//   * `reachableByFlows` is called ON the sink WITH the source, not the other way about.
+//
+// Parameters (comma-separated where plural):
+//   cpgFile, sources, sinks, sanitizers (optional), function (optional: restrict to this enclosing method)
+//
+// Output: a fenced JSON array of {sink, sinkLine, sinkMethod, source, sanitized, length}. The fence exists
+// because Joern prints a banner, pass logs 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
+  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)
+  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 source is any call whose code contains one of the patterns. Matching on `code` rather than a method
+  // name keeps `request.args["x"]` (an indexAccess over a fieldAccess) and `req.body.name` both reachable
+  // without a per-framework rule for each.
+  def sourceNodes = cpg.call.filter(c => sourcePatterns.exists(p => c.code.contains(p)))
 
-  // 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
+  // A sink call: matched by short name (`system`), by leading code (`os.system(...)`), or by resolved
+  // full name (`os.py:<module>.system`), so both bare and dotted forms in the spec hit.
+  def sinkCalls = {
+    val all = cpg.call.filter(c =>
+      sinkNames.exists(n => c.name == n || c.code.startsWith(n + "(") || c.code.startsWith(n) || c.methodFullName.contains(n))
+    )
     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
+  val rows = sinkCalls.l.flatMap { sink =>
+    // Data flows into the arguments; asking the call node itself finds nothing.
+    val flows = sink.argument.reachableByFlows(sourceNodes).l
     flows.map { flow =>
       val elements = flow.elements.map(_.code).l
+      val sanitized = sanitizerNames.nonEmpty && elements.exists(code => sanitizerNames.exists(s => code.contains(s)))
       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),
+        "sanitized"  -> sanitized,
         "length"     -> elements.size
       )
     }
diff --git a/src/openultrasast/model/candidates.py b/src/openultrasast/model/candidates.py
index f2f1ac7..71c022c 100644
--- a/src/openultrasast/model/candidates.py
+++ b/src/openultrasast/model/candidates.py
@@ -18,7 +18,10 @@ from __future__ import annotations
 from collections.abc import Mapping, Sequence
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Literal
+from typing import TYPE_CHECKING, Literal
+
+if TYPE_CHECKING:
+    from .taxonomy import FamilyTaxonomy
 
 CandidateKind = Literal["call", "bind", "operation", "config"]
 
@@ -255,7 +258,7 @@ class CandidateReport:
         }
 
 
-def ceiling(cases: Sequence[object], *, taxonomy: object, generator: str = "ir") -> CandidateReport:
+def ceiling(cases: Sequence[object], *, taxonomy: FamilyTaxonomy, generator: str = "ir") -> CandidateReport:
     """Per family and per slice, the fraction of labeled functions a generator can put a candidate inside.
 
     ``generator`` is ``ir`` or ``ruleset``. The ruleset arm exists so the comparison that rejected it — 12.1%
@@ -269,7 +272,7 @@ def ceiling(cases: Sequence[object], *, taxonomy: object, generator: str = "ir")
     gaps: list[tuple[str, str, str]] = []
     sites: dict[str, list[int]] = {}
     for case in cases:
-        family = labeled_family(case)
+        family = labeled_family(case, taxonomy)
         slice_name = str(getattr(case, "slice", "") or "")
         region_path = str(getattr(case, "relpath", "") or "")
         with tempfile.TemporaryDirectory(prefix="ousast-ceiling-") as scratch:
@@ -329,17 +332,37 @@ def _distribution(counts: list[int]) -> dict[str, int]:
     }
 
 
-def labeled_family(case: object) -> str:
-    """The family a pair is labeled with, or ``unknown``.
+def labeled_family(case: object, taxonomy: FamilyTaxonomy | None = None) -> str:
+    """The family a pair belongs to, derived deterministically: declared label, else obligation, else mechanism,
+    else CWE. ``unknown`` when none of those resolve.
 
-    The deleted ``learning/rounds.py`` fell back to the LLM classifier here. Nothing does now: an unlabeled
-    pair is honestly ``unknown`` rather than guessed, which is the same discipline the evidence ladder applies
-    to a claim the model cannot arbitrate.
+    The deleted ``learning/classify.py`` had exactly this chain and only asked an LLM when every link failed —
+    which, for a labeled corpus, was never. Group 1 first reduced this to the declared label alone, which
+    silently emptied every family whose pairs are labeled by CWE rather than by name; the chain is restored
+    here because it was never the noisy part.
     """
+    from .taxonomy import load_families
+
+    families: FamilyTaxonomy = taxonomy if taxonomy is not None else load_families()
+    declared: list[str] = []
+    derived: list[str] = []
     for row in getattr(case, "expected", ()) or ():
-        family = getattr(row, "family", None)
-        if family:
-            return str(family)
+        label = getattr(row, "family", None)
+        if label and any(family.id == label for family in families.families):
+            declared.append(str(label))
+            continue
+        if getattr(row, "obligation", None):
+            derived.append("access_control")
+        mechanism = getattr(row, "mechanism", None)
+        by_mechanism = families.family_of_mechanism(str(mechanism)) if mechanism else None
+        if by_mechanism is not None and by_mechanism.id != "unknown":
+            derived.append(by_mechanism.id)
+            continue
+        by_cwe = families.family_of_cwe(str(getattr(row, "cwe", "") or ""))
+        if by_cwe is not None and by_cwe.id != "unknown":
+            derived.append(by_cwe.id)
+    for family in declared or derived:
+        return family
     return "unknown"
 
 
diff --git a/src/openultrasast/model/specs.py b/src/openultrasast/model/specs.py
index d8dd24d..6a31e7b 100644
--- a/src/openultrasast/model/specs.py
+++ b/src/openultrasast/model/specs.py
@@ -76,13 +76,28 @@ def classify_guard_text(text: str) -> str:
 
 @dataclass(frozen=True)
 class TaintSpec:
-    """What a flow family's source→sink−sanitizer query matches on, for one language."""
+    """What a flow family's source-to-sink-minus-sanitizer query matches on, for one language.
+
+    ``sanitizers`` holds only *cleansing calls* -- a node on the flow path that makes the value safe, such as
+    ``ast.literal_eval``. It deliberately excludes every fact that carries a **shape qualifier**, because
+    those describe a safe *form of the sink call*, not a call that cleans a value:
+
+        parameterized        ``execute(sql, params)`` binds rather than interpolates
+        literal_format_arg   ``printf("literal", x)`` has a constant format string
+
+    Flattening either kind into the sanitizer list is actively wrong: the sink becomes its own sanitizer, so
+    every flow through it reports as already-clean and nothing is ever entailed. Both were present in the
+    shipped facts (`execute` for Python, the whole printf family for C) and both surfaced on the first live
+    Joern run. The names are kept in ``safe_shape_sinks`` so the shape test can be modelled properly in task
+    4.2 rather than silently mis-modelled here.
+    """
 
     family: str
     language: str
     sources: tuple[str, ...]
     sinks: tuple[str, ...]
     sanitizers: tuple[str, ...]
+    safe_shape_sinks: tuple[str, ...] = ()
 
 
 @dataclass(frozen=True)
@@ -113,7 +128,12 @@ def taint_specs(
     scoped = (facts if facts is not None else load_facts()).for_language(language)
     families = taxonomy if taxonomy is not None else load_families()
     sources = tuple(sorted({pattern for fact in scoped.sources for pattern in fact.patterns}))
-    sanitizers = tuple(sorted({call for fact in scoped.sanitizers for call in fact.calls}))
+    # A shape qualifier means "this sink is safe in this form", never "this call cleans the value".
+    def _is_shape(fact: object) -> bool:
+        return bool(getattr(fact, "parameterized", False)) or getattr(fact, "literal_format_arg", None) is not None
+
+    sanitizers = tuple(sorted({call for fact in scoped.sanitizers if not _is_shape(fact) for call in fact.calls}))
+    safe_shapes = tuple(sorted({call for fact in scoped.sanitizers if _is_shape(fact) for call in fact.calls}))
     by_family: dict[str, set[str]] = {}
     for sink in scoped.sinks:
         family = families.family_of_cwe(sink.cwe)
@@ -121,7 +141,14 @@ def taint_specs(
             continue
         by_family.setdefault(family.id, set()).update(sink.calls)
     return {
-        family_id: TaintSpec(family=family_id, language=language, sources=sources, sinks=tuple(sorted(calls)), sanitizers=sanitizers)
+        family_id: TaintSpec(
+            family=family_id,
+            language=language,
+            sources=sources,
+            sinks=tuple(sorted(calls)),
+            sanitizers=sanitizers,
+            safe_shape_sinks=safe_shapes,
+        )
         for family_id, calls in sorted(by_family.items())
     }
 
diff --git a/tests/test_model_specs.py b/tests/test_model_specs.py
index 93e212d..a4f523c 100644
--- a/tests/test_model_specs.py
+++ b/tests/test_model_specs.py
@@ -99,3 +99,30 @@ def test_the_specs_module_exposes_no_writer() -> None:
     ]
     assert not writers, f"specs must be read-only, found writers: {writers}"
     assert "open(" not in SPECS.read_text().replace("# ", ""), "specs must not open a file for writing"
+
+
+def test_a_sink_is_never_its_own_sanitizer() -> None:
+    """A parameterized fact names a safe sink *shape*, not a cleansing call on the path.
+
+    Two shipped facts have this shape: `parameterized_execute` ("execute with bound parameters is safe") and
+    C's `literal_format` ("printf with a constant format string is safe"). Flattened into the sanitizer list
+    each makes its sink its own sanitizer, so every flow through it reports as already-clean and nothing is
+    ever entailed. Both were caught on the first live Joern run — the second only because the first was fixed
+    as a class rather than as an instance.
+    """
+    from openultrasast.model.specs import taint_specs
+
+    for language in ("python", "javascript", "java", "c"):
+        for spec in taint_specs(language=language).values():
+            overlap = set(spec.sinks) & set(spec.sanitizers)
+            assert not overlap, f"{language}/{spec.family}: sink is its own sanitizer: {sorted(overlap)}"
+
+
+def test_safe_shape_sinks_are_kept_for_the_shape_test() -> None:
+    from openultrasast.model.specs import taint_specs
+
+    spec = taint_specs(language="python")["injection"]
+    assert "execute" in spec.safe_shape_sinks, "the safe-shape names must survive for a later shape test"
+    assert "execute" not in spec.sanitizers
+    c_memory = taint_specs(language="c").get("memory")
+    assert c_memory is not None and "printf" in c_memory.safe_shape_sinks