diff --git a/benchmarks/learn/engine_trace.py b/benchmarks/learn/engine_trace.py
new file mode 100644
index 0000000..377f637
--- /dev/null
+++ b/benchmarks/learn/engine_trace.py
@@ -0,0 +1,841 @@
+"""Resumable, serial labelled-function traces; host fetches pinned sources, Docker stays offline.
+
+Pair blob identities resolve through catalogs to real source commits. The analyzer
+is git archive HEAD src plus a hashed snapshot of this uncommitted tool.
+"""
+
+from __future__ import annotations
+
+import argparse
+import ast
+import fcntl
+import hashlib
+import json
+import os
+import re
+import shutil
+import subprocess
+import tempfile
+import textwrap
+import time
+import tomllib
+from collections import Counter, defaultdict
+from pathlib import Path
+
+from engine_trace_worker import EXCLUDED_DIRS, safe_path, unit_record, write_json
+
+from openultrasast.learn.examples import read_jsonl
+from openultrasast.learn.labels import load_sources, repo_name
+from openultrasast.model.specs import taint_specs
+from openultrasast.model.taint import request_params
+from openultrasast.pairs import _materialize_side, load_pair_catalog
+from openultrasast.plane.engine_alerts import export_pin
+from openultrasast.plane.memory import FileStore
+from openultrasast.plane.tasks.alerts import engine_languages
+from openultrasast.preprocess import detect_language
+
+ROOT = Path(__file__).resolve().parents[2]
+DEFAULT_OUT = Path.home() / "ousast-results/plane/engine-trace"
+SNAPSHOT_FILES = (
+    "benchmarks/learn/engine_trace.py",
+    "benchmarks/learn/engine_trace_worker.py",
+    "benchmarks/learn/engine_trace_parse.py",
+    "src/openultrasast/model/taint.py",
+    "src/openultrasast/cpg/queries/taint.sc",
+)
+MATERIALIZATION_VERSION = 3
+EXPORT_CAP = 20 * 1024 * 1024
+
+STATUSES = ("path", "asked-nothing", "failed", "timeout", "unsupported")
+
+
+def digest(value):
+    return hashlib.sha256(json.dumps(value, sort_keys=True).encode()).hexdigest()
+
+
+def pin_name(pin):
+    return digest([pin["repo"], pin["pin"]]) + ".json"
+
+
+def join_units(units, examples, include_typescript=False):
+    index = {row["id"]: row for row in examples}
+    missing = [u["unit"] for u in units if u["unit"] not in index]
+    if missing:
+        raise ValueError(f"example snapshot missing {len(missing)}/{len(units)} unit IDs; supply --examples JSONL; first: {missing[0]}")
+    covered = engine_languages() - {"typescript"}
+    if include_typescript:
+        covered |= {"typescript"}
+    result = []
+    for unit in units:
+        example = index[unit["unit"]]
+        if any(example[key] != unit[key] for key in ("family", "label", "group")):
+            raise ValueError(f"unit/example identity mismatch: {unit['unit']}")
+        file, sep, function = example["candidate"].partition("::")
+        if not sep or not function:
+            raise ValueError(f"unit has no labelled function: {unit['unit']}")
+        language = detect_language(Path(file)) or example.get("language", "other")
+        model_language = "javascript" if language == "typescript" else language
+        spec = taint_specs(language=model_language).get(unit["family"]) if language in covered else None
+        result.append(
+            {
+                **unit,
+                "repo": example["repo"],
+                "pin": example["pin"],
+                "file": file,
+                "function": function,
+                "language": language,
+                "source": example["source"],
+                "excerpt": example.get("excerpt", example["source"] == "pairs" or example["source"].startswith("advisory-fixes")),
+                "pin_role": example.get("pin_role", ""),
+                "supported": spec is not None,
+                "reason": "" if spec else f"no admitted taint model: {language}/{unit['family']}",
+            }
+        )
+    return result
+
+
+def make_plan(
+    units,
+    out,
+    only_family=None,
+    limit=0,
+    *,
+    include_typescript=False,
+    only_source=(),
+    rerun_status=("failed",),
+    allow_excerpt_fallback=False,
+):
+    grouped = defaultdict(list)
+    for unit in units:
+        if (not only_family or unit["family"] == only_family) and (not only_source or unit["source"] in only_source):
+            # Include admission options even when a unit's support happens to stay
+            # unchanged. Old results cannot mark a different selection as complete.
+            identity = {
+                "unit": unit,
+                "include_typescript": include_typescript,
+                "materialization_version": MATERIALIZATION_VERSION,
+                "allow_excerpt_fallback": allow_excerpt_fallback,
+            }
+            grouped[unit["repo"], unit["pin"]].append({**unit, "input_digest": digest(identity)})
+    plan = []
+    for (repo, pin), rows in sorted(grouped.items()):
+        item = {
+            "repo": repo,
+            "pin": pin,
+            "materialization_version": MATERIALIZATION_VERSION,
+            "units": sorted(rows, key=lambda r: r["unit"]),
+        }
+        item["selection"] = digest(item["units"])
+        done = out / pin_name(item)
+        if done.exists():
+            old = json.loads(done.read_text())
+            recorded = {r["unit"]: r.get("input_digest") for r in old.get("units", [])}
+            if (
+                old.get("done")
+                and old.get("materialization_version") == MATERIALIZATION_VERSION
+                and not any(r.get("status") in rerun_status for r in old.get("units", []) if r["unit"] in {u["unit"] for u in rows})
+                and all(recorded.get(r["unit"]) == r["input_digest"] for r in rows)
+            ):
+                continue
+        sources = {r["source"] for r in rows}
+        # Inventory medians: harvest sides 31s, advisory sides 47s; v1 pins 182s, v2 pins 498s.
+        item["estimated_seconds"] = (
+            (47 if any(s.startswith("advisory") for s in sources) else 31)
+            if any(r.get("excerpt") for r in rows) or any(s.startswith("advisory") for s in sources)
+            else (182 if sources == {"population-v1"} else 498)
+        )
+        if not any(r["supported"] for r in rows):
+            item["estimated_seconds"] = 0
+        plan.append(item)
+    # limit counts runnable pins, not unsupported records.
+    if limit:
+        chosen = []
+        count = 0
+        for pin in plan:
+            runnable = any(r["supported"] for r in pin["units"])
+            if runnable and count >= limit:
+                continue
+            chosen.append(pin)
+            count += runnable
+        return chosen
+    return plan
+
+
+class Inputs:
+    """Resolve spent corpora; fetch pair source commits into the host cache."""
+
+    def __init__(self, root, cache, allow_excerpt_fallback=False):
+        self.cache = cache
+        self.allow_excerpt_fallback = allow_excerpt_fallback
+        self.catalog_rows = defaultdict(list)
+        for catalog in (root / "benchmarks/pairs").rglob("catalog.toml"):
+            for item in tomllib.loads(catalog.read_text()).get("pair", []):
+                self.catalog_rows[item["name"]].append((catalog.parent.name, item))
+        self.pairs = []
+        self.repos = defaultdict(list)
+        sources = load_sources()
+        for source in sources.sources:
+            if source.kind == "pairs":
+                self.pairs.extend(load_pair_catalog(root / source.files[0]))
+            elif source.kind == "population":
+                if source.id not in {"population-v1", "population-v2"}:
+                    continue
+                data = tomllib.loads((root / source.files[0]).read_text())
+                for case in data.get("case", []):
+                    self.repos[repo_name(case["repo"])].append((cache / "independent" / case["id"], case))
+            elif source.kind == "recipes":
+                for file in source.files:
+                    data = tomllib.loads((root / file).read_text())
+                    recipe = data.get("repo", data)
+                    # Recipes carry a top-level URL and a pinned checkout cache.
+                    if isinstance(recipe, dict):
+                        self.repos[repo_name(str(recipe.get("url", "")))].append((cache / "repos" / str(data.get("name", "")), data))
+        self.hashes = {}
+
+    def pair_for(self, row):
+        if row["pin_role"] not in {"fixed", "vulnerable", "vuln"}:
+            raise ValueError(f"ambiguous pair side: {row['pin_role']}")
+        side = "fixed" if row["pin_role"] == "fixed" else "vuln"
+        candidates = [
+            c for c in self.pairs if c.relpath == row["file"] and repo_name(c.repo or f"{c.slice}:{c.name}") == repo_name(row["repo"])
+        ]
+        matches = []
+        for case in candidates:
+            path = case.fixed_file if side == "fixed" else case.vuln_file
+            if path.is_file():
+                if path not in self.hashes:
+                    data = path.read_bytes()
+                    # Harvest hashes decoded text, so universal newline conversion is part
+                    # of its identity (one recorded ThreatByte excerpt uses CRLF).
+                    normalized = path.read_text().encode("utf-8")
+                    self.hashes[path] = {
+                        hashlib.sha1(b"blob " + str(len(body)).encode() + b"\0" + body).hexdigest() for body in (data, normalized)
+                    }
+                if row["pin"] in self.hashes[path]:
+                    matches.append(case)
+        if len(matches) == 1:
+            return matches[0], side
+        if len(candidates) == 1:
+            # Earlier stores used label_pin rather than the excerpt digest.
+            from openultrasast.learn.examples import label_pin
+
+            case = candidates[0]
+            if row["pin"] == label_pin({"source": "pairs", "source_ref": case.name, "pin_role": row["pin_role"]}):
+                return case, side
+        raise ValueError(f"cannot uniquely resolve excerpt {row['repo']}@{row['pin']}:{row['file']}")
+
+    def catalog_for(self, case):
+        matches = [
+            (source, item)
+            for source, item in self.catalog_rows[case.name]
+            if item.get("relpath") == case.relpath and repo_name(item.get("repo", "")) == repo_name(case.repo)
+        ]
+        if len(matches) != 1:
+            raise ValueError(f"ambiguous catalog source for {case.name}")
+        return matches[0]
+
+    def source_identity(self, case, side):
+        _, item = self.catalog_for(case)
+        parent, commit = item.get("parent", ""), item.get("commit", "")
+        if side not in {"vuln", "fixed"} or not all(re.fullmatch(r"[0-9a-fA-F]{40}", sha) for sha in (parent, commit)) or parent == commit:
+            raise ValueError(f"ambiguous catalog commits for {case.name}: need distinct parent and commit SHAs")
+        repo = item.get("repo", "")
+        if not repo:
+            raise ValueError(f"missing catalog repository for {case.name}")
+        return repo, parent if side == "vuln" else commit
+
+    def fetch_source(self, repo, commit, roots):
+        clone = self.cache / "trace-sources" / digest([repo, commit, roots])
+        clone.mkdir(parents=True, exist_ok=True)
+        url = repo if repo.startswith("https://") else f"https://github.com/{repo.removesuffix('.git')}.git"
+        env = {**os.environ, "GIT_TERMINAL_PROMPT": "0"}
+        env.pop("GIT_NO_LAZY_FETCH", None)  # sparse checkout may fetch the selected blobs
+
+        def git(*args):
+            return subprocess.run(["git", "-C", str(clone), *args], env=env, capture_output=True, text=True, check=True, timeout=300)
+
+        git("init")
+        git("config", "remote.origin.url", url)
+        git("config", "remote.origin.promisor", "true")
+        git("config", "remote.origin.partialclonefilter", "blob:none")
+        git("fetch", "--depth", "1", "--filter=blob:none", url, commit)
+        git("sparse-checkout", "init", "--cone")
+        git("sparse-checkout", "set", "--", *roots)
+        git("checkout", "--detach", "--force", commit)
+        actual = git("rev-parse", "HEAD").stdout.strip()
+        if actual != commit:
+            raise ValueError(f"fetched commit mismatch: {actual} != {commit}")
+        return clone
+
+    def export_local(self, clone, pin, target):
+        with tempfile.TemporaryDirectory(prefix="trace-export-", dir=target.parent) as scratch:
+            export_pin(clone, pin["pin"], Path(scratch))
+            roots = sorted({source_root(r["file"], r["language"]) for r in pin["units"]})
+            proof = export_sources(Path(scratch), target, pin["units"], roots)
+        return {**proof, "mode": "source", "commit": pin["pin"]}
+
+    def materialize(self, pin, target):
+        rows = pin["units"]
+        for row in rows:
+            safe_path(target, row["file"])
+            if row.get("mapping_error"):
+                raise ValueError(row["mapping_error"])
+        if any(r.get("excerpt") or r["source"].startswith("advisory-fixes") for r in rows):
+            resolved = [self.pair_for(row) for row in rows]
+            identities = set()
+            catalog_files = []
+            for case, side in resolved:
+                _, item = self.catalog_for(case)
+                parent, commit = item.get("parent", ""), item.get("commit", "")
+                if parent != commit and all(re.fullmatch(r"[0-9a-fA-F]{40}", sha) for sha in (parent, commit)):
+                    identities.add(self.source_identity(case, side))
+                else:
+                    file = case.fixed_file if side == "fixed" else case.vuln_file
+                    data = file.read_text()
+                    if is_catalog_excerpt(data):
+                        data = parseable_excerpt(file, data)
+                        mode = "excerpt-parseable"
+                    else:
+                        # Validate BOTH twins before treating this as a complete-file pair.
+                        for twin in (case.vuln_file, case.fixed_file):
+                            verify_catalog_file(twin)
+                        mode = "catalog-file"
+                    catalog_files.append((case, side, mode, data))
+            if catalog_files:
+                if identities:
+                    raise ValueError("mixed source and catalog-file materialization at one pair pin")
+                for case, side, mode, data in catalog_files:
+                    _materialize_side(target, case, side=side)
+                    if mode == "excerpt-parseable":
+                        safe_path(target, case.relpath).write_text(data)
+                verify_functions(target, rows)
+                modes = {case.relpath: mode for case, _, mode, _ in catalog_files}
+                for row in rows:
+                    if modes[row["file"]] == "excerpt-parseable":
+                        row["evidence_scope"] = "in-function-only"
+                return {
+                    "mode": "excerpt-parseable" if "excerpt-parseable" in modes.values() else "catalog-file",
+                    "files": sum(p.is_file() for p in target.rglob("*")),
+                    "bytes": sum(p.stat().st_size for p in target.rglob("*") if p.is_file()),
+                    "roots": sorted({source_root(r["file"], r["language"]) for r in rows}),
+                }
+            if len(identities) != 1:
+                raise ValueError("ambiguous source commits at one pair pin")
+            repo, commit = identities.pop()
+            roots = sorted({source_root(row["file"], row["language"]) for row in rows})
+            try:
+                clone = self.fetch_source(repo, commit, roots)
+                proof = export_sources(clone, target, rows, roots)
+            except (OSError, ValueError, subprocess.SubprocessError) as exc:
+                if not self.allow_excerpt_fallback:
+                    raise
+                shutil.rmtree(target, ignore_errors=True)
+                for case, side in resolved:
+                    _materialize_side(target, case, side=side)
+                return {"mode": "excerpt-fallback", "reason": str(exc), "repo": repo, "commit": commit}
+            verify_functions(target, rows)
+            return {**proof, "mode": "source", "repo": repo, "commit": commit}
+        if all(r["source"] == "dev-php" for r in rows):
+            # Export the exact requested commit from the local recipe checkout.
+            for recipe in (ROOT / "benchmarks/repos").glob("*.toml"):
+                data = tomllib.loads(recipe.read_text())
+                if repo_name(str(data.get("url", ""))) == repo_name(pin["repo"]):
+                    checkout = self.cache / "repos" / data["name"] / pin["pin"][:12]
+                    return self.export_local(checkout, pin, target)
+        for clone, _ in self.repos[repo_name(pin["repo"])]:
+            check = subprocess.run(["git", "-C", str(clone), "cat-file", "-e", pin["pin"] + "^{commit}"], capture_output=True)
+            if check.returncode == 0:
+                return self.export_local(clone, pin, target)
+        raise ValueError(f"no local clone contains {pin['repo']}@{pin['pin']}")
+
+
+def is_catalog_excerpt(data):
+    # Older Python exports have this header but no upstream_start field.
+    return re.match(r"\s*(?://|#)\s*Provenance:", data) is not None
+
+
+def parseable_excerpt(path, data):
+    """Dedent the code independently of its unindented provenance comments."""
+    language = detect_language(path)
+    lines = data.splitlines(keepends=True)
+    start = 0
+    while start < len(lines) and (not lines[start].strip() or lines[start].lstrip().startswith(("#", "//"))):
+        start += 1
+    header = "".join(lines[:start])
+    if language == "python":
+        header = re.sub(r"(?m)^(\s*)//", r"\1#", header)
+    data = header + textwrap.dedent("".join(lines[start:]))
+    if not "".join(lines[start:]).strip():
+        raise ValueError(f"excerpt does not parse: empty code ({path.name})")
+    if language == "python":
+        try:
+            ast.parse(data, filename=str(path))
+        except SyntaxError as exc:
+            raise ValueError(f"excerpt does not parse: {exc}") from exc
+    else:
+        from openultrasast.semantic.cst import parse_with_cst
+
+        parsed = parse_with_cst(str(path), data, language)
+        if parsed is None:
+            raise ValueError(f"excerpt does not parse: no cheap parser for {language} ({path.name})")
+        if not parsed.parse_ok:
+            raise ValueError(f"excerpt does not parse: {language} {parsed.reason} ({path.name})")
+    return data
+
+
+def verify_catalog_file(path):
+    """Admit complete source only; syntactically valid provenance excerpts still fail."""
+    data = path.read_text()
+    if is_catalog_excerpt(data):
+        raise ValueError(f"catalog side is an upstream excerpt without distinct commits: {path.name}")
+    language = detect_language(path)
+    if not data.strip():
+        raise ValueError(f"catalog side is empty: {path.name}")
+    if language == "python":
+        try:
+            ast.parse(data, filename=str(path))
+        except SyntaxError as exc:
+            raise ValueError(f"catalog side is not complete Python source: {path.name}: {exc.msg}") from exc
+    else:
+        from openultrasast.semantic.extra import grammar_for
+
+        grammar = grammar_for(language)
+        if grammar is None:
+            raise ValueError(f"cannot verify complete catalog source for {language}: {path.name}")
+        from tree_sitter import Parser
+
+        if Parser(grammar).parse(data.encode()).root_node.has_error:
+            raise ValueError(f"catalog side is not complete {language} source: {path.name}")
+
+
+def source_root(file, language):
+    parts = Path(file).parts
+    if language == "java":
+        for i in range(len(parts) - 2):
+            if parts[i : i + 3] == ("src", "main", "java"):
+                return "/".join(parts[: i + 3])
+    # Monorepo and source-layout containers are not themselves a package.
+    if len(parts) > 2 and parts[0] in {"packages", "apps", "src", "web"}:
+        return "/".join(parts[:2])
+    return parts[0] if len(parts) > 1 else "."
+
+
+def local_imports(source, rows):
+    """Cheap one-level Python imports from labelled files, without importing code."""
+    found = set()
+    for row in rows:
+        if row["language"] != "python":
+            continue
+        file = safe_path(source, row["file"])
+        try:
+            tree = ast.parse(file.read_bytes())
+        except (SyntaxError, ValueError):
+            continue
+        for node in ast.walk(tree):
+            if isinstance(node, ast.Import):
+                modules, bases = [alias.name for alias in node.names], [source, source / "src"]
+            elif isinstance(node, ast.ImportFrom):
+                modules = [node.module or ""]
+                modules += [".".join(filter(None, (node.module, alias.name))) for alias in node.names if alias.name != "*"]
+                bases = [file.parents[node.level - 1]] if node.level and node.level <= len(file.parents) else [source, source / "src"]
+            else:
+                continue
+            for base in bases:
+                for module in modules:
+                    path = base.joinpath(*module.split("."))
+                    for candidate in (path.with_suffix(".py"), path / "__init__.py"):
+                        if candidate.is_relative_to(source) and candidate.is_file() and not candidate.is_symlink():
+                            rel = candidate.relative_to(source).as_posix()
+                            if not any(part in EXCLUDED_DIRS for part in Path(rel).parts):
+                                found.add(rel)
+    return found
+
+
+def export_sources(source, target, rows, roots=(".",), cap=EXPORT_CAP):
+    """Copy labels intact, narrowing oversized roots before bounding context bytes."""
+    labelled = {r["file"] for r in rows}
+    languages = {r["language"] for r in rows}
+    files = set()
+    protected = {parent.as_posix() for rel in labelled for parent in Path(rel).parents}
+    for directory, dirs, names in os.walk(source):
+        dirs[:] = sorted(
+            d
+            for d in dirs
+            if d != ".git"
+            and not (Path(directory) / d).is_symlink()
+            and (d not in EXCLUDED_DIRS or (Path(directory) / d).relative_to(source).as_posix() in protected)
+        )
+        for name in names:
+            file = Path(directory) / name
+            rel = file.relative_to(source).as_posix()
+            excluded = any(part in EXCLUDED_DIRS for part in Path(rel).parts[:-1])
+            if not file.is_symlink() and (
+                rel in labelled
+                or (
+                    not excluded
+                    and detect_language(file) in languages
+                    and any(root == "." or rel == root or rel.startswith(root + "/") for root in roots)
+                )
+            ):
+                files.add(rel)
+    missing = labelled - files
+    if missing:
+        raise ValueError(f"labelled files missing or excluded from export: {sorted(missing)}")
+    imports = local_imports(source, rows)
+    files.update(imports)
+    sizes = {rel: safe_path(source, rel).stat().st_size for rel in files}
+    original_roots = sorted(roots)
+    roots = original_roots
+    # Avoid lexicographically truncating a whole large package such as vllm.
+    if sum(sizes.values()) > cap:
+        roots = sorted({str(Path(rel).parent) for rel in labelled})
+        files = {
+            rel for rel in files if rel in labelled or rel in imports or any(root == "." or rel.startswith(root + "/") for root in roots)
+        }
+    proof = {
+        "bytes": 0,
+        "files": 0,
+        "cap_bytes": cap,
+        "cap_hit": False,
+        "excludes": sorted(EXCLUDED_DIRS),
+        "roots": roots,
+        "requested_roots": original_roots,
+        "scope_narrowed": roots != original_roots,
+        "import_files": sorted(imports),
+    }
+    for rel in sorted(files, key=lambda rel: (rel not in labelled, rel)):
+        file = safe_path(source, rel)
+        size = sizes[rel]
+        # The cap bounds context only. No labelled source is ever cut or omitted.
+        if proof["bytes"] + size > cap:
+            proof["cap_hit"] = True
+            if rel not in labelled:
+                continue
+        dest = safe_path(target, rel)
+        dest.parent.mkdir(parents=True, exist_ok=True)
+        dest.write_bytes(file.read_bytes())
+        proof["bytes"] += size
+        proof["files"] += 1
+    return proof
+
+
+def verify_functions(root, rows):
+    for row in rows:
+        data = safe_path(root, row["file"]).read_text(errors="replace")
+        if not re.search(r"(?<![\w$])" + re.escape(row["function"]) + r"(?![\w$])", data):
+            raise ValueError(f"labelled function not found in fetched source: {row['file']}::{row['function']}")
+
+
+def freeze_source(target):
+    target.mkdir(parents=True)
+    archive = target.parent / "source.tar"
+    with archive.open("wb") as stream:
+        subprocess.run(["git", "-C", str(ROOT), "archive", "HEAD", "src"], stdout=stream, check=True)
+    subprocess.run(["tar", "-xf", str(archive), "-C", str(target)], check=True)
+    archive.unlink()
+    overlay = {}
+    for name in SNAPSHOT_FILES:
+        data = (ROOT / name).read_bytes()
+        dest = target / name
+        dest.parent.mkdir(parents=True, exist_ok=True)
+        dest.write_bytes(data)
+        overlay[name] = hashlib.sha256(data).hexdigest()
+    head = subprocess.run(["git", "-C", str(ROOT), "rev-parse", "HEAD"], capture_output=True, text=True, check=True).stdout.strip()
+    return {"head": head, "overlay_sha256": overlay}
+
+
+def docker_command(source, checkout, out, name, image, deadline, question_deadline):
+    return [
+        "docker",
+        "run",
+        "--rm",
+        "--init",
+        "--name",
+        name,
+        "--network",
+        "none",
+        "--memory",
+        "3g",
+        "--pull",
+        "never",
+        "--entrypoint",
+        "python",
+        "-e",
+        "PYTHONPATH=/frozen/src",
+        "-v",
+        f"{source}:/frozen:ro",
+        "-v",
+        f"{checkout}:/case:ro",
+        "-v",
+        f"{out}:/out",
+        image,
+        "/frozen/benchmarks/learn/engine_trace_worker.py",
+        "--worker",
+        "/out/pin.json",
+        "--deadline",
+        str(deadline),
+        "--question-deadline",
+        str(question_deadline),
+    ]
+
+
+def prepare_questions(pin, root, question_deadline):
+    """Prepare metadata on the host, before entering the dependency-minimal image."""
+    active = [u for u in pin["units"] if u["supported"]]
+    from openultrasast.learn.features import entry_names
+    from openultrasast.mapping import php_hook_callbacks
+    from openultrasast.plane.harvest import declares
+    from openultrasast.preprocess import preprocess_repository
+
+    _, targets = preprocess_repository(root)
+    entries, _ = entry_names(root, {u["file"] for u in active})
+    hooks = ";".join(f"{h}:{f}" for h, functions in php_hook_callbacks(root, targets).items() for f in functions)
+    requests = {}
+    for u in active:
+        specs = taint_specs(language="javascript" if u["language"] == "typescript" else u["language"])
+        if not declares(safe_path(root, u["file"]).read_text(errors="replace").splitlines(), u["file"], u["function"]):
+            raise ValueError(f"labelled function not declared: {u['file']}::{u['function']}")
+        excerpt = u.get("evidence_scope") == "in-function-only" or pin.get("materialization", {}).get("mode") == "excerpt-fallback"
+        params = request_params(
+            specs[u["family"]],
+            file=u["file"],
+            function=u["function"],
+            trace=True,
+            parameter_sources=excerpt or u["function"] in entries,
+            call_depth=0 if excerpt else 3,
+            hook_callbacks=hooks,
+        )
+        params["questionDeadline"] = str(question_deadline)
+        requests[u["unit"]] = params
+    return requests
+
+
+def run_pin(pin, source, provenance, inputs, out, args):
+    with tempfile.TemporaryDirectory(prefix="trace-pin-", dir=out) as scratch:
+        scratch = Path(scratch)
+        checkout, output = scratch / "case", scratch / "output"
+        output.mkdir()
+        try:
+            pin = {**pin, "materialization": inputs.materialize(pin, checkout)}
+            write_json(output / "pin.json", {**pin, "questions": prepare_questions(pin, checkout, args.question_deadline)})
+            name = "ousast-trace-" + pin_name(pin)[:16]
+            command = docker_command(source, checkout, output, name, args.image, args.deadline, args.question_deadline)
+            started = time.monotonic()
+            done = None
+            try:
+                done = subprocess.run(command, capture_output=True, text=True, timeout=args.deadline)
+                status, reason = "failed", f"container exit {done.returncode}: {done.stderr[-1000:]}"
+                (output / "container.log").write_text(done.stdout + "\n" + done.stderr)
+            except subprocess.TimeoutExpired:
+                done = None
+                status, reason = "timeout", f"container deadline {args.deadline}s"
+            finally:
+                # Never start the next pin unless this container is known to be gone.
+                try:
+                    removed = subprocess.run(["docker", "rm", "-f", name], capture_output=True, timeout=20)
+                except subprocess.SubprocessError as exc:
+                    raise RuntimeError(f"cannot confirm container {name} stopped; refusing another launch") from exc
+                if done is None and removed.returncode != 0:
+                    raise RuntimeError(f"cannot remove timed-out container {name}; refusing another launch")
+            result_path = output / "result.json"
+            if not result_path.exists():
+                reason = f"container produced no result: {reason}"
+            record = json.loads(result_path.read_text()) if result_path.exists() else {**pin, "units": []}
+            if not record.get("done"):
+                previous = {u["unit"]: u for u in record["units"]}
+                record["units"] = []
+                for unit in pin["units"]:
+                    row = previous.get(unit["unit"], unit_record(unit, status))
+                    if row["status"] not in {"path", "asked-nothing"}:
+                        row.update(
+                            status=status if unit["supported"] else "unsupported", reason=reason if unit["supported"] else unit["reason"]
+                        )
+                    record["units"].append(row)
+            record["container_exit"] = done.returncode if done else None
+            if done is not None and done.returncode != 0:
+                for row in record["units"]:
+                    if row["supported"]:
+                        row.update(status="failed", reason=reason)
+            record.update(done=True, seconds=time.monotonic() - started, analyzer=provenance, image=args.image)
+            if (output / "container.log").exists():
+                shutil.copyfile(output / "container.log", out / pin_name(pin).replace(".json", ".log"))
+        except (OSError, ValueError, subprocess.SubprocessError) as exc:
+            record = {
+                **pin,
+                "done": True,
+                "analyzer": provenance,
+                "units": [unit_record(u, "failed" if u["supported"] else "unsupported", str(exc)) for u in pin["units"]],
+            }
+        save_pin(out / pin_name(pin), record)
+        print(
+            json.dumps({"repo": pin["repo"], "pin": pin["pin"], "statuses": dict(Counter(u["status"] for u in record["units"]))}),
+            flush=True,
+        )
+        return any(u["status"] in {"failed", "timeout"} for u in record["units"])
+
+
+def save_pin(path, record):
+    if path.exists():
+        old = json.loads(path.read_text())
+        selected = {u["unit"] for u in record["units"]}
+        record["units"].extend(u for u in old.get("units", []) if u["unit"] not in selected)
+    write_json(path, record)
+
+
+def summary(out):
+    if not out.is_dir():
+        raise ValueError(f"results root does not exist: {out}")
+    family, source = defaultdict(Counter), defaultdict(Counter)
+    pairs = defaultdict(dict)
+    units = {}
+    result_files = 0
+    for path in sorted(out.glob("*.json")):
+        record = json.loads(path.read_text())
+        if not record.get("done"):
+            continue
+        result_files += 1
+        for unit in record["units"]:
+            if unit["unit"] in units:
+                raise ValueError(f"duplicate result for {unit['unit']}")
+            units[unit["unit"]] = unit
+            family[unit["family"]][unit["status"]] += 1
+            source[unit["source"]][unit["status"]] += 1
+            if unit["pair"]:
+                pairs[unit["family"], unit["pair"]][unit["label"]] = unit["status"]
+    asymmetry = Counter()
+    completed = Counter()
+    for sides in pairs.values():
+        if set(sides) != {0, 1}:
+            asymmetry["incomplete-pair"] += 1
+            continue
+        key = {(True, False): "vulnerable-only", (False, True): "fixed-only", (True, True): "both", (False, False): "neither"}[
+            sides[1] == "path", sides[0] == "path"
+        ]
+        asymmetry[key] += 1
+        if all(s in {"path", "asked-nothing"} for s in sides.values()):
+            completed[key] += 1
+    return {
+        "result_files_read": result_files,
+        "units": len(units),
+        "by_family": {k: {s: v[s] for s in STATUSES} for k, v in sorted(family.items())},
+        "by_source": {k: {s: v[s] for s in STATUSES} for k, v in sorted(source.items())},
+        "pair_asymmetry": dict(asymmetry),
+        "pair_asymmetry_both_answered": dict(completed),
+    }
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--units", type=Path, default=ROOT / "plane/experiments/exp-005-graph-slice.units.jsonl")
+    parser.add_argument("--examples", type=Path, help="frozen example JSONL instead of the local memory store")
+    parser.add_argument("--memory", type=Path, default=Path.home() / "ousast-results/plane/memory")
+    parser.add_argument("--cache", type=Path, default=Path.home() / ".cache/openultrasast")
+    parser.add_argument("--out", type=Path, default=DEFAULT_OUT)
+    parser.add_argument("--image", default="openultrasast:dev")
+    parser.add_argument("--deadline", type=int, default=900)
+    parser.add_argument("--question-deadline", type=int, default=120)
+    parser.add_argument("--limit", type=int, default=0)
+    parser.add_argument("--only-family")
+    parser.add_argument("--only-source", default="", help="comma-separated source catalog IDs")
+    parser.add_argument("--rerun-status", default="failed,timeout", help="comma-separated statuses to re-run (including timeout)")
+    parser.add_argument(
+        "--allow-excerpt-fallback", action="store_true", help="allow fragments only if fetching/exporting resolved source fails"
+    )
+    parser.add_argument("--include-typescript", action="store_true")
+    parser.add_argument("--dry-run", action="store_true")
+    parser.add_argument("--summary", action="store_true")
+    args = parser.parse_args()
+    if min(args.deadline, args.question_deadline) <= 0 or args.limit < 0:
+        parser.error("deadlines must be positive; limit must be nonnegative")
+    rerun_status = set(filter(None, args.rerun_status.split(",")))
+    if rerun_status - set(STATUSES):
+        parser.error("unknown --rerun-status value")
+    if args.summary:
+        print(json.dumps(summary(args.out), indent=2))
+        return 0
+    # Even inherited opt-ins must not make catalog resolution fetch.
+    os.environ["GIT_NO_LAZY_FETCH"] = "1"
+    os.environ.pop("OPENULTRASAST_PAIRS_NETWORK", None)
+    os.environ.pop("OPENULTRASAST_REPOS_NETWORK", None)
+    examples = read_jsonl(args.examples) if args.examples else [r.row for r in FileStore(args.memory).rows(kind="example")]
+    try:
+        units = join_units(read_jsonl(args.units), examples, args.include_typescript)
+    except ValueError as exc:
+        parser.error(str(exc))
+    inputs = Inputs(ROOT, args.cache, args.allow_excerpt_fallback)
+    # Resolve pair source names for leak-audit coverage before calculating the resume key.
+    for unit in units:
+        if unit.get("excerpt") or unit["source"].startswith("advisory-fixes"):
+            try:
+                case, _ = inputs.pair_for(unit)
+                catalog_source, _ = inputs.catalog_for(case)
+                unit["source"] = catalog_source if catalog_source.startswith("advisory-fixes") else case.slice
+                unit["excerpt"] = True
+            except ValueError as exc:
+                unit["mapping_error"] = str(exc)
+    plan = make_plan(
+        units,
+        args.out,
+        args.only_family,
+        args.limit,
+        include_typescript=args.include_typescript,
+        only_source=set(filter(None, args.only_source.split(","))),
+        rerun_status=rerun_status,
+        allow_excerpt_fallback=args.allow_excerpt_fallback,
+    )
+    # Planning is read-only, including unsupported pins. All persistence stays below this return.
+    if args.dry_run:
+        print(
+            json.dumps(
+                {
+                    "pins": len(plan),
+                    "runnable_pins": sum(any(u["supported"] for u in p["units"]) for p in plan),
+                    "units": sum(len(p["units"]) for p in plan),
+                    "estimated_seconds": sum(p["estimated_seconds"] for p in plan),
+                    "estimate_basis": "inventory medians: excerpt 31/47s, repository 182/498s; not a deadline guarantee",
+                    "plan": plan,
+                },
+                indent=2,
+            )
+        )
+        return 0
+    args.out.mkdir(parents=True, exist_ok=True)
+    lock = (args.out / ".lock").open("w")
+    try:
+        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+    except BlockingIOError:
+        parser.error("another trace runner holds this results root")
+    failures = False
+    if any(u["supported"] for pin in plan for u in pin["units"]):
+        image = subprocess.run(
+            ["docker", "image", "inspect", "--format", "{{.Id}}", args.image], capture_output=True, text=True, check=True
+        )
+        args.image = image.stdout.strip()
+        if not args.image:
+            raise ValueError("Docker returned no local image identity")
+    with tempfile.TemporaryDirectory(prefix="trace-source-", dir=args.out) as scratch:
+        source = Path(scratch) / "frozen"
+        provenance = freeze_source(source)
+        progress = {"pins_done": 0, "pins_total": len(plan), "last_pin": None, "statuses": {}}
+        counts = Counter()
+        write_json(args.out / "progress.json", progress)
+        for pin in plan:
+            if (args.out / "STOP").exists():
+                break
+            if not any(u["supported"] for u in pin["units"]):
+                save_pin(
+                    args.out / pin_name(pin),
+                    {**pin, "done": True, "units": [unit_record(u, "unsupported", u["reason"]) for u in pin["units"]]},
+                )
+            else:
+                failures |= run_pin(pin, source, provenance, inputs, args.out, args)
+            result = json.loads((args.out / pin_name(pin)).read_text())
+            selected = {u["unit"] for u in pin["units"]}
+            counts.update(u["status"] for u in result["units"] if u["unit"] in selected)
+            progress.update(pins_done=progress["pins_done"] + 1, last_pin={"repo": pin["repo"], "pin": pin["pin"]}, statuses=dict(counts))
+            write_json(args.out / "progress.json", progress)
+    return 2 if failures else 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/benchmarks/learn/engine_trace_parse.py b/benchmarks/learn/engine_trace_parse.py
new file mode 100644
index 0000000..1522ae4
--- /dev/null
+++ b/benchmarks/learn/engine_trace_parse.py
@@ -0,0 +1,51 @@
+"""Lossless parsing of the optional shortest-path trace, with explicit evidence limits."""
+
+from __future__ import annotations
+
+import re
+from collections.abc import Mapping
+from typing import Any
+
+_STEP = re.compile(r"^(.*):(-?\d+) \[([^\]]+)\] ([\s\S]*)$")
+
+
+def parse_trace(row: Mapping[str, Any]) -> dict[str, Any]:
+    """Preserve row-level discharge evidence; never guess which node sanitized a path.
+
+    The query emits basename-only files and truncated code. A bound can be off-path;
+    it remains in ``guard`` even when no trace step exactly matches it. Invalid steps
+    raise instead of turning an unreadable trace into an empty path.
+    """
+    raw = row.get("trace", [])
+    if not isinstance(raw, list):
+        raise ValueError("trace must be an array")
+    steps = []
+    bound = str(row.get("bound") or "")
+    for index, text in enumerate(raw):
+        match = _STEP.fullmatch(text) if isinstance(text, str) else None
+        if match is None:
+            raise ValueError(f"invalid trace step: {text!r}")
+        location, line, node_label, code = match.groups()
+        file, separator, method = location.partition(":")
+        roles = ["source"] if index == 0 else ["step"]
+        if index == len(raw) - 1:
+            roles = ["source", "sink"] if index == 0 else ["sink"]
+        if bound and code == bound:
+            roles.append("guard")
+        steps.append(
+            {
+                "file": file,
+                "method": method if separator else "?",
+                "line": int(line),
+                "node_label": node_label,
+                "code": code,
+                "roles": roles,
+                "raw": text,
+            }
+        )
+    return {
+        "steps": steps,
+        "sanitized": bool(row.get("sanitized", False)),
+        "sanitizer": {"on_path": bool(row.get("sanitized", False)), "step": None},
+        "guard": {"text": bound, "bounded": bool(row.get("bounded", False))},
+    }
diff --git a/benchmarks/learn/engine_trace_worker.py b/benchmarks/learn/engine_trace_worker.py
new file mode 100644
index 0000000..ca8e9ed
--- /dev/null
+++ b/benchmarks/learn/engine_trace_worker.py
@@ -0,0 +1,453 @@
+"""Container-only Joern trace worker; host prepares all corpus and question metadata.
+
+Keep this module independent of learn/plane and their optional dependencies. The
+frozen host invokes this file directly in the shipped, offline analyzer image.
+"""
+
+from __future__ import annotations
+
+import argparse
+import contextlib
+import hashlib
+import json
+import os
+import selectors
+import signal
+import subprocess
+import tempfile
+import time
+from collections import Counter
+from pathlib import Path
+
+from engine_trace_parse import parse_trace
+
+from openultrasast.cpg.backend import DATAFLOW_OVERLAY, JoernBackend, _render
+
+
+def write_json(path, value):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    tmp = path.with_suffix(".tmp")
+    tmp.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n")
+    tmp.replace(path)
+
+
+def safe_path(root, name):
+    path = (root / name).resolve()
+    if not path.is_relative_to(root.resolve()):
+        raise ValueError(f"path escapes input root: {name}")
+    return path
+
+
+FRONTENDS = {"php2cpg", "pysrc2cpg", "jssrc2cpg", "javasrc2cpg", "c2cpg", "joern-parse"}
+MIN_CPG_BYTES = 1024
+HEAP_MB = 2560
+ENGINE_BUG = "engine bug: jssrc2cpg ObjectPropertyCallLinker"
+EXCLUDED_DIRS = frozenset({"tests", "test", "docs", "node_modules", "vendor", "dist", "build", "templates", "migrations"})
+
+
+def instrument_failure(proof):
+    if any(stage.get("oom") for stage in proof.get("jvm", [])):
+        return "JVM out of memory"
+    if not proof.get("files") or not proof.get("bytes"):
+        return "instrument read zero files or zero bytes"
+    if proof.get("cpg_bytes", 0) <= MIN_CPG_BYTES:
+        return "frontend wrote no CPG"
+    for stage in proof.get("jvm", []):
+        if stage["command"] in FRONTENDS:
+            if stage.get("cpg_bytes", 0) <= MIN_CPG_BYTES:
+                return "frontend wrote no CPG"
+        elif stage["command"] == "joern" and stage["seconds"] < 5:
+            return f"implausibly fast JVM step: {stage['command']} in {stage['seconds']:.3f}s (<5s)"
+    if not proof.get("jvm"):
+        return "no JVM step recorded"
+    return ""
+
+
+class MeasuredBackend(JoernBackend):
+    """Use the shipped backend with a process watchdog, no concurrent query threads."""
+
+    def __init__(self, deadline, question_deadline, checkpoint):
+        super().__init__(build_timeout=deadline, query_timeout=deadline, session_transport=False, heap_mb=HEAP_MB)
+        self.deadline = time.monotonic() + deadline
+        self.question_deadline = question_deadline
+        self.checkpoint = checkpoint
+        self.stages = []
+        self.asked = []
+        self.answers = {}
+        self.completion_counts = Counter()
+        self.timed_out = False
+        self.fatal = ""
+        self.labelled_files = []
+        self.retry_pending = False
+        self.retries = []
+
+    def _apply_overlays(self, cpg_path, scratch):
+        result = super()._apply_overlays(cpg_path, scratch)
+        # An optional overlay failure must not prevent the taint-only attempt.
+        if self.fatal == ENGINE_BUG and not self.timed_out:
+            self.retry_pending = True
+            self.fatal = ""
+        return result
+
+    def _batch_once(self, cpg_path, query, requests):
+        if self.fatal or self.timed_out:
+            return None
+        if not self.retry_pending:
+            answers = super()._batch_once(cpg_path, query, requests)
+            if self.fatal != ENGINE_BUG:
+                return answers
+        if query != "taint" or self.timed_out or self.retries:
+            self.fatal = ENGINE_BUG
+            return None
+        self.retry_pending = False
+        self.fatal = ""
+        retry = {"reason": ENGINE_BUG, "mode": "taint-only", "completed": False}
+        self.retries.append(retry)
+        # Do not reuse rows/markers from the failed attempt as completion evidence.
+        self.asked.clear()
+        self.answers.clear()
+        self.completion_counts.clear()
+        original = self.queries_dir
+        with tempfile.TemporaryDirectory(prefix="trace-taint-", dir=cpg_path.parent) as scratch:
+            script = (original / "taint.sc").read_text()
+            marker = "importCpg(cpgFile)"
+            if script.count(marker) != 1:
+                self.fatal = ENGINE_BUG
+                return None
+            # Loading without enhancement skips the failing JS postprocessing passes.
+            # A raw graph without dataflow cannot establish a negative answer.
+            script = script.replace(
+                marker,
+                "importCpg(cpgFile, enhance = false)\n"
+                f'  require(cpg.metaData.overlays.l.contains("{DATAFLOW_OVERLAY}"), '
+                '"taint-only retry requires saved dataflowOss overlay")',
+            )
+            self.queries_dir = Path(scratch)
+            (self.queries_dir / "taint.sc").write_text(script)
+            try:
+                answers = super()._batch_once(cpg_path, query, requests)
+            finally:
+                self.queries_dir = original
+        retry["completed"] = answers is not None and not self.fatal and not self.timed_out
+        if not retry["completed"]:
+            self.fatal = ENGINE_BUG
+        return answers
+
+    def _run(self, command, *, timeout, cwd=None):
+        if self.fatal or self.timed_out:
+            return None
+        if time.monotonic() >= self.deadline:
+            self.timed_out = True
+            return None
+        started = time.monotonic()
+        end = min(self.deadline, started + timeout)
+        question_end = end
+        binary = Path(command[0]).name
+        frontend = binary in FRONTENDS
+        query = binary == "joern" and "--script" in command
+        output = bytearray()
+        pending = b""
+        process = subprocess.Popen(
+            command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, cwd=cwd, env=self._jvm_env(), start_new_session=True
+        )
+        try:
+            with selectors.DefaultSelector() as selector:
+                selector.register(process.stdout, selectors.EVENT_READ)
+                while selector.get_map():
+                    if time.monotonic() >= min(end, question_end):
+                        self.timed_out = True
+                        break
+                    for key, _ in selector.select(min(0.2, max(0, min(end, question_end) - time.monotonic()))):
+                        chunk = os.read(key.fileobj.fileno(), 65536)
+                        if not chunk:
+                            selector.unregister(key.fileobj)
+                            continue
+                        output.extend(chunk)
+                        pending += chunk
+                        while b"\n" in pending:
+                            line, pending = pending.split(b"\n", 1)
+                            try:
+                                event = json.loads(line)
+                            except ValueError:
+                                continue
+                            if not isinstance(event, dict):
+                                continue
+                            if "__question__" in event:
+                                self.asked.append(event["__question__"])
+                                question_end = min(end, time.monotonic() + self.question_deadline)
+                            if "id" in event and "rows" in event:
+                                self.answers[event["id"]] = event["rows"]
+                                self.completion_counts[event["id"]] += 1
+                                question_end = end
+                            self.checkpoint()
+            if not self.timed_out:
+                process.wait(timeout=max(0.01, end - time.monotonic()))
+        finally:
+            with contextlib.suppress(ProcessLookupError):
+                os.killpg(process.pid, signal.SIGKILL)
+            process.wait(timeout=5)
+            process.stdout.close()
+        elapsed = time.monotonic() - started
+        if frontend or query:
+            stage = {
+                "command": binary,
+                "argv": command,
+                "heap_mb": self._heap_mb(),
+                "java_tool_options": self._jvm_env()["JAVA_TOOL_OPTIONS"],
+                "seconds": elapsed,
+                "exit": process.returncode,
+                "oom": b"OutOfMemoryError" in output,
+            }
+            lines = output.decode(errors="replace").splitlines()
+            stage["labelled_file_logs"] = {
+                file: [line for line in lines if file in line or Path(file).name in line]
+                for file in self.labelled_files
+                if any(file in line or Path(file).name in line for line in lines)
+            }
+            self.stages.append(stage)
+            if query and process.returncode != 0:
+                # stderr is merged into stdout above; drop JVM frames before taking
+                # the tail so they cannot crowd out the exception's message.
+                lines = [
+                    line.strip()
+                    for line in output.decode(errors="replace").splitlines()
+                    if line.strip() and not line.strip().startswith(("at ", "... "))
+                ]
+                stage["output_tail"] = "\n".join(lines[-20:])
+                if not self.timed_out:
+                    self.fatal = f"joern query exited {process.returncode}"
+            if frontend:
+                # Capture the exact frontend output before build failure cleanup or overlays.
+                output_arg = next((i + 1 for i, arg in enumerate(command[:-1]) if arg in {"-o", "--output"}), None)
+                cpg_path = (Path(cwd or Path.cwd()) / command[output_arg]).resolve() if output_arg is not None else None
+                stage["cpg_path"] = str(cpg_path) if cpg_path is not None else None
+                stage["cpg_bytes"] = cpg_path.stat().st_size if cpg_path is not None and cpg_path.is_file() else 0
+                if stage["cpg_bytes"] <= MIN_CPG_BYTES:
+                    self.fatal = "frontend wrote no CPG"
+            elif elapsed < 5 and not self.fatal:
+                self.fatal = f"implausibly fast JVM step: {command[0]} in {elapsed:.3f}s (<5s)"
+        if b"ObjectPropertyCallLinker failed" in output and b"Assignment statement with 3 arguments" in output:
+            self.fatal = ENGINE_BUG
+            stage["engine_bug"] = ENGINE_BUG
+        if b"OutOfMemoryError" in output:
+            self.fatal = "JVM out of memory"
+        self.checkpoint()
+        return subprocess.CompletedProcess(command, process.returncode if not self.fatal else 1, output.decode(errors="replace"), "")
+
+
+def unit_record(unit, status, reason="", **extra):
+    return {
+        **unit,
+        "status": status,
+        "reason": reason,
+        "witness_rows": [],
+        "traces": [],
+        "instrument": {"files": [], "bytes": 0, "cpg_path": None, "cpg_bytes": 0, "jvm": []},
+        "questions": {},
+        "questions_asked": [],
+        "questions_completed": [],
+        **extra,
+    }
+
+
+def worker_language(pin, root, out, deadline, question_deadline, persist=write_json):
+    proof = {
+        "files": [],
+        "bytes": 0,
+        "cpg_path": None,
+        "cpg_bytes": 0,
+        "jvm": [],
+        "heap_mb": HEAP_MB,
+        "excludes": sorted(EXCLUDED_DIRS),
+        "materialization": pin.get("materialization", {}),
+    }
+    record = {
+        **pin,
+        "done": False,
+        "instrument": proof,
+        "units": [unit_record(u, "unsupported" if not u["supported"] else "timeout", u["reason"]) for u in pin["units"]],
+    }
+
+    def checkpoint():
+        # Checkpoints retain raw completed rows even if the outer container deadline fires.
+        # They remain timeout until the finished JVM's instrument proof is validated.
+        for stage in backend.stages:
+            if "cpg_path" in stage:
+                proof["cpg_path"] = stage["cpg_path"]
+                proof["cpg_bytes"] = stage["cpg_bytes"]
+        for row in record["units"]:
+            rid = row["unit"]
+            row["instrument"] = proof
+            row["questions_asked"] = [rid] if rid in backend.asked else []
+            if rid in backend.answers:
+                row["questions_completed"] = [rid]
+                row["witness_rows"] = [r for r in backend.answers[rid] if isinstance(r, dict) and "kind" not in r and "sink" in r]
+        persist(out, record)
+
+    backend = MeasuredBackend(deadline, question_deadline, checkpoint)
+    proof["jvm"] = backend.stages
+    proof["retries"] = getattr(backend, "retries", [])
+    backend.labelled_files = sorted({u["file"] for u in pin["units"] if u["supported"]})
+    cpg = None
+    try:
+        active = [u for u in pin["units"] if u["supported"]]
+        for file in sorted({u["file"] for u in active}):
+            data = safe_path(root, file).read_bytes()
+            proof["files"].append({"file": file, "bytes": len(data), "sha256": hashlib.sha256(data).hexdigest()})
+            proof["bytes"] += len(data)
+            print(f"read {file}: {len(data)} bytes", flush=True)
+        if not proof["bytes"]:
+            raise ValueError("instrument read zero files or zero bytes")
+        languages = {"javascript" if u["language"] == "typescript" else u["language"] for u in active}
+        if len(languages) != 1:
+            raise ValueError(f"one pin has multiple frontend languages: {sorted(languages)}")
+        language = next(iter(languages))
+        requests = {rid: {key: _render(value) for key, value in params.items()} for rid, params in pin["questions"].items()}
+        if set(requests) != {u["unit"] for u in active}:
+            raise ValueError("prepared questions do not match supported units")
+        record["questions"] = requests
+        for unit_row in record["units"]:
+            unit_row["questions"] = {unit_row["unit"]: requests[unit_row["unit"]]} if unit_row["unit"] in requests else {}
+        persist(out, record)
+        cpg = backend.build(root, language=language)
+        if cpg is None:
+            raise ValueError(backend.fatal or backend.last_failure or ("pin deadline" if backend.timed_out else "CPG build failed"))
+        if proof["cpg_path"] is None:
+            proof["cpg_path"] = str(cpg.cpg_path)
+            proof["cpg_bytes"] = cpg.cpg_path.stat().st_size if cpg.cpg_path.is_file() else 0
+        failure = instrument_failure(proof)
+        if failure:
+            raise ValueError(failure)
+        if cpg.run_batch_once is None:
+            raise ValueError("backend offers no batch operation")
+        answers = cpg.run_batch_once("taint", requests) or {}
+        record["census"] = answers.get("__census__", [])
+        failure = backend.fatal or instrument_failure(proof)
+        if not backend.asked and not backend.timed_out:
+            failure = failure or "query produced no markers"
+        census = record["census"]
+        if census and int(census[0].get("files", 0)) <= 0:
+            failure = failure or "query census read zero files"
+        elif not census and not backend.timed_out:
+            failure = failure or "query census is missing"
+        for index, unit in enumerate(pin["units"]):
+            if not unit["supported"]:
+                continue
+            rid = unit["unit"]
+            extra = {
+                "questions": {rid: requests[rid]},
+                "questions_asked": [rid] if rid in backend.asked else [],
+                "questions_completed": [rid] if rid in answers else [],
+                "instrument": proof,
+            }
+            rows = answers.get(rid)
+            # A batch across backend shards is complete only when every shard answered.
+            # The legacy merger can return a union even when one shard was killed.
+            shards = int(census[0].get("shards", 1)) if census else 1
+            shard_incomplete = shards > 1 and getattr(backend, "completion_counts", {}).get(rid, 0) < shards
+            if rows is not None:
+                raw_flows = [r for r in rows if isinstance(r, dict) and "kind" not in r and "sink" in r]
+                extra.update(witness_rows=raw_flows, traces=[parse_trace(r) for r in raw_flows])
+            if shard_incomplete:
+                extra["questions_completed"] = []
+            names = census[0].get("file_names", []) if census else []
+            file_missing = not any(str(p).removeprefix("./") == unit["file"] or str(p).endswith("/" + unit["file"]) for p in names)
+            dropped = file_missing and rows is not None
+            if failure or dropped:
+                status, reason = "failed", failure or "labelled file was dropped or absent from query census"
+                if dropped:
+                    logs = [line for stage in backend.stages for line in stage.get("labelled_file_logs", {}).get(unit["file"], [])]
+                    reason += "\n" + ("\n".join(dict.fromkeys(logs)) if logs else "frontend emitted no log lines mentioning labelled file")
+            elif rows is None or shard_incomplete:
+                status, reason = ("timeout" if backend.timed_out else "failed"), "question did not complete"
+            elif not rows and rid not in backend.asked:
+                status, reason = "failed", "query produced no markers"
+            else:
+                flows = [r for r in rows if isinstance(r, dict) and "kind" not in r and "sink" in r]
+                if any(not row.get("trace") for row in flows):
+                    raise ValueError("taint witness returned without requested trace")
+                extra.update(witness_rows=flows, traces=[parse_trace(r) for r in flows])
+                status, reason = ("path" if flows else "asked-nothing"), ""
+            record["units"][index] = unit_record(unit, status, reason, **extra)
+    except (OSError, ValueError, RuntimeError) as exc:
+        for index, unit in enumerate(pin["units"]):
+            if unit["supported"]:
+                record["units"][index] = unit_record(
+                    unit,
+                    "timeout" if backend.timed_out and not backend.fatal else "failed",
+                    str(exc),
+                    instrument=proof,
+                    questions={unit["unit"]: record["questions"][unit["unit"]]} if unit["unit"] in record.get("questions", {}) else {},
+                    witness_rows=[r for r in backend.answers.get(unit["unit"], []) if isinstance(r, dict) and "sink" in r],
+                    questions_completed=[unit["unit"]] if unit["unit"] in backend.answers else [],
+                    questions_asked=[unit["unit"]] if unit["unit"] in backend.asked else [],
+                )
+    finally:
+        if cpg and cpg.cleanup:
+            cpg.cleanup()
+        for stage in backend.stages:
+            if stage["command"] == "joern" and stage.get("exit") and "output_tail" in stage:
+                detail = f"joern query exited {stage['exit']}: {stage['output_tail'] or 'no output'}"
+                for row in record["units"]:
+                    if row["supported"] and row["status"] in {"failed", "timeout"}:
+                        row["reason"] = f"{row['reason']}\n{detail}".strip()
+        record["questions_asked"] = list(dict.fromkeys(backend.asked))
+        record["questions_completed"] = [rid for u in record["units"] for rid in u["questions_completed"]]
+        record["done"] = True
+        persist(out, record)
+    return record
+
+
+def worker(pin, root, out, deadline, question_deadline):
+    """Run language partitions sequentially under one shared container deadline."""
+    languages = sorted({"javascript" if u["language"] == "typescript" else u["language"] for u in pin["units"] if u["supported"]})
+    if len(languages) <= 1:
+        return worker_language(pin, root, out, deadline, question_deadline)
+    end = time.monotonic() + deadline
+    record = {
+        **pin,
+        "done": False,
+        "units": [
+            unit_record(u, "timeout" if u["supported"] else "unsupported", "pin deadline" if u["supported"] else u["reason"])
+            for u in pin["units"]
+        ],
+        "instruments": {},
+    }
+
+    def persist(_, partial):
+        updates = {u["unit"]: u for u in partial["units"]}
+        record["units"] = [updates.get(u["unit"], u) for u in record["units"]]
+        record["instruments"][language] = partial["instrument"]
+        record["questions_asked"] = [rid for u in record["units"] for rid in u["questions_asked"]]
+        record["questions_completed"] = [rid for u in record["units"] for rid in u["questions_completed"]]
+        write_json(out, record)
+
+    write_json(out, record)
+    for language in languages:
+        remaining = end - time.monotonic()
+        if remaining <= 0:
+            break
+        units = [
+            u for u in pin["units"] if u["supported"] and ("javascript" if u["language"] == "typescript" else u["language"]) == language
+        ]
+        questions = {u["unit"]: pin["questions"][u["unit"]] for u in units}
+        worker_language({**pin, "units": units, "questions": questions}, root, out, remaining, question_deadline, persist)
+    record["done"] = True
+    write_json(out, record)
+    return record
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--worker", type=Path, required=True)
+    parser.add_argument("--deadline", type=int, default=900)
+    parser.add_argument("--question-deadline", type=int, default=120)
+    args = parser.parse_args()
+    if min(args.deadline, args.question_deadline) <= 0:
+        parser.error("deadlines must be positive")
+    worker(json.loads(args.worker.read_text()), Path("/case"), Path("/out/result.json"), args.deadline, args.question_deadline)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/src/openultrasast/cpg/queries/config.sc b/src/openultrasast/cpg/queries/config.sc
index 70b517a..3b00c97 100644
--- a/src/openultrasast/cpg/queries/config.sc
+++ b/src/openultrasast/cpg/queries/config.sc
@@ -53,7 +53,8 @@
     }
 
   val calls = {
-    val all = cpg.call.filter(c => matches(c.name, c.code)).l
+    // CPS-IT/mailqueue Extension::KEY has a null .method; exclude methodless calls before attribution.
+    val all = cpg.call.filter(c => Option(c.method).nonEmpty && matches(c.name, c.code)).l
     val inFile = if (fileS.isEmpty) all else all.filter(_.method.filename.endsWith(fileS))
     if (function.isEmpty) inFile else inFile.filter(_.method.name == function)
   }
diff --git a/src/openultrasast/cpg/queries/dominance.sc b/src/openultrasast/cpg/queries/dominance.sc
index 34fb0ef..23eb179 100644
--- a/src/openultrasast/cpg/queries/dominance.sc
+++ b/src/openultrasast/cpg/queries/dominance.sc
@@ -172,7 +172,8 @@
   // repository this was measured on, and unscoped rows attributed the model's unguarded query to the
   // handler. The sibling comparison is a statement about a FILE contradicting itself, which is what the
   // rung's justification says, so the file is also the right scope for it.
-  val allOperations = cpg.call.filter(isOperation).l
+  // CPS-IT/mailqueue Extension::KEY has a null .method; exclude methodless calls before attribution.
+  val allOperations = cpg.call.filter(c => Option(c.method).nonEmpty).filter(isOperation).l
   val opCalls = if (fileS.isEmpty) allOperations else allOperations.filter(_.method.filename.endsWith(fileS))
 
   // ---- The module boundary an application puts between its route and its query --------------------------
@@ -236,7 +237,7 @@
     if (fileS.isEmpty) Nil
     else
       fileMethods(fileS).flatMap { m =>
-        m.call.l.flatMap { via =>
+        m.call.l.filter(c => Option(c.method).nonEmpty).flatMap { via =>
           via.callee.l
             .filter(_.isExternal)
             .flatMap(stub => stubTarget(stub.fullName))
diff --git a/src/openultrasast/cpg/queries/taint.sc b/src/openultrasast/cpg/queries/taint.sc
index 5c13769..d7c155a 100644
--- a/src/openultrasast/cpg/queries/taint.sc
+++ b/src/openultrasast/cpg/queries/taint.sc
@@ -56,7 +56,8 @@
     contextFilter: String = "",
     contextPaths: String = "",
     requests: String = "",
-    requestsFile: String = ""
+    requestsFile: String = "",
+    trace: String = ""
 ) = {
   importCpg(cpgFile)
 
@@ -225,6 +226,7 @@
   // module region exists to cover the code that is in NO function, so its membership is direct.
   val moduleScope = function == "<module>"
 
+  // CPS-IT/mailqueue Extension::KEY is a class-level PHP call with a null .method; exclude it from scope.
   def nestedInLabeled(m: io.shiftleft.codepropertygraph.generated.nodes.Method): Boolean =
     if (moduleScope) m.name == function
     else
@@ -495,7 +497,7 @@
       case c: io.shiftleft.codepropertygraph.generated.nodes.Call if originAnchors.contains(c.name) => true
       case c: io.shiftleft.codepropertygraph.generated.nodes.Call if concatenations.contains(c.name) =>
         c.argument.l.sortBy(_.argumentIndex).headOption.exists(a => fixesOrigin(a, depth + 1))
-      case i: io.shiftleft.codepropertygraph.generated.nodes.Identifier =>
+      case i: io.shiftleft.codepropertygraph.generated.nodes.Identifier if i.method != null =>
         // A variable fixes the origin when EVERY assignment to it in the method does.
         val writes = i.method.ast.isCall.nameExact(ASSIGNMENT).l.filter(_.argument.l.sortBy(_.argumentIndex).headOption.exists {
           case t: io.shiftleft.codepropertygraph.generated.nodes.Identifier => t.name == i.name
@@ -789,7 +791,7 @@
         case _ => false
       }
       ternary || (end :: elements.filter(stored)).distinctBy(_.id).exists {
-        case cfg: io.shiftleft.codepropertygraph.generated.nodes.CfgNode =>
+        case cfg: io.shiftleft.codepropertygraph.generated.nodes.CfgNode if cfg.method != null =>
           val method = cfg.method
           method.ast.isControlStructure.controlStructureType("IF").l.exists { cs =>
             cs.condition.l.headOption.exists { cond =>
@@ -897,7 +899,7 @@
   def fieldSourceNodes =
     if (fieldSrc != "true") Iterator.empty
     else {
-      val reads = cpg.call.nameExact(FIELD_ACCESS).filter(c => inScope(c.method)).l.filterNot(writeTarget)
+      val reads = cpg.call.nameExact(FIELD_ACCESS).filter(c => Option(c.method).exists(inScope)).l.filterNot(writeTarget)
       if (reads.isEmpty) Iterator.empty
       else reads.filter(r => taintedPrefixes(r.code.trim, fileS, r.method) && !overwrittenBefore(r)).iterator
     }
@@ -957,7 +959,7 @@
     val framework: List[io.shiftleft.codepropertygraph.generated.nodes.CfgNode] =
       callback.ast.isCall.filter(isSourceCall).l
     val fields: List[io.shiftleft.codepropertygraph.generated.nodes.CfgNode] =
-      callback.ast.isCall.nameExact(FIELD_ACCESS).l.filter(r => taintedPrefixes(r.code.trim, callback.filename, r.method))
+      callback.ast.isCall.nameExact(FIELD_ACCESS).l.filter(r => Option(r.method).exists(m => taintedPrefixes(r.code.trim, callback.filename, m)))
     framework ++ fields
   }
 
@@ -1012,7 +1014,7 @@
   def hookSourceNodes =
     if (hookTable.isEmpty || hookApply.isEmpty) Iterator.empty
     else {
-      val applies = cpg.call.filter(c => hookApply.contains(c.name)).filter(c => inScope(c.method)).l
+      val applies = cpg.call.filter(c => hookApply.contains(c.name)).filter(c => Option(c.method).exists(inScope)).l
       if (applies.isEmpty) Iterator.empty
       else {
         val keys = applies.flatMap { call =>
@@ -1023,7 +1025,7 @@
         else
           cpg.call
             .nameExact(INDEX_ACCESS)
-            .filter(c => inScope(c.method))
+            .filter(c => Option(c.method).exists(inScope))
             .l
             .filter(access => keys.contains(keyOf(access)))
             // ...and actually derived from the filtered value, not merely sharing a key name with it.
@@ -1107,7 +1109,7 @@
   def sinkCalls =
     sinkCandidatesMemo
       .getOrElseUpdate(sinksS, cpg.call.filter(c => sinkNames.exists(n => sinkMatches(c, n))).l)
-      .filter(c => inScope(c.method))
+      .filter(c => Option(c.method).exists(inScope))
 
   // ---- stage one of the two-stage join WITHOUT dataflow, for evidence mode (flow-aware-ranking, phase 2) --
   //
@@ -1169,7 +1171,7 @@
     )
 
   def fieldCarriedKind: String =
-    strongest(cpg.call.nameExact(FIELD_ACCESS).filter(c => inScope(c.method)).l.map(r => prefixKind(r.code.trim, fedFields(r.method.filename))))
+    strongest(cpg.call.nameExact(FIELD_ACCESS).filter(c => Option(c.method).exists(inScope)).l.map(r => prefixKind(r.code.trim, fedFields(r.method.filename))))
 
   def callbackFedKind(callback: String): String =
     callbackFedMemo.getOrElseUpdate(
@@ -1186,7 +1188,7 @@
   def hookCarriedKind: String =
     if (hookTable.isEmpty || hookApply.isEmpty) ""
     else
-      strongest(cpg.call.filter(c => hookApply.contains(c.name)).filter(c => inScope(c.method)).l.flatMap { call =>
+      strongest(cpg.call.filter(c => hookApply.contains(c.name)).filter(c => Option(c.method).exists(inScope)).l.flatMap { call =>
         val hook = call.argument.l.headOption.map(a => unquote(a.code)).getOrElse("")
         hookTable.getOrElse(hook, Nil).map(callbackFedKind)
       })
@@ -1232,7 +1234,7 @@
       else {
         val confirmed = cpg.call
           .nameExact(FIELD_ACCESS)
-          .filter(c => inScope(c.method))
+          .filter(c => Option(c.method).exists(inScope))
           .l
           .exists(r => prefixKind(r.code.trim, fedFields(r.method.filename)) == "parameter" && taintedPrefixes(r.code.trim, r.method.filename, r.method))
         if (confirmed) "source" else "parameter"
@@ -1276,9 +1278,9 @@
       // Field carrying is deliberately same-file in the existing abstraction. Retain
       // its file context conservatively, never claim every method carries the value.
       val fieldFiles = if (fieldCarriedKind.isEmpty) Set.empty[String] else
-        cpg.call.nameExact(FIELD_ACCESS).filter(c => inScope(c.method)).l.map(_.method.filename).toSet
+        cpg.call.nameExact(FIELD_ACCESS).filter(c => Option(c.method).exists(inScope)).l.map(_.method.filename).toSet
       val callbacks = if (hookApply.isEmpty) Nil else cpg.call.filter(c => hookApply.contains(c.name))
-        .filter(c => inScope(c.method)).l.flatMap { call =>
+        .filter(c => Option(c.method).exists(inScope)).l.flatMap { call =>
           val hook = call.argument.l.headOption.map(a => unquote(a.code)).getOrElse("")
           hookTable.getOrElse(hook, Nil).flatMap(name => cpg.method.nameExact(name).filterNot(_.isExternal).l)
         }.distinct
@@ -1400,7 +1402,7 @@
     def traced(flow: io.joern.dataflowengineoss.language.Path): ujson.Arr =
       ujson.Arr.from(spliced(flow.elements.l).map { node =>
         val where = node match {
-          case c: io.shiftleft.codepropertygraph.generated.nodes.CfgNode => c.method.filename.split("/").last + ":" + c.method.name
+          case c: io.shiftleft.codepropertygraph.generated.nodes.CfgNode => Option(c.method).map(m => m.filename.split("/").last + ":" + m.name).getOrElse("?")
           case _                                                          => "?"
         }
         s"$where:${node.lineNumber.getOrElse(-1)} [${node.label}] ${node.code.take(120)}"
@@ -1435,7 +1437,7 @@
           "sinkArg0Literal" -> arg0Literal,
           "bounded"         -> bounds.nonEmpty,
           "bound"            -> bounds.headOption.getOrElse("").take(120),
-          "inLabeledScope"  -> (function.isEmpty || nestedInLabeled(sink.method) || reachableMethods.contains(sink.method.fullName))
+          "inLabeledScope"  -> (function.isEmpty || Option(sink.method).exists(nestedInLabeled) || reachableMethods.contains(sink.method.fullName))
         )
         if (traceS == "true") row("trace") = traced(shortest)
         row
@@ -1489,6 +1491,12 @@
       // collapsed the sizer to three sink visits per JVM start on the plugin.
       val started = System.nanoTime()
       def field(name: String): String = req.obj.get(name).map(_.str).getOrElse("")
+      // Opt-in heartbeat for the trace runner's per-question process watchdog.
+      // Killing the JVM preserves completed streamed rows without unsafe query threads.
+      if (field("questionDeadline").nonEmpty) {
+        println(ujson.write(ujson.Obj("__question__" -> id)))
+        System.out.flush()
+      }
       val paramSrc = req.obj.get("parameterSources").map(_.str).getOrElse("false")
       val fieldParamSrc = req.obj.get("fieldParameterSources").map(_.str).getOrElse(paramSrc)
       val fieldSrc = req.obj.get("fieldSources").map(_.str).getOrElse("true")
@@ -1532,7 +1540,7 @@
   } else {
     println("---OUSAST-CPG-BEGIN---")
     println(
-      ujson.write(ujson.Arr(rowsFor(sources, sinks, sanitizers, hookCallbacks, dispatchApply, dispatchValue, function, parameterSources, parameterSources, fieldSources, evidenceOnly, file, callDepth, boundedSinks): _*))
+      ujson.write(ujson.Arr(rowsFor(sources, sinks, sanitizers, hookCallbacks, dispatchApply, dispatchValue, function, parameterSources, parameterSources, fieldSources, evidenceOnly, file, callDepth, boundedSinks, traceS = trace): _*))
     )
   }
   println("---OUSAST-CPG-END---")
diff --git a/src/openultrasast/cpg/queries/vocabulary.sc b/src/openultrasast/cpg/queries/vocabulary.sc
index 77fd498..f874005 100644
--- a/src/openultrasast/cpg/queries/vocabulary.sc
+++ b/src/openultrasast/cpg/queries/vocabulary.sc
@@ -51,11 +51,13 @@
   val sourceMemo = scala.collection.mutable.Map.empty[String, Boolean]
   def nearSource(c: io.shiftleft.codepropertygraph.generated.nodes.Call): Boolean =
     if (sourcePatterns.isEmpty) false
-    else
+    // CPS-IT/mailqueue Extension::KEY has a null .method; it cannot be near a method source.
+    else Option(c.method).exists { m =>
       sourceMemo.getOrElseUpdate(
-        c.method.fullName,
-        c.method.ast.isCall.exists(x => sourcePatterns.exists(p => boundedPattern(p).matcher(x.code).find()))
+        m.fullName,
+        m.ast.isCall.exists(x => sourcePatterns.exists(p => boundedPattern(p).matcher(x.code).find()))
       )
+    }
 
   def isModelled(c: io.shiftleft.codepropertygraph.generated.nodes.Call): Boolean = {
     val callee = calleeText(c.code)
diff --git a/src/openultrasast/model/taint.py b/src/openultrasast/model/taint.py
index f4214e4..b942f74 100644
--- a/src/openultrasast/model/taint.py
+++ b/src/openultrasast/model/taint.py
@@ -28,6 +28,7 @@ def request_params(
     call_depth: int = 0,
     hook_callbacks: str = "",
     field_parameter_sources: bool | None = None,
+    trace: bool = False,
 ) -> dict[str, object]:
     """The query parameters this arbiter sends. Shared with the batcher so there is one definition, not two.
 
@@ -40,6 +41,7 @@ def request_params(
     means unscoped, which is what the single-region pair path still uses.
     """
     return {
+        **({"trace": "true"} if trace else {}),
         "sources": spec.sources,
         "sinks": spec.sinks,
         "sanitizers": spec.sanitizers,
diff --git a/tests/test_cpg_queries_compile.py b/tests/test_cpg_queries_compile.py
index 7aa9120..b6d159c 100644
--- a/tests/test_cpg_queries_compile.py
+++ b/tests/test_cpg_queries_compile.py
@@ -20,6 +20,7 @@ host's pytest mounted in, which needs no network and installs nothing:
 from __future__ import annotations
 
 import json
+import re
 import subprocess
 from pathlib import Path
 
@@ -28,6 +29,18 @@ import pytest
 QUERIES = Path("src/openultrasast/cpg/queries")
 
 
+@pytest.mark.parametrize("name", ["taint", "dominance", "config", "census", "overlay", "vocabulary"])
+def test_query_scope_checks_guard_nullable_methods(name: str) -> None:
+    """PHP class constants can have no method; one unsafe scope check aborts the batch."""
+    script = QUERIES / f"{name}.sc"
+    source = script.read_text()
+    assert source, f"{script} is empty"
+    # Require Option(node.method).exists(predicate), keeping the null guard at the call site.
+    unsafe = re.compile(r"\b(?:inScope|nestedInLabeled)\s*\(\s*[\w.]+\s*\.\s*method\s*\)")
+    sites = [source.count("\n", 0, match.start()) + 1 for match in unsafe.finditer(source)]
+    assert not sites, f"{script}: unguarded method scope checks at lines {sites}"
+
+
 def _joern() -> str | None:
     import shutil
 
diff --git a/tests/test_engine_trace.py b/tests/test_engine_trace.py
new file mode 100644
index 0000000..d09a848
--- /dev/null
+++ b/tests/test_engine_trace.py
@@ -0,0 +1,1163 @@
+"""Offline controls for the trace experiment; no Docker, Joern or network."""
+
+from __future__ import annotations
+
+import importlib.util
+import json
+import subprocess
+import sys
+from dataclasses import replace
+from pathlib import Path
+
+import pytest
+
+from openultrasast.cpg.backend import BEGIN, END, CpgResult, JoernBackend
+from openultrasast.model.specs import taint_specs
+from openultrasast.model.taint import request_params
+
+SCRIPT = Path(__file__).resolve().parents[1] / "benchmarks/learn/engine_trace.py"
+_spec = importlib.util.spec_from_file_location("engine_trace", SCRIPT)
+assert _spec and _spec.loader
+trace = importlib.util.module_from_spec(_spec)
+with pytest.MonkeyPatch.context() as patch:
+    patch.syspath_prepend(str(SCRIPT.parent))
+    from engine_trace_parse import parse_trace
+
+    _spec.loader.exec_module(trace)
+worker = sys.modules["engine_trace_worker"]
+
+
+def row(**extra):
+    return {
+        "sink": "os.system(cmd)",
+        "source": "request.args['cmd']",
+        "sinkFile": "app.py",
+        "sinkMethod": "run",
+        "sinkLine": 4,
+        "sourceKind": "framework",
+        "length": 3,
+        "paths": 2,
+        "sanitized": False,
+        "bounded": False,
+        "bound": "",
+        "trace": ["app.py:run:2 [CALL] request.args['cmd']", "app.py:run:3 [IDENTIFIER] cmd", "app.py:run:4 [CALL] os.system(cmd)"],
+        **extra,
+    }
+
+
+def unit(id="a", pin="pin", **extra):
+    return {
+        "unit": id,
+        "repo": "owner/repo",
+        "pin": pin,
+        "family": "injection",
+        "source": "vibe-py",
+        "excerpt": True,
+        "file": "app.py",
+        "function": "run",
+        "language": "python",
+        "pin_role": "vulnerable",
+        "pair": "pair",
+        "label": 1,
+        "fold": "outer-0",
+        "group": "owner/repo",
+        "supported": True,
+        "reason": "",
+        **extra,
+    }
+
+
+def test_trace_roles_and_path_level_sanitizer():
+    result = parse_trace(row(sanitized=True, bound="cmd", bounded=True))
+    assert [s["roles"] for s in result["steps"]] == [["source"], ["step", "guard"], ["sink"]]
+    assert result["steps"][0] == {
+        "file": "app.py",
+        "method": "run",
+        "line": 2,
+        "node_label": "CALL",
+        "code": "request.args['cmd']",
+        "roles": ["source"],
+        "raw": "app.py:run:2 [CALL] request.args['cmd']",
+    }
+    assert result["sanitizer"] == {"on_path": True, "step": None}
+    assert result["guard"] == {"text": "cmd", "bounded": True}
+    off_path = parse_trace(row(bound="len < 20", bounded=True))
+    assert all("guard" not in s["roles"] for s in off_path["steps"])
+    assert off_path["guard"]["text"] == "len < 20"
+
+
+def test_trace_unknown_location_and_method_colons():
+    parsed = parse_trace(row(trace=["?:-1 [UNKNOWN] x", "a.php:A::run:4 [CALL] sink(x)"]))
+    assert parsed["steps"][0]["line"] == -1
+    assert parsed["steps"][0]["method"] == "?"
+    assert parsed["steps"][1]["method"] == "A::run"
+    assert parse_trace(row(trace=["a.py:run:2 [CALL] x"]))["steps"][0]["roles"] == ["source", "sink"]
+    with pytest.raises(ValueError, match="invalid trace"):
+        parse_trace(row(trace=["unreadable"]))
+
+
+def test_rendered_default_scan_params_do_not_request_trace(tmp_path):
+    spec = taint_specs(language="python")["injection"]
+    rendered = []
+
+    def runner(command, **kwargs):
+        request_file = next(arg.split("=", 1)[1] for arg in command if arg.startswith("requestsFile="))
+        rendered.append(json.loads(Path(request_file).read_text()))
+        return subprocess.CompletedProcess(command, 0, BEGIN + '\n{"r": []}\n' + END, "")
+
+    backend = JoernBackend(runner=runner, session_transport=False)
+    cpg = tmp_path / "cpg.bin"
+    cpg.write_bytes(b"graph")
+    backend._batch_once(cpg, "taint", {"r": request_params(spec)})
+    backend._batch_once(cpg, "taint", {"r": request_params(spec, trace=True)})
+    assert "trace" not in rendered[0]["r"]
+    assert rendered[1]["r"]["trace"] == "true"
+    script = backend.queries_dir / "taint.sc"
+    assert "traceS = trace" in script.read_text()  # single-query forwarding too
+
+
+def test_plan_groups_skips_done_and_retains_other_families(tmp_path):
+    units = [unit(), unit("b", label=0), unit("c", "other")]
+    plan = trace.make_plan(units, tmp_path)
+    assert sorted(len(p["units"]) for p in plan) == [1, 2]
+    done = next(p for p in plan if p["pin"] == "pin")
+    trace.write_json(tmp_path / trace.pin_name(done), {**done, "done": True})
+    assert len(trace.make_plan(units, tmp_path)) == 1
+    assert trace.make_plan(units, tmp_path, only_family="path") == []
+    assert len(trace.make_plan(units, tmp_path, limit=1)) == 1
+    # An interrupted pin is rerun, even if a checkpoint file exists.
+    trace.write_json(tmp_path / trace.pin_name(done), {**done, "done": False})
+    assert len(trace.make_plan(units, tmp_path)) == 2
+
+
+def test_join_requires_exact_examples_and_ts_opt_in():
+    u = unit()
+    example = {**u, "id": "a", "candidate": "app.ts::run", "source": "pairs", "language": "typescript"}
+    with pytest.raises(ValueError, match="missing 1/1"):
+        trace.join_units([u], [])
+    assert not trace.join_units([u], [example])[0]["supported"]
+    assert trace.join_units([u], [example], include_typescript=True)[0]["supported"]
+
+
+class FakeBackend:
+    seconds = 20
+    census_files = 1
+    timed_out = False
+    fatal = ""
+    last_failure = ""
+    seen = []
+
+    def __init__(self, deadline, question_deadline, checkpoint):
+        self.stages = [{"command": "joern-parse", "seconds": self.seconds, "exit": 0, "cpg_bytes": 2048}]
+        self.asked = []
+        self.answers = {}
+
+    def build(self, root, language):
+        graph = root / "graph.bin"
+        graph.write_bytes(b"x" * 2048)
+
+        def batch(query, requests):
+            self.seen.append(requests)
+            self.asked.extend(requests)
+            self.stages.append({"command": "joern", "seconds": self.seconds, "exit": 0})
+            return {
+                **{rid: [row()] if rid == "a" else [] for rid in requests},
+                "__census__": [{"files": str(self.census_files), "file_names": ["app.py"]}],
+            }
+
+        return CpgResult(graph, lambda q, p: None, run_batch_once=batch)
+
+
+def run_fake(tmp_path, monkeypatch, backend=FakeBackend, content="def run():\n    pass\n"):
+    monkeypatch.setattr(worker, "MeasuredBackend", backend)
+    (tmp_path / "app.py").write_text(content)
+    pin = {"repo": "owner/repo", "pin": "pin", "units": [unit(), unit("b")]}
+    out = tmp_path / "result.json"
+    pin["questions"] = trace.prepare_questions(pin, tmp_path, 120) if content else {}
+    worker.worker(pin, tmp_path, out, 900, 120)
+    return json.loads(out.read_text())
+
+
+def test_worker_one_batch_and_preserves_witnesses(tmp_path, monkeypatch):
+    FakeBackend.seen = []
+    result = run_fake(tmp_path, monkeypatch)
+    assert [u["status"] for u in result["units"]] == ["path", "asked-nothing"]
+    assert len(FakeBackend.seen) == 1
+    assert all(p["trace"] == "true" and p["function"] == "run" and p["file"] == "app.py" for p in FakeBackend.seen[0].values())
+    assert result["units"][0]["witness_rows"] == [row()]
+    assert result["units"][0]["traces"][0]["steps"][-1]["roles"] == ["sink"]
+    assert result["units"][1]["questions_completed"] == ["b"]
+    assert result["instrument"]["files"][0]["bytes"] > 0
+
+
+@pytest.mark.parametrize("json_roundtrip", [False, True])
+def test_worker_writes_string_requests_for_php_and_python(tmp_path, monkeypatch, json_roundtrip):
+    written = {}
+    submitted = {}
+
+    class RequestBackend(worker.MeasuredBackend):
+        def build(self, root, language):
+            graph = root / f"{language}.bin"
+            graph.write_bytes(b"x" * 2048)
+            self.stages.append({"command": "joern-parse", "seconds": 20, "exit": 0, "cpg_bytes": 2048})
+
+            def batch(query, requests):
+                submitted.update(requests)
+                return self._batch_once(graph, query, requests)
+
+            return CpgResult(graph, lambda q, p: None, run_batch_once=batch)
+
+        def _run(self, command, **kwargs):
+            request_file = Path(next(arg.split("=", 1)[1] for arg in command if arg.startswith("requestsFile=")))
+            data = request_file.read_bytes()
+            print(f"read {request_file.name}: {len(data)} bytes")
+            requests = json.loads(data)
+            written.update(requests)
+            self.asked.extend(requests)
+            self.stages.append({"command": "joern", "seconds": 20, "exit": 0})
+            answers = {**dict.fromkeys(requests, []), "__census__": [{"files": "1", "file_names": ["app.php", "app.py"]}]}
+            return subprocess.CompletedProcess(command, 0, BEGIN + "\n" + json.dumps(answers) + "\n" + END, "")
+
+    monkeypatch.setattr(worker, "MeasuredBackend", RequestBackend)
+    (tmp_path / "app.php").write_text("<?php function run() {}\n")
+    (tmp_path / "app.py").write_text("def run(): pass\n")
+    pin = {"units": [unit("php", language="php", file="app.php"), unit("python")]}
+    pin["questions"] = trace.prepare_questions(pin, tmp_path, 120)
+    assert all(isinstance(params["sources"], tuple) and params["sources"] for params in pin["questions"].values())
+    expected = {rid: ",".join(sorted(params["sources"])) for rid, params in pin["questions"].items()}
+    # Cover both direct tuple callers and arrays decoded from the host's pin.json.
+    if json_roundtrip:
+        pin = json.loads(json.dumps(pin))
+    out = tmp_path / "result.json"
+    worker.worker(pin, tmp_path, out, 900, 120)
+    result = json.loads(out.read_text())
+    assert all(row["status"] == "asked-nothing" for row in result["units"])
+    assert set(written) == set(submitted) == {"php", "python"}
+    for requests in (written, submitted, {rid: p for row in result["units"] for rid, p in row["questions"].items()}):
+        assert all(isinstance(value, str) for params in requests.values() for value in params.values())
+        assert {rid: params["sources"] for rid, params in requests.items()} == expected
+
+
+def test_failed_query_records_non_stack_stderr_tail_in_unit_reason(tmp_path, monkeypatch):
+    error = 'ujson.Value$InvalidData: Expected ujson.Str (data: ["$_COOKIE","$_GET"])'
+    joern = tmp_path / "joern"
+    joern.write_text(
+        f"#!{sys.executable}\nimport sys\n"
+        "print('read bytes:', len(open(sys.argv[-1], 'rb').read()), flush=True)\n"
+        "for i in range(25): print(f'diagnostic {i}', file=sys.stderr)\n"
+        f"print({error!r}, file=sys.stderr)\n"
+        "for i in range(30): print(f'\\tat example.Frame.method(Frame.java:{i})', file=sys.stderr)\n"
+        "print('\\t... 30 more', file=sys.stderr)\n"
+        "sys.exit(1)\n"
+    )
+    joern.chmod(0o755)
+
+    class FailedQueryBackend(worker.MeasuredBackend):
+        def build(self, root, language):
+            graph = root / "graph.bin"
+            graph.write_bytes(b"x" * 2048)
+            self.stages.append({"command": "joern-parse", "seconds": 20, "exit": 0, "cpg_bytes": 2048})
+
+            def batch(query, requests):
+                done = self._run([str(joern), "--script", str(root / "app.py")], timeout=10, cwd=root)
+                assert done.returncode == 1
+                assert "read bytes: 20" in done.stdout
+                return None
+
+            return CpgResult(graph, lambda q, p: None, run_batch_once=batch)
+
+    result = run_fake(tmp_path, monkeypatch, FailedQueryBackend)
+    stage = result["instrument"]["jvm"][-1]
+    assert stage["exit"] == 1
+    assert stage["output_tail"].splitlines() == [*[f"diagnostic {i}" for i in range(6, 25)], error]
+    for row in result["units"]:
+        assert row["status"] == "failed"
+        assert "joern query exited 1" in row["reason"]
+        assert error in row["reason"]
+        assert "Frame.java" not in row["reason"]
+        assert "implausibly fast" not in row["reason"]
+
+
+@pytest.mark.parametrize("failure", ["fast", "zero-input", "zero-census"])
+def test_invalid_instrument_is_persisted_failed(tmp_path, monkeypatch, failure):
+    class BadBackend(FakeBackend):
+        seconds = 0.5 if failure == "fast" else 20
+        census_files = 0 if failure == "zero-census" else 1
+
+    result = run_fake(tmp_path, monkeypatch, BadBackend, "" if failure == "zero-input" else "def run():\n    pass\n")
+    assert result["done"]
+    assert all(u["status"] == "failed" for u in result["units"])
+    assert all(u["reason"] for u in result["units"])
+    assert "<5s" in result["units"][0]["reason"] if failure == "fast" else "zero" in result["units"][0]["reason"]
+
+
+def test_missing_labelled_function_is_failure(tmp_path, monkeypatch):
+    with pytest.raises(ValueError, match="not declared"):
+        run_fake(tmp_path, monkeypatch, content="def other():\n    pass\n")
+
+
+def test_docker_command_is_serial_offline_frozen():
+    cmd = trace.docker_command(Path("/export"), Path("/pin"), Path("/output"), "one", "openultrasast:dev", 900, 60)
+    assert cmd[cmd.index("--memory") + 1] == "3g"
+    assert cmd[cmd.index("--network") + 1] == "none"
+    assert cmd[cmd.index("--pull") + 1] == "never"
+    assert "/export:/frozen:ro" in cmd and "/pin:/case:ro" in cmd
+    assert cmd[-1] == "60"
+    assert "/frozen/benchmarks/learn/engine_trace_worker.py" in cmd
+    assert "benchmarks/learn/engine_trace_worker.py" in trace.SNAPSHOT_FILES
+    assert "benchmarks/learn/engine_trace_parse.py" in trace.SNAPSHOT_FILES
+
+
+def test_summary_separates_unanswered_pairs(tmp_path):
+    rows = [
+        trace.unit_record(unit(), "path"),
+        trace.unit_record(unit("b", label=0), "asked-nothing"),
+        trace.unit_record(unit("c", pair="other"), "failed"),
+        trace.unit_record(unit("d", pair="other", label=0), "timeout"),
+    ]
+    trace.write_json(tmp_path / "one.json", {"done": True, "units": rows})
+    result = trace.summary(tmp_path)
+    assert result["pair_asymmetry"] == {"vulnerable-only": 1, "neither": 1}
+    assert result["pair_asymmetry_both_answered"] == {"vulnerable-only": 1}
+    assert result["by_family"]["injection"]["failed"] == 1
+
+
+def test_frozen_source_overlays_uncommitted_files_without_live_mount(tmp_path, monkeypatch):
+    root = tmp_path / "repo"
+    root.mkdir()
+    for name in trace.SNAPSHOT_FILES:
+        dest = root / name
+        dest.parent.mkdir(parents=True, exist_ok=True)
+        dest.write_text("snapshot " + name)
+    monkeypatch.setattr(trace, "ROOT", root)
+    commands = []
+
+    def run(cmd, **kwargs):
+        commands.append(cmd)
+        return subprocess.CompletedProcess(cmd, 0, "head-sha\n", "")
+
+    monkeypatch.setattr(trace.subprocess, "run", run)
+    frozen = tmp_path / "frozen"
+    proof = trace.freeze_source(frozen)
+    assert commands[0][-3:] == ["archive", "HEAD", "src"]
+    for name in trace.SNAPSHOT_FILES:
+        assert (frozen / name).read_bytes() == (root / name).read_bytes()
+        assert len(proof["overlay_sha256"][name]) == 64
+
+
+def test_question_watchdog_retains_completed_rows_without_joern(tmp_path):
+    import sys
+
+    code = (
+        "import json,time\n"
+        'print(json.dumps({"__question__":"a"}),flush=True)\n'
+        'print(json.dumps({"id":"a","rows":[]}),flush=True)\n'
+        'print(json.dumps({"__question__":"b"}),flush=True)\n'
+        "time.sleep(60)\n"
+    )
+    backend = worker.MeasuredBackend(10, 0.05, lambda: None)
+    done = backend._run([sys.executable, "-c", code], timeout=10, cwd=tmp_path)
+    assert backend.timed_out
+    assert backend.asked == ["a", "b"]
+    assert backend.answers == {"a": []}
+    assert done.returncode != 0
+    assert backend.stages == []
+    assert not backend.fatal
+
+
+def test_pair_mapping_handles_harvest_normalized_crlf(tmp_path):
+    import hashlib
+    from types import SimpleNamespace
+
+    file = tmp_path / "excerpt.py"
+    file.write_bytes(b"def run():\r\n    pass\r\n")
+    normalized = file.read_text().encode()
+    pin = hashlib.sha1(b"blob " + str(len(normalized)).encode() + b"\0" + normalized).hexdigest()
+    case = SimpleNamespace(name="case", repo="owner/repo", relpath="app.py", vuln_file=file, fixed_file=file)
+    inputs = object.__new__(trace.Inputs)
+    inputs.pairs = [case]
+    inputs.hashes = {}
+    assert inputs.pair_for(unit(pin=pin)) == (case, "vuln")
+
+
+def test_missing_batch_answers_are_timeout_not_empty(tmp_path, monkeypatch):
+    class TimeoutBackend(FakeBackend):
+        def build(self, root, language):
+            graph = root / "graph.bin"
+            graph.write_bytes(b"x" * 2048)
+
+            def batch(query, requests):
+                self.timed_out = True
+                self.asked.append("a")
+                return None
+
+            return CpgResult(graph, lambda q, p: None, run_batch_once=batch)
+
+    result = run_fake(tmp_path, monkeypatch, TimeoutBackend)
+    assert [r["status"] for r in result["units"]] == ["timeout", "timeout"]
+    assert result["units"][0]["questions_asked"] == ["a"]
+    assert result["units"][1]["questions_asked"] == []
+
+
+def test_dry_run_makes_no_subprocess_calls(tmp_path, monkeypatch, capsys):
+    import sys
+
+    u = unit()
+    examples = [{**u, "id": "a", "candidate": "app.py::run", "source": "population-v1", "excerpt": False}]
+    units = tmp_path / "units.jsonl"
+    units.write_text(json.dumps(u) + "\n")
+    example_file = tmp_path / "examples.jsonl"
+    example_file.write_text(json.dumps(examples[0]) + "\n")
+    monkeypatch.setattr(
+        sys, "argv", [str(SCRIPT), "--dry-run", "--units", str(units), "--examples", str(example_file), "--out", str(tmp_path / "results")]
+    )
+    monkeypatch.setattr(trace.subprocess, "run", lambda *a, **kw: pytest.fail("dry-run spawned a subprocess"))
+    assert trace.main() == 0
+    plan = json.loads(capsys.readouterr().out)
+    assert plan["runnable_pins"] == 1 and plan["units"] == 1
+    assert plan["estimated_seconds"] == 182
+    assert not (tmp_path / "results").exists()
+
+
+def test_family_filtered_results_merge_without_losing_prior_units(tmp_path):
+    first = trace.make_plan([unit(), unit("b", family="path")], tmp_path, only_family="injection")[0]
+    trace.save_pin(tmp_path / trace.pin_name(first), {**first, "done": True})
+    second = trace.make_plan([unit(), unit("b", family="path")], tmp_path, only_family="path")[0]
+    trace.save_pin(tmp_path / trace.pin_name(second), {**second, "done": True})
+    assert trace.make_plan([unit(), unit("b", family="path")], tmp_path) == []
+
+
+def test_failed_launch_persists_failed_pin_and_resume_replans_it(tmp_path, monkeypatch):
+    from types import SimpleNamespace
+
+    class Inputs:
+        def materialize(self, pin, target):
+            target.mkdir()
+            (target / "app.py").write_text("def run():\n    pass\n")
+            return {"mode": "source"}
+
+    commands = []
+
+    def run(cmd, **kwargs):
+        commands.append(cmd)
+        return subprocess.CompletedProcess(cmd, 0, "", "")
+
+    monkeypatch.setattr(trace.subprocess, "run", run)
+    pin = trace.make_plan([unit()], tmp_path)[0]
+    args = SimpleNamespace(image="openultrasast:dev", deadline=900, question_deadline=120)
+    assert trace.run_pin(pin, tmp_path / "frozen", {}, Inputs(), tmp_path, args)
+    result = json.loads((tmp_path / trace.pin_name(pin)).read_text())
+    assert result["units"][0]["status"] == "failed"
+    assert "produced no result" in result["units"][0]["reason"]
+    assert [c[1] for c in commands if c[0] == "docker"] == ["run", "rm"]
+    assert len(trace.make_plan([unit()], tmp_path)) == 1
+    assert trace.make_plan([unit()], tmp_path, rerun_status=()) == []
+
+
+def import_container_worker():
+    """Import in a fresh interpreter so host imports cannot mask the boundary."""
+    script = SCRIPT.with_name("engine_trace_worker.py")
+    code = """
+import json, pathlib, runpy, sys
+sys.modules['yaml'] = None
+sys.modules['boto3'] = None
+script = pathlib.Path(sys.argv[1])
+size = len(script.read_bytes())
+# Direct script execution in the image puts the sibling modules on sys.path.
+sys.path.insert(0, str(script.parent))
+runpy.run_path(str(script), run_name='trace_import_check')
+print(json.dumps({'bytes': size, 'modules': sorted(
+    name for name in sys.modules if name == 'openultrasast' or name.startswith('openultrasast.')
+)}))
+"""
+    result = subprocess.run([sys.executable, "-c", code, str(script)], capture_output=True, text=True, timeout=30)
+    assert result.returncode == 0, result.stderr
+    return json.loads(result.stdout)
+
+
+def test_trace_container_import_without_yaml_or_boto3():
+    assert import_container_worker()["bytes"] > 0
+
+
+def test_trace_container_transitive_import_boundary():
+    # Explicit inventory: additions need review, particularly learn.examples/plane.*.
+    assert import_container_worker()["modules"] == [
+        "openultrasast",
+        "openultrasast.contracts",
+        "openultrasast.cpg",
+        "openultrasast.cpg.artifact",
+        "openultrasast.cpg.backend",
+        "openultrasast.cpg.session",
+        "openultrasast.model",
+        "openultrasast.model.contracts",
+    ]
+
+
+@pytest.mark.parametrize("existing", [False, True])
+def test_trace_dry_run_unsupported_units_write_nothing(tmp_path, monkeypatch, capsys, existing):
+    out = tmp_path / "results"
+    if existing:
+        out.mkdir()
+        (out / ".lock").write_bytes(b"existing lock")
+        (out / "checkpoint.json").write_text('{"done": false}')
+        (out / "previous.json").write_text('{"done": true, "units": []}')
+    before = {p.name: (p.read_bytes(), p.stat().st_mtime_ns) for p in out.glob("*")}
+    units = [unit("ts", file="app.ts"), unit("py", pin="python")]
+    examples = [{**u, "id": u["unit"], "candidate": u["file"] + "::run", "source": "population-v1", "excerpt": False} for u in units]
+    units_file, examples_file = tmp_path / "units.jsonl", tmp_path / "examples.jsonl"
+    units_file.write_text("\n".join(map(json.dumps, units)))
+    examples_file.write_text("\n".join(map(json.dumps, examples)))
+    monkeypatch.setattr(
+        sys, "argv", [str(SCRIPT), "--dry-run", "--units", str(units_file), "--examples", str(examples_file), "--out", str(out)]
+    )
+    monkeypatch.setattr(trace.subprocess, "run", lambda *a, **kw: pytest.fail("dry run spawned subprocess"))
+    monkeypatch.setattr(trace, "write_json", lambda *a: pytest.fail("dry run wrote JSON"))
+    assert trace.main() == 0
+    plan = json.loads(capsys.readouterr().out)
+    assert plan["pins"] == 2 and plan["runnable_pins"] == 1
+    assert out.exists() == existing
+    assert {p.name: (p.read_bytes(), p.stat().st_mtime_ns) for p in out.glob("*")} == before
+
+
+def test_trace_real_unsupported_run_then_typescript_opt_in_replans(tmp_path, monkeypatch, capsys):
+    u = unit(file="app.ts")
+    example = {**u, "id": "a", "candidate": "app.ts::run", "source": "population-v1", "excerpt": False}
+    units_file, examples_file, out = tmp_path / "units.jsonl", tmp_path / "examples.jsonl", tmp_path / "results"
+    units_file.write_text(json.dumps(u) + "\n")
+    examples_file.write_text(json.dumps(example) + "\n")
+    argv = [str(SCRIPT), "--units", str(units_file), "--examples", str(examples_file), "--out", str(out)]
+    monkeypatch.setattr(sys, "argv", argv)
+    monkeypatch.setattr(trace, "freeze_source", lambda target: {})
+    monkeypatch.setattr(trace.subprocess, "run", lambda *a, **kw: pytest.fail("unsupported run invoked Docker"))
+    assert trace.main() == 0
+    result_file = out / trace.pin_name(u)
+    first = json.loads(result_file.read_text())
+    assert first["done"] and first["units"][0]["status"] == "unsupported"
+    progress = json.loads((out / "progress.json").read_text())
+    assert progress == {"pins_done": 1, "pins_total": 1, "last_pin": {"repo": u["repo"], "pin": u["pin"]}, "statuses": {"unsupported": 1}}
+    assert trace.make_plan(trace.join_units([u], [example]), out) == []
+
+    called = []
+
+    def run_pin(pin, source, provenance, inputs, results, args):
+        called.append(pin)
+        trace.save_pin(
+            results / trace.pin_name(pin), {**pin, "done": True, "units": [trace.unit_record(r, "asked-nothing") for r in pin["units"]]}
+        )
+        return False
+
+    def inspect_image(command, **kwargs):
+        assert command[:3] == ["docker", "image", "inspect"]
+        return subprocess.CompletedProcess(command, 0, "sha256:fake\n", "")
+
+    monkeypatch.setattr(trace, "run_pin", run_pin)
+    monkeypatch.setattr(trace.subprocess, "run", inspect_image)
+    monkeypatch.setattr(sys, "argv", [*argv, "--include-typescript"])
+    assert trace.main() == 0
+    assert len(called) == 1 and called[0]["units"][0]["supported"]
+    assert called[0]["selection"] != first["selection"]
+    assert json.loads(result_file.read_text())["units"][0]["status"] == "asked-nothing"
+    assert trace.make_plan(trace.join_units([u], [example], include_typescript=True), out, include_typescript=True) == []
+
+
+def test_trace_resume_identity_includes_selection_even_if_support_unchanged(tmp_path):
+    first = trace.make_plan([unit()], tmp_path)[0]
+    trace.save_pin(tmp_path / trace.pin_name(first), {**first, "done": True})
+    assert trace.make_plan([unit()], tmp_path) == []
+    assert trace.make_plan([unit()], tmp_path, rerun_status=()) == []
+    assert len(trace.make_plan([unit()], tmp_path, include_typescript=True)) == 1
+
+
+@pytest.mark.parametrize("binary", sorted(worker.FRONTENDS))
+def test_fast_frontend_validates_exact_output(tmp_path, binary):
+    frontend = tmp_path / binary
+    frontend.write_text(
+        f"#!{sys.executable}\n"
+        "import pathlib, sys, time\n"
+        "source = pathlib.Path(sys.argv[1])\n"
+        "print('read bytes:', len(source.read_bytes()), flush=True)\n"
+        "time.sleep(2 if pathlib.Path(sys.argv[0]).name == 'php2cpg' else 0)\n"
+        "pathlib.Path(sys.argv[3]).write_bytes(b'x' * 34000)\n"
+    )
+    frontend.chmod(0o755)
+    (tmp_path / "source.php").write_text("<?php echo 'hello';")
+    backend = worker.MeasuredBackend(30, 10, lambda: None)
+    flag = "--output" if binary == "joern-parse" else "-o"
+    done = backend._run([str(frontend), "source.php", flag, "out.graph"], timeout=10, cwd=tmp_path)
+    assert done.returncode == 0
+    assert "read bytes: 19" in done.stdout
+    assert not backend.fatal
+    stage = backend.stages[0]
+    assert stage["seconds"] < 5
+    if binary == "php2cpg":
+        assert stage["seconds"] >= 2
+    assert stage["cpg_path"] == str(tmp_path / "out.graph")
+    assert stage["cpg_bytes"] == 34000
+    proof = {"files": ["source.php"], "bytes": 19, **stage, "jvm": backend.stages}
+    assert worker.instrument_failure(proof) == ""
+
+
+@pytest.mark.parametrize("size", [None, 0, 1024])
+def test_frontend_without_sufficient_output_fails(tmp_path, size):
+    frontend = tmp_path / "php2cpg"
+    frontend.write_text(
+        f"#!{sys.executable}\nimport pathlib, sys\n"
+        + (f"pathlib.Path(sys.argv[2]).write_bytes(b'x' * {size})\n" if size is not None else "")
+    )
+    frontend.chmod(0o755)
+    # A different non-empty CPG must not mask the missing requested output.
+    (tmp_path / "unrelated.bin").write_bytes(b"x" * 34000)
+    backend = worker.MeasuredBackend(30, 10, lambda: None)
+    done = backend._run([str(frontend), "-o", "out.bin"], timeout=10, cwd=tmp_path)
+    assert done.returncode == 1
+    assert backend.stages[0]["exit"] == 0
+    assert backend.stages[0]["cpg_bytes"] == (size or 0)
+    assert backend.fatal == "frontend wrote no CPG"
+
+
+def test_fast_joern_script_is_failed_launch(tmp_path):
+    joern = tmp_path / "joern"
+    joern.write_text(f"#!{sys.executable}\n")
+    joern.chmod(0o755)
+    backend = worker.MeasuredBackend(30, 10, lambda: None)
+    done = backend._run([str(joern), "--script", "query.sc"], timeout=10, cwd=tmp_path)
+    assert done.returncode == 1
+    assert backend.stages[0]["exit"] == 0
+    assert "implausibly fast JVM step" in backend.fatal
+
+
+@pytest.mark.parametrize("markers", [[], ["a"], ["a", "b"]])
+def test_empty_answers_require_each_question_marker(tmp_path, monkeypatch, markers):
+    class EmptyBackend(FakeBackend):
+        def build(self, root, language):
+            cpg = super().build(root, language)
+            batch = cpg.run_batch_once
+
+            def empty_batch(query, requests):
+                answers = batch(query, requests)
+                self.asked = markers
+                return {**answers, **dict.fromkeys(requests, [])}
+
+            return replace(cpg, run_batch_once=empty_batch)
+
+    result = run_fake(tmp_path, monkeypatch, EmptyBackend)
+    for row in result["units"]:
+        assert row["status"] == ("asked-nothing" if row["unit"] in markers else "failed")
+        assert row["reason"] == ("" if row["unit"] in markers else "query produced no markers")
+
+
+def test_frontend_measurement_survives_failed_build_cleanup(tmp_path, monkeypatch):
+    class CleanupBackend(worker.MeasuredBackend):
+        def build(self, root, language):
+            frontend = root / "php2cpg"
+            frontend.write_text(
+                f"#!{sys.executable}\nimport pathlib, sys\npathlib.Path(sys.argv[2]).write_bytes(b'x' * 34000)\nsys.exit(1)\n"
+            )
+            frontend.chmod(0o755)
+            self._run([str(frontend), "-o", "actual.graph"], timeout=10, cwd=root)
+            (root / "actual.graph").unlink()
+            return None
+
+    result = run_fake(tmp_path, monkeypatch, CleanupBackend)
+    assert all(row["status"] == "failed" for row in result["units"])
+    assert result["instrument"]["cpg_path"] == str(tmp_path / "actual.graph")
+    assert result["instrument"]["cpg_bytes"] == 34000
+    assert not (tmp_path / "actual.graph").exists()
+
+
+def pair_inputs(tmp_path):
+    from types import SimpleNamespace
+
+    inputs = trace.Inputs(trace.ROOT, tmp_path / "cache")
+    excerpt = tmp_path / "excerpt.py"
+    excerpt.write_text("# commit: " + "f" * 40 + "\n# parent: " + "f" * 40 + "\ndef run(): pass\n")
+    case = SimpleNamespace(name="fixture", repo="owner/repo", relpath="pkg/app.py", vuln_file=excerpt, fixed_file=excerpt, context_files=())
+    inputs.catalog_rows[case.name] = [
+        ("advisory-fixes-2", {"repo": case.repo, "relpath": case.relpath, "parent": "a" * 40, "commit": "b" * 40})
+    ]
+    return inputs, case
+
+
+@pytest.mark.parametrize("side,sha", [("vuln", "a" * 40), ("fixed", "b" * 40)])
+def test_pair_materializes_catalog_side_not_header(tmp_path, monkeypatch, side, sha):
+    inputs, case = pair_inputs(tmp_path)
+    case.vuln_file.write_text("# Provenance: example\n    def run(self): pass\n")
+    fetched = tmp_path / "fetched"
+    (fetched / "pkg").mkdir(parents=True)
+    (fetched / case.relpath).write_text("class Handler:\n    def run(self): pass\n")
+    calls = []
+
+    def fetch(repo, commit, roots):
+        calls.append((repo, commit, roots))
+        return fetched
+
+    monkeypatch.setattr(inputs, "fetch_source", fetch)
+    monkeypatch.setattr(inputs, "pair_for", lambda row: (case, side))
+    target = tmp_path / "export"
+    proof = inputs.materialize({"units": [unit(file=case.relpath)]}, target)
+    assert calls == [("owner/repo", sha, ["pkg"])]
+    assert proof["commit"] == sha and proof["mode"] == "source"
+    assert (target / case.relpath).read_text().startswith("class Handler:")
+
+
+@pytest.mark.parametrize("change", [{"parent": ""}, {"parent": "b" * 40}, {"commit": "HEAD"}])
+def test_catalog_ambiguous_commits_refused(tmp_path, change):
+    inputs, case = pair_inputs(tmp_path)
+    inputs.catalog_rows[case.name][0][1].update(change)
+    with pytest.raises(ValueError, match="ambiguous catalog commits"):
+        inputs.source_identity(case, "vuln")
+
+
+def test_ambiguous_catalog_and_side_refused(tmp_path):
+    inputs, case = pair_inputs(tmp_path)
+    inputs.catalog_rows[case.name] *= 2
+    with pytest.raises(ValueError, match="ambiguous catalog source"):
+        inputs.source_identity(case, "fixed")
+    with pytest.raises(ValueError, match="ambiguous pair side"):
+        inputs.pair_for(unit(pin_role="unknown"))
+
+
+def test_sparse_export_filters_languages_dirs_and_caps_context(tmp_path):
+    source, target = tmp_path / "source", tmp_path / "export"
+    contents = {
+        "pkg/app.py": "def run(): pass\n",
+        "pkg/context.py": "x" * 200,
+        "pkg/other.js": "function run() {}",
+        "other/app.py": "other",
+    }
+    contents.update({f"pkg/{directory}/a.py": "excluded" for directory in worker.EXCLUDED_DIRS})
+    for rel, data in contents.items():
+        path = source / rel
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(data)
+    proof = trace.export_sources(source, target, [unit(file="pkg/app.py")], ["pkg"], cap=50)
+    assert [p.relative_to(target).as_posix() for p in target.rglob("*") if p.is_file()] == ["pkg/app.py"]
+    assert proof["cap_hit"] and proof["bytes"] == len(contents["pkg/app.py"])
+    assert proof["excludes"] == sorted(worker.EXCLUDED_DIRS)
+    assert trace.source_root("module/src/main/java/org/Handler.java", "java") == "module/src/main/java"
+    assert trace.source_root("pkg/sub/app.ts", "typescript") == "pkg"
+    assert trace.source_root("app.js", "javascript") == "."
+
+
+def test_wrong_fetched_function_fails_without_fallback(tmp_path, monkeypatch):
+    inputs, case = pair_inputs(tmp_path)
+    source = tmp_path / "source"
+    (source / "pkg").mkdir(parents=True)
+    (source / case.relpath).write_text("def runner(): pass\n")
+    inputs.allow_excerpt_fallback = True
+    monkeypatch.setattr(inputs, "pair_for", lambda row: (case, "vuln"))
+    monkeypatch.setattr(inputs, "fetch_source", lambda *args: source)
+    with pytest.raises(ValueError, match="labelled function not found in fetched source"):
+        inputs.materialize({"units": [unit(file=case.relpath)]}, tmp_path / "export")
+
+
+def test_excerpt_fallback_requires_explicit_flag(tmp_path, monkeypatch):
+    inputs, case = pair_inputs(tmp_path)
+    monkeypatch.setattr(inputs, "pair_for", lambda row: (case, "vuln"))
+
+    def unavailable(*args):
+        raise ValueError("fetch failed")
+
+    monkeypatch.setattr(inputs, "fetch_source", unavailable)
+    pin = {"units": [unit(file=case.relpath)]}
+    with pytest.raises(ValueError, match="fetch failed"):
+        inputs.materialize(pin, tmp_path / "export")
+    inputs.allow_excerpt_fallback = True
+    proof = inputs.materialize(pin, tmp_path / "fallback")
+    assert proof["mode"] == "excerpt-fallback" and proof["reason"] == "fetch failed"
+
+
+def test_fetch_renders_sparse_pinned_host_commands(tmp_path, monkeypatch):
+    inputs, _ = pair_inputs(tmp_path)
+    commands = []
+
+    def git(cmd, **kwargs):
+        commands.append(cmd[3:])
+        assert kwargs["env"]["GIT_TERMINAL_PROMPT"] == "0"
+        assert "GIT_NO_LAZY_FETCH" not in kwargs["env"]
+        return subprocess.CompletedProcess(cmd, 0, "a" * 40 + "\n", "")
+
+    monkeypatch.setattr(trace.subprocess, "run", git)
+    clone = inputs.fetch_source("owner/repo", "a" * 40, ["module/src/main/java"])
+    assert clone.is_relative_to(inputs.cache)
+    assert ["fetch", "--depth", "1", "--filter=blob:none", "https://github.com/owner/repo.git", "a" * 40] in commands
+    assert ["sparse-checkout", "set", "--", "module/src/main/java"] in commands
+    assert ["checkout", "--detach", "--force", "a" * 40] in commands
+
+
+def test_explicit_heap_in_frontend_and_query_commands(tmp_path, monkeypatch):
+    backend = worker.MeasuredBackend(900, 120, lambda: None)
+    commands = []
+
+    def run(command, **kwargs):
+        commands.append(command)
+        return subprocess.CompletedProcess(command, 0, BEGIN + '\n{"r": []}\n' + END, "")
+
+    monkeypatch.setattr(backend, "_run", run)
+    monkeypatch.setattr("openultrasast.cpg.backend.shutil.which", lambda name: "/opt/joern-cli/" + name)
+    graph = tmp_path / "cpg.bin"
+    graph.write_bytes(b"x" * 2048)
+    backend._build_with_frontend(tmp_path, graph, tmp_path, "python")
+    backend._batch_once(graph, "taint", {"r": {}})
+    assert {Path(cmd[0]).name for cmd in commands} == {"pysrc2cpg", "joern"}
+    assert all("-J-Xmx2560m" in cmd for cmd in commands)
+    assert "-Xmx2560m" in backend._jvm_env()["JAVA_TOOL_OPTIONS"]
+
+
+@pytest.mark.parametrize("binary", ["pysrc2cpg", "joern"])
+def test_oom_output_overrides_missing_graph_or_markers(tmp_path, binary):
+    launcher = tmp_path / binary
+    launcher.write_text(f"#!{sys.executable}\nprint('java.lang.OutOfMemoryError: Java heap space')\n")
+    launcher.chmod(0o755)
+    backend = worker.MeasuredBackend(30, 10, lambda: None)
+    done = backend._run([str(launcher), "--script", "query.sc"], timeout=10)
+    assert done.returncode == 1
+    assert backend.fatal == "JVM out of memory"
+    assert worker.instrument_failure({"jvm": backend.stages}) == "JVM out of memory"
+
+
+def test_two_language_pin_builds_sequential_frontends_and_routes_questions(tmp_path, monkeypatch):
+    events = []
+
+    class LanguagesBackend(FakeBackend):
+        def build(self, root, language):
+            events.append(("build", language))
+            graph = root / (language + ".bin")
+            graph.write_bytes(b"x" * 2048)
+            self.stages = [
+                {"command": {"python": "pysrc2cpg", "javascript": "jssrc2cpg"}[language], "seconds": 20, "exit": 0, "cpg_bytes": 2048}
+            ]
+
+            def batch(query, requests):
+                events.append(("query", language, sorted(requests)))
+                self.asked.extend(requests)
+                return {
+                    **dict.fromkeys(requests, []),
+                    "__census__": [{"files": "1", "file_names": ["app.py" if language == "python" else "app.js"]}],
+                }
+
+            return CpgResult(graph, lambda q, p: None, run_batch_once=batch, cleanup=lambda: events.append(("cleanup", language)))
+
+    monkeypatch.setattr(worker, "MeasuredBackend", LanguagesBackend)
+    (tmp_path / "app.py").write_text("def run(): pass\n")
+    (tmp_path / "app.js").write_text("function run() {}\n")
+    pin = {"units": [unit(), unit("js", language="javascript", file="app.js")]}
+    pin["questions"] = trace.prepare_questions(pin, tmp_path, 120)
+    assert all(params["callDepth"] == "3" for params in pin["questions"].values())
+    result = worker.worker(pin, tmp_path, tmp_path / "result.json", 900, 120)
+    assert events == [
+        ("build", "javascript"),
+        ("query", "javascript", ["js"]),
+        ("cleanup", "javascript"),
+        ("build", "python"),
+        ("query", "python", ["a"]),
+        ("cleanup", "python"),
+    ]
+    assert all(u["status"] == "asked-nothing" for u in result["units"])
+    assert set(result["instruments"]) == {"python", "javascript"}
+    assert all(u["instrument"]["heap_mb"] == 2560 for u in result["units"])
+
+
+def test_resume_materialization_version_and_status_selection(tmp_path):
+    units = [unit(source="advisory-fixes-2"), unit("other", pin="other", source="population-v1")]
+    pin = trace.make_plan(units, tmp_path, only_source={"advisory-fixes-2"})[0]
+    record = {**pin, "done": True, "units": [trace.unit_record(pin["units"][0], "asked-nothing")]}
+    path = tmp_path / trace.pin_name(pin)
+    trace.write_json(path, record)
+    assert trace.make_plan(units, tmp_path, only_source={"advisory-fixes-2"}) == []
+    assert len(trace.make_plan(units, tmp_path, only_source={"advisory-fixes-2"}, rerun_status={"failed", "asked-nothing"})) == 1
+    trace.write_json(path, {**record, "materialization_version": trace.MATERIALIZATION_VERSION - 1})
+    assert len(trace.make_plan(units, tmp_path, only_source={"advisory-fixes-2"})) == 1
+
+
+def test_sparse_fetch_from_local_git_without_network(tmp_path, monkeypatch):
+    origin = tmp_path / "origin"
+    origin.mkdir()
+    real_run = subprocess.run
+
+    def git(*args):
+        return real_run(["git", "-C", str(origin), *args], capture_output=True, text=True, check=True).stdout.strip()
+
+    git("init")
+    git("config", "user.name", "Offline Test")
+    git("config", "user.email", "offline@example.invalid")
+    for rel in ("module/src/main/java/Handler.java", "module/src/main/java/docs/Other.java", "elsewhere/Other.java"):
+        path = origin / rel
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text("class Handler { void run() {} }\n")
+    git("add", ".")
+    git("commit", "-m", "local fixture")
+    sha = git("rev-parse", "HEAD")
+    inputs, _ = pair_inputs(tmp_path)
+
+    def local_only(cmd, **kwargs):
+        # Substitute a local repository for the one URL; no network-capable command runs.
+        cmd = [str(origin) if arg == "https://github.com/owner/repo.git" else arg for arg in cmd]
+        return real_run(cmd, **kwargs)
+
+    monkeypatch.setattr(trace.subprocess, "run", local_only)
+    clone = inputs.fetch_source("owner/repo", sha, ["module/src/main/java"])
+    assert not (clone / "elsewhere/Other.java").exists()
+    target = tmp_path / "export"
+    labelled = unit(file="module/src/main/java/Handler.java", language="java")
+    proof = trace.export_sources(clone, target, [labelled], ["module/src/main/java"])
+    trace.verify_functions(target, [labelled])
+    assert proof["files"] == 1 and proof["bytes"] > 0
+
+
+@pytest.mark.parametrize("side", ["vuln", "fixed"])
+def test_complete_catalog_twins_and_context_need_no_commits(tmp_path, monkeypatch, side):
+    inputs, case = pair_inputs(tmp_path)
+    inputs.catalog_rows[case.name][0][1].update(parent="", commit="")
+    fixed = tmp_path / "fixed.py"
+    fixed.write_text("def run(): return 'fixed'\n")
+    case.fixed_file = fixed
+    context = tmp_path / "context.py"
+    context.write_text("registered = True\n")
+    case.context_files = (("pkg/context.py", context, context),)
+    monkeypatch.setattr(inputs, "pair_for", lambda row: (case, side))
+    monkeypatch.setattr(inputs, "fetch_source", lambda *args: pytest.fail("catalog files must not fetch"))
+    target = tmp_path / "export"
+    proof = inputs.materialize({"units": [unit(file=case.relpath)]}, target)
+    assert proof["mode"] == "catalog-file"
+    assert (target / case.relpath).read_bytes() == (fixed if side == "fixed" else case.vuln_file).read_bytes()
+    assert (target / "pkg/context.py").read_bytes() == context.read_bytes()
+
+
+@pytest.mark.parametrize(
+    "body,reason",
+    [
+        ("# Provenance: example upstream_start: 25\n    def run():\n        return (\n", "excerpt does not parse:.*never closed"),
+        ("def run(:\n", "not complete Python"),
+    ],
+)
+def test_incomplete_catalog_without_commits_fails_specifically(tmp_path, monkeypatch, body, reason):
+    inputs, case = pair_inputs(tmp_path)
+    inputs.catalog_rows[case.name][0][1].update(parent="", commit="")
+    case.vuln_file.write_text(body)
+    monkeypatch.setattr(inputs, "pair_for", lambda row: (case, "vuln"))
+    with pytest.raises(ValueError, match=reason):
+        inputs.materialize({"units": [unit(file=case.relpath)]}, tmp_path / "export")
+
+
+@pytest.mark.parametrize("comment,newline", [("#", "\n"), ("//", "\n"), ("#", "\r\n")])
+@pytest.mark.parametrize("upstream", [True, False])
+def test_parseable_python_method_excerpt_is_in_function_only(tmp_path, monkeypatch, comment, newline, upstream):
+    inputs, case = pair_inputs(tmp_path)
+    inputs.catalog_rows[case.name][0][1].update(parent="", commit="")
+    header = f"{comment} Provenance: example\n" + (f"{comment} upstream_start: 25\n" if upstream else "")
+    body = header + "\n    def run(self, cmd):\n        os.system(cmd)\n"
+    case.vuln_file.write_bytes(body.replace("\n", newline).encode())
+    monkeypatch.setattr(inputs, "pair_for", lambda row: (case, "vuln"))
+    monkeypatch.setattr(inputs, "fetch_source", lambda *args: pytest.fail("excerpt attempted fetch"))
+    pin = {"units": [unit(file=case.relpath)]}
+    target = tmp_path / "export"
+    pin["materialization"] = inputs.materialize(pin, target)
+    assert pin["materialization"]["mode"] == "excerpt-parseable"
+    assert (target / case.relpath).read_text() == header.replace("//", "#") + "\ndef run(self, cmd):\n    os.system(cmd)\n"
+    assert trace.unit_record(pin["units"][0], "path")["evidence_scope"] == "in-function-only"
+    params = trace.prepare_questions(pin, target, 60)["a"]
+    assert params["callDepth"] == "0" and params["parameterSources"] == "true"
+
+
+def test_java_fragment_is_not_parseable_excerpt(tmp_path, monkeypatch):
+    inputs, case = pair_inputs(tmp_path)
+    inputs.catalog_rows[case.name][0][1].update(parent="", commit="")
+    case.relpath = "pkg/App.java"
+    case.vuln_file = case.fixed_file = tmp_path / "excerpt.java"
+    case.vuln_file.write_text("// Provenance: example\n// upstream_start: 25\n    public void run() {\n        execute(cmd);\n")
+    inputs.catalog_rows[case.name][0][1]["relpath"] = case.relpath
+    monkeypatch.setattr(inputs, "pair_for", lambda row: (case, "vuln"))
+    with pytest.raises(ValueError, match="excerpt does not parse:"):
+        inputs.materialize({"units": [unit(file=case.relpath, language="java")]}, tmp_path / "export")
+
+
+def test_excerpt_without_cheap_parser_is_rejected(tmp_path, monkeypatch):
+    monkeypatch.setattr("openultrasast.semantic.extra.grammar_for", lambda language: None)
+    with pytest.raises(ValueError, match="excerpt does not parse: no cheap parser for javascript"):
+        trace.parseable_excerpt(tmp_path / "app.js", "// Provenance: example upstream_start: 25\nfunction run() {}\n")
+
+
+def test_parseable_javascript_excerpt_uses_repository_parser(tmp_path):
+    from openultrasast.semantic.extra import grammar_for
+
+    if grammar_for("javascript") is None:
+        pytest.skip("optional JavaScript grammar unavailable")
+    assert (
+        trace.parseable_excerpt(tmp_path / "app.js", "// Provenance: example\n    function run(cmd) { execute(cmd); }\n")
+        == "// Provenance: example\nfunction run(cmd) { execute(cmd); }\n"
+    )
+
+
+def test_export_keeps_labels_over_cap_and_excluded_ancestors(tmp_path):
+    source, target = tmp_path / "source", tmp_path / "export"
+    for rel in ("templates/preferences.php", "templates/other.php", "other.php"):
+        path = source / rel
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text("<?php function run() {}\n")
+    proof = trace.export_sources(source, target, [unit(file="templates/preferences.php", language="php")], cap=1)
+    assert (target / "templates/preferences.php").read_bytes() == (source / "templates/preferences.php").read_bytes()
+    assert not (target / "templates/other.php").exists()
+    assert proof["cap_hit"] and proof["bytes"] > proof["cap_bytes"]
+
+
+def test_oversized_package_narrows_before_truncation_and_keeps_import(tmp_path):
+    source, target = tmp_path / "source", tmp_path / "export"
+    for rel, body in {
+        "pkg/sub/app.py": "from helper import value\ndef run(): pass\n",
+        "pkg/sub/context.py": "x = 1\n",
+        "pkg/huge.py": "#" * 500,
+        "helper.py": "value = 1\n",
+    }.items():
+        path = source / rel
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(body)
+    proof = trace.export_sources(source, target, [unit(file="pkg/sub/app.py")], ["pkg"], cap=100)
+    assert proof["roots"] == ["pkg/sub"] and proof["requested_roots"] == ["pkg"]
+    assert proof["scope_narrowed"] and not proof["cap_hit"]
+    assert (target / "helper.py").exists() and (target / "pkg/sub/context.py").exists()
+    assert not (target / "pkg/huge.py").exists()
+
+
+def test_local_repository_export_scopes_packages(tmp_path, monkeypatch):
+    inputs, _ = pair_inputs(tmp_path)
+
+    def export(clone, pin, dest):
+        for rel in ("packages/server/run.js", "packages/other/skip.js"):
+            path = dest / rel
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_text("function run() {}\n")
+
+    monkeypatch.setattr(trace, "export_pin", export)
+    proof = inputs.export_local(
+        tmp_path, {"pin": "sha", "units": [unit(file="packages/server/run.js", language="javascript")]}, tmp_path / "out"
+    )
+    assert proof["roots"] == ["packages/server"]
+    assert not (tmp_path / "out/packages/other/skip.js").exists()
+
+
+def test_census_membership_is_uncapped_and_warning_is_not_absence(tmp_path, monkeypatch):
+    class LargeBackend(FakeBackend):
+        def build(self, root, language):
+            cpg = super().build(root, language)
+            original = cpg.run_batch_once
+
+            def batch(query, requests):
+                result = original(query, requests)
+                result["__census__"] = [{"files": "3001", "file_names": [f"f{i}.py" for i in range(3000)] + ["/case/app.py"]}]
+                return result
+
+            return replace(cpg, run_batch_once=batch, unparsed=("app.py",))
+
+    result = run_fake(tmp_path, monkeypatch, LargeBackend)
+    assert [u["status"] for u in result["units"]] == ["path", "asked-nothing"]
+    for name in ("census.sc", "taint.sc"):
+        assert '"file_names" -> ujson.Arr.from(cpg.file.name.l)' in (trace.ROOT / "src/openultrasast/cpg/queries" / name).read_text()
+
+
+def test_absent_label_includes_frontend_parse_log(tmp_path, monkeypatch):
+    class MissingBackend(FakeBackend):
+        def build(self, root, language):
+            cpg = super().build(root, language)
+            self.stages[0]["labelled_file_logs"] = {"app.py": ["Failed to parse app.py: invalid syntax"]}
+
+            def batch(query, requests):
+                self.asked.extend(requests)
+                return {**dict.fromkeys(requests, []), "__census__": [{"files": "1", "file_names": ["<unknown>"]}]}
+
+            return replace(cpg, run_batch_once=batch)
+
+    result = run_fake(tmp_path, monkeypatch, MissingBackend)
+    assert all(u["status"] == "failed" and "Failed to parse app.py: invalid syntax" in u["reason"] for u in result["units"])
+
+
+def test_frontend_retains_matching_log_lines(tmp_path):
+    launcher = tmp_path / "pysrc2cpg"
+    launcher.write_text(f"#!{sys.executable}\nprint('Failed to parse /case/pkg/app.py: syntax')\n")
+    launcher.chmod(0o755)
+    backend = worker.MeasuredBackend(30, 10, lambda: None)
+    backend.labelled_files = ["pkg/app.py"]
+    backend._run([str(launcher)], timeout=10)
+    assert backend.stages[0]["labelled_file_logs"]["pkg/app.py"] == ["Failed to parse /case/pkg/app.py: syntax"]
+
+
+@pytest.mark.parametrize("success", [True, False])
+def test_engine_bug_retries_taint_only_once(tmp_path, monkeypatch, success):
+    backend = worker.MeasuredBackend(900, 120, lambda: None)
+    calls = []
+
+    def batch(self, graph, query, requests):
+        calls.append((self.queries_dir / "taint.sc").read_text())
+        if len(calls) == 1:
+            self.fatal = worker.ENGINE_BUG
+            self.asked.append("stale")
+            self.answers["stale"] = []
+            return None
+        assert not self.asked and not self.answers
+        assert "enhance = false" in calls[-1]
+        assert "requires saved dataflowOss overlay" in calls[-1]
+        if success:
+            return {"a": [], "__census__": [{"files": "1", "file_names": ["app.py"]}]}
+        self.fatal = "joern query exited 1"
+        return None
+
+    monkeypatch.setattr(JoernBackend, "_batch_once", batch)
+    result = backend._batch_once(tmp_path / "cpg.bin", "taint", {"a": {}})
+    assert len(calls) == 2 and backend.retries[0]["completed"] == success
+    assert (result is not None) == success
+    assert backend.fatal == ("" if success else worker.ENGINE_BUG)
+    if not success:
+        backend._batch_once(tmp_path / "cpg.bin", "taint", {"a": {}})
+        assert len(backend.retries) == 1
+
+
+def test_named_engine_bug_from_process_output(tmp_path):
+    launcher = tmp_path / "joern"
+    launcher.write_text(
+        f"#!{sys.executable}\nprint('Pass io.joern.x2cpg.frontendspecific.jssrc2cpg.ObjectPropertyCallLinker failed')\n"
+        "print('java.lang.RuntimeException: Assignment statement with 3 arguments')\nraise SystemExit(1)\n"
+    )
+    launcher.chmod(0o755)
+    backend = worker.MeasuredBackend(30, 10, lambda: None)
+    backend._run([str(launcher), "--script", "query.sc"], timeout=10)
+    assert backend.fatal == worker.ENGINE_BUG
+
+
+def test_rerun_failed_and_timeout_only_preserves_completed_v2(tmp_path):
+    units = [unit(status, pin=status) for status in trace.STATUSES]
+    for pin in trace.make_plan(units, tmp_path):
+        trace.save_pin(tmp_path / trace.pin_name(pin), {**pin, "done": True, "units": [trace.unit_record(pin["units"][0], pin["pin"])]})
+    assert {p["pin"] for p in trace.make_plan(units, tmp_path, rerun_status={"failed", "timeout"})} == {"failed", "timeout"}
+
+
+def test_successful_retry_keeps_result_and_records_attempt(tmp_path, monkeypatch):
+    class RetriedBackend(FakeBackend):
+        def __init__(self, *args):
+            super().__init__(*args)
+            self.retries = [{"reason": worker.ENGINE_BUG, "mode": "taint-only", "completed": True}]
+            self.stages.append({"command": "joern", "seconds": 20, "exit": 1, "output_tail": worker.ENGINE_BUG})
+
+    result = run_fake(tmp_path, monkeypatch, RetriedBackend)
+    assert [u["status"] for u in result["units"]] == ["path", "asked-nothing"]
+    assert all(u["reason"] == "" for u in result["units"])
+    assert result["instrument"]["retries"][0]["completed"]
+
+
+def test_overlay_engine_bug_allows_one_taint_only_attempt(tmp_path, monkeypatch):
+    backend = worker.MeasuredBackend(900, 120, lambda: None)
+
+    def overlay(self, graph, scratch):
+        self.fatal = worker.ENGINE_BUG
+        return False
+
+    monkeypatch.setattr(JoernBackend, "_apply_overlays", overlay)
+    assert not backend._apply_overlays(tmp_path / "cpg.bin", tmp_path)
+    assert backend.retry_pending and not backend.fatal
+
+    def batch(self, graph, query, requests):
+        assert "enhance = false" in (self.queries_dir / "taint.sc").read_text()
+        return {"a": []}
+
+    monkeypatch.setattr(JoernBackend, "_batch_once", batch)
+    assert backend._batch_once(tmp_path / "cpg.bin", "taint", {"a": {}}) == {"a": []}
+    assert len(backend.retries) == 1