learn: engine_trace materialises catalog pairs without commits (dedented parseable excerpts, in-function scope), protects labelled files in exports, scopes large repos, records frontend diagnostics, retries the jssrc2cpg engine bug, writes progress.json

Implemented with Codex.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

benchmarks/learn/engine_trace.py

@@ -7,6 +7,7 @@ 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
@@ -15,6 +16,7 @@ import re
 import shutil
 import subprocess
 import tempfile
+import textwrap
 import time
 import tomllib
 from collections import Counter, defaultdict
@@ -41,7 +43,7 @@ SNAPSHOT_FILES = (
     "src/openultrasast/model/taint.py",
     "src/openultrasast/cpg/queries/taint.sc",
 )
-MATERIALIZATION_VERSION = 2
+MATERIALIZATION_VERSION = 3
 EXPORT_CAP = 20 * 1024 * 1024
 
 STATUSES = ("path", "asked-nothing", "failed", "timeout", "unsupported")
@@ -268,7 +270,8 @@ class Inputs:
     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))
-            proof = export_sources(Path(scratch), target, pin["units"])
+            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):
@@ -279,7 +282,43 @@ class Inputs:
                 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 = {self.source_identity(case, side) for case, side in resolved}
+            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()
@@ -295,7 +334,7 @@ class Inputs:
                     _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, "roots": roots}
+            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"):
@@ -310,43 +349,168 @@ class Inputs:
         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):
-    """Preserve labelled paths first; cap context and exclude non-source trees."""
+    """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 not in EXCLUDED_DIRS and d != ".git" and not (Path(directory) / d).is_symlink())
+        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()
-            if (
+            excluded = any(part in EXCLUDED_DIRS for part in Path(rel).parts[:-1])
-                not file.is_symlink()
+            if not file.is_symlink() and (
-                and detect_language(file) in languages
+                rel in labelled
-                and any(root == "." or rel.startswith(root + "/") for root in roots)
+                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)}")
-    proof = {"bytes": 0, "files": 0, "cap_bytes": cap, "cap_hit": False, "excludes": sorted(EXCLUDED_DIRS)}
+    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 = file.stat().st_size
+        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 in labelled:
+            if rel not in labelled:
-                raise ValueError("labelled source exceeds 20 MB export cap")
+                continue
-            continue
         dest = safe_path(target, rel)
         dest.parent.mkdir(parents=True, exist_ok=True)
         dest.write_bytes(file.read_bytes())
@@ -431,7 +595,7 @@ def prepare_questions(pin, root, question_deadline):
         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 = pin.get("materialization", {}).get("mode") == "excerpt-fallback"
+        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"],
@@ -573,7 +737,7 @@ def main():
     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", help="re-run completed pins with any selected unit in these statuses")
+    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"
     )
@@ -652,6 +816,9 @@ def main():
     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
@@ -660,8 +827,13 @@ def main():
                     args.out / pin_name(pin),
                     {**pin, "done": True, "units": [unit_record(u, "unsupported", u["reason"]) for u in pin["units"]]},
                 )
-                continue
+            else:
-            failures |= run_pin(pin, source, provenance, inputs, args.out, args)
+                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
 
 

benchmarks/learn/engine_trace_worker.py

@@ -14,11 +14,12 @@ import os
 import selectors
 import signal
 import subprocess
+import tempfile
 import time
 from collections import Counter
 from pathlib import Path
 
-from openultrasast.cpg.backend import JoernBackend, _render
+from openultrasast.cpg.backend import DATAFLOW_OVERLAY, JoernBackend, _render
 from openultrasast.model.trace import parse_trace
 
 
@@ -39,6 +40,7 @@ def safe_path(root, name):
 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"})
 
 
@@ -74,6 +76,61 @@ class MeasuredBackend(JoernBackend):
         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:
@@ -140,6 +197,12 @@ class MeasuredBackend(JoernBackend):
                 "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
@@ -162,6 +225,9 @@ class MeasuredBackend(JoernBackend):
                     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()
@@ -219,6 +285,8 @@ def worker_language(pin, root, out, deadline, question_deadline, persist=write_j
 
     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"]]
@@ -281,13 +349,14 @@ def worker_language(pin, root, out, deadline, question_deadline, persist=write_j
                 extra.update(witness_rows=raw_flows, traces=[parse_trace(r) for r in raw_flows])
             if shard_incomplete:
                 extra["questions_completed"] = []
-            dropped = any(p.endswith(unit["file"]) for p in cpg.unparsed)
             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)
-            if file_missing and rows is not None:
+            dropped = file_missing and rows is not None
-                dropped = True
             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:
@@ -319,7 +388,7 @@ def worker_language(pin, root, out, deadline, question_deadline, persist=write_j
             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"]:
+                    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"]]

tests/test_engine_trace.py

@@ -530,6 +530,8 @@ def test_trace_real_unsupported_run_then_typescript_opt_in_replans(tmp_path, mon
     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 = []
@@ -676,6 +678,7 @@ def pair_inputs(tmp_path):
 @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")
@@ -897,3 +900,261 @@ def test_sparse_fetch_from_local_git_without_network(tmp_path, monkeypatch):
     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