wip(archive): fix-mechanisms arm — parked WIP preserved before worktree prune

Uncommitted work from the fix-mechanisms worktree (mechanisms.py + cst/extra/cli
edits), committed so the worktree can be removed without loss. Branch is 0 commits
on top of a 172-behind base; re-integrate via rebase if resumed.

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

.github/workflows/engine-image.yml

@@ -0,0 +1,63 @@
+name: Engine image
+on:
+  workflow_dispatch:
+  push:
+    branches: [main]
+    paths:
+      - 'src/**'
+      - 'benchmarks/learn/engine_trace*.py'
+      - 'benchmarks/learn/engine_task_entry.py'
+      - 'benchmarks/learn/engine_worker_service.py'
+      - 'Dockerfile'
+      - 'plane/Dockerfile.engine-task'
+      - 'ops/frontend-retention/install.py'
+      - 'pyproject.toml'
+      - '.github/workflows/engine-image.yml'
+permissions:
+  contents: read
+  packages: write
+concurrency:
+  group: engine-image-${{ github.ref }}
+  cancel-in-progress: false
+jobs:
+  image:
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          - package: openultrasast
+            dockerfile: Dockerfile
+          - package: ousast-engine-task
+            dockerfile: plane/Dockerfile.engine-task
+    steps:
+      - uses: actions/checkout@v5
+      - uses: docker/setup-buildx-action@v3
+      - uses: docker/login-action@v3
+        with:
+          registry: ghcr.io
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+      - id: names
+        env:
+          OWNER: ${{ github.repository_owner }}
+          PACKAGE: ${{ matrix.package }}
+        run: |
+          echo "image=ghcr.io/${OWNER,,}/$PACKAGE" >> "$GITHUB_OUTPUT"
+          echo "sha=sha-${GITHUB_SHA:0:7}" >> "$GITHUB_OUTPUT"
+      - id: build
+        uses: docker/build-push-action@v6
+        with:
+          context: .
+          file: ${{ matrix.dockerfile }}
+          platforms: linux/amd64
+          push: true
+          tags: |
+            ${{ steps.names.outputs.image }}:${{ steps.names.outputs.sha }}
+            ${{ steps.names.outputs.image }}:main
+      - env:
+          IMAGE: ${{ steps.names.outputs.image }}
+          DIGEST: ${{ steps.build.outputs.digest }}
+        run: |
+          echo "$IMAGE@$DIGEST"
+          echo "$IMAGE@$DIGEST" >> "$GITHUB_STEP_SUMMARY"

.kiro/specs/learned-decision-engine/fix-mechanisms-plan.md

@@ -0,0 +1,80 @@
+# Plan: learn fix mechanisms into the engine (draft for review, 2026-10-04)
+
+Status: approved 2026-10-04 by the maintainer. Increment 1 = discharge facts only; coverage additions (Req 7) come later.
+Decisions: minimum 2 proposing repositories before a candidate is tried; admission per language (not pooled by callee name).
+
+## Why
+
+The Joern trace run over exp-005's 501 units shows the engine cannot tell a vulnerable function from its
+fix. In the 105 pairs where both sides were analysed, a taint path appears on both sides in 12, on the
+vulnerable side only in 2, on the fixed side only in 0, and on neither in 91
+(`benchmarks/measurements/2026-10-04-joern-slice-leak-audit/record.json`).
+
+So the fix is invisible to the engine. Either the flow survives and the engine does not know the fix's
+mechanism (an `allowed_classes` argument, a path normalisation check, an escaping call), or the engine's
+sources and sinks miss the function on both sides. A classifier fed this evidence cannot improve on it.
+
+The corpus now holds 710 complete fix pairs (`2026-10-04-decision-engine-harvest-3/record.json`). Each fix
+diff shows what was added. That is the material the engine lacks.
+
+## Goal
+
+Grow the engine's discharge knowledge (sanitizers, guards, safe-argument forms) and, second, its source and
+sink coverage, from the fix diffs. Admit a fact only when it improves held-out separation. Spend no model
+money.
+
+## Requirements (proposed)
+
+1. **Mining.** From each fix pair, extract what the fixed function adds on the vulnerable flow: new calls,
+   new conditions guarding the sink, new or changed sink arguments. Record each as a candidate fact with its
+   language, family, form (sanitizer call, guard predicate, safe argument, sink restriction) and the pairs
+   that propose it. Mining is deterministic (tree-sitter diff of the two function bodies). No model calls.
+2. **Generalisation is explicit.** A candidate is a pattern, not a repository name: a callee name, an
+   argument key, a predicate shape. Candidates naming project-specific identifiers are rejected (Req 8: no
+   application-specific trees).
+3. **Admission by held-out separation.** Folds are by repository, as in the decision engine. A candidate is
+   admitted for a fold only if, on that fold's held-out pairs, adding it to the fact tables:
+   - increases pairs where the vulnerable side keeps a path and the fixed side loses it or becomes
+     sanitized;
+   - and does not remove any held-out vulnerable-side path (no recall loss).
+   A candidate proposed by fewer than 2 distinct repositories in the training folds is not tried.
+4. **No train-on-test.** Mining and admission never read held-out pairs; the closed-loop leak recorded in
+   memory (`closed-loop-train-on-test-leak`) is the failure this rule exists to prevent. Population v3 is
+   never read.
+5. **Measurement by the trace runner.** Each admission round re-runs the engine on the affected pins only
+   (the runner's resume and selection already support this; on k3s the worker Deployment runs them in
+   parallel). The outcome is the pair-asymmetry table per family, before and after, on held-out folds.
+6. **Gate before production.** An admitted fact enters `src/openultrasast/ruleset/semantic/*.toml` only after
+   the detection gates (`gate`, `map_gate`, `pair_gate`) show no regression, and only as a registered
+   experiment whose decision rule was fixed before the run.
+7. **Coverage second.** After discharge facts, the same loop proposes source and sink additions from pairs
+   where neither side has a path (91 of 105 today), admitted under the same held-out rule.
+
+## How it relates to what exists
+
+- `corpus-seeded-mechanisms` already has a closed loop that admits and retracts mechanisms by leave-one-out.
+  This plan reuses its lever and store, and points it at discharge facts mined from real fix diffs instead
+  of variant search.
+- `benchmarks/pairs/mechanisms.toml` (23 mechanisms) labels what kind of bug a pair is. The new candidates
+  describe what a fix does. They live in the fact tables, not in the mechanism vocabulary.
+
+## Measures of success
+
+- Pair asymmetry on held-out folds: vulnerable-only paths rise from 2 of 105 toward a substantial share,
+  with zero fixed-only paths.
+- No held-out recall loss on vulnerable sides.
+- Then the decision engine is re-run with the new engine evidence (exp-005 design), and the pooled block gate
+  (exp-004 design) is re-measured.
+
+## Cost and time
+
+- Model cost: $0 (mining and admission are deterministic; the classifier re-run afterwards is a separate,
+  approved spend).
+- Engine time: one pass over affected pins per admission round; about 2 hours per full pass on k3s plus the
+  VM, about 6 hours on the VM alone.
+
+## Open questions for the maintainer
+
+1. Should coverage additions (Req 7) be in the same increment or a later one?
+2. Minimum proposing repositories: 2 is the proposal. Higher is safer, lower finds more.
+3. Admission per family and language, or pooled across languages for the same callee name?

Dockerfile

@@ -52,7 +52,7 @@ COPY benchmarks ./benchmarks
 # The core install is zero-dependency by design; `semantic` adds the tree-sitter grammars the candidate
 # enumerator uses. The LLM endpoint stays optional and is configured by environment, never baked in.
 RUN python3 -m venv /venv \
- && /venv/bin/pip install --no-cache-dir -e ".[semantic]"
+ && /venv/bin/pip install --no-cache-dir -e ".[semantic,s3]"
 ENV PATH="/venv/bin:${PATH}"
 
 # Analysis never needs root.

README.md

@@ -363,3 +363,81 @@ own terms:
 - Joern (Apache-2.0) is downloaded at image build time.
 - The tree-sitter wheels of the `semantic` extra are MIT.
 - The published benchmark cases were licence-checked one by one. v1's OpenCVE case is BUSL-1.1, see [RELEASE_NOTES.md](RELEASE_NOTES.md).
+
+## Engine traces on kube-ax
+
+`benchmarks/learn/engine_trace.py --executor ax` submits AX v0.3.1 Tasks with
+`ax apply -f -` and removes them with `ax delete task NAME`. The operator supplies
+kubeconfig and AX access; the dispatcher creates no Kubernetes resources or secrets.
+The `ousast-engine-task` GHCR package **must be made Public once** in its package
+settings so the actors can pull it without registry credentials.
+
+The task image contains the current engine and worker code. Rebuild it after code
+changes and pass its digest with `--image`; no analyzer download is needed. The host
+materializes each pin as usual and uploads source and prepared questions under
+`engine-queue/<run>/<pin>/`. Only object-scoped presigned URLs and deadline parameters
+enter Task env. Credentials stay on the coordinator. Configure `OUSAST_MEMORY=s3://…`
+and the store credentials in the coordinator's `.env` or environment, with
+`S3_ENDPOINT=https://files.because-security.com`. Presigned links expire after the
+pin deadline plus ten minutes. IP and HTTP endpoints are rejected: the operator's
+actor EgressPolicy must allow **files.because-security.com:443**.
+
+Completion is the result object, even while AX reports Running. Failed phases and
+host deadlines produce explicit failure/timeout rows. Task and staging-object cleanup
+runs on success and failure; cleanup failure stops further dispatch. Per-pin files,
+`--rerun-status`, resume and `progress.json` retain the existing trace-runner behavior.
+AX results include `executor: "ax"`, `worker_ip`, and the image digest. Placement is
+random; `--parallel` defaults to two AX tasks and cannot exceed two.
+
+AX workers use `--heap-profile cluster` with
+`-Xms1700m -Xmx1700m -XX:MaxMetaspaceSize=256m -XX:ReservedCodeCacheSize=128m -XX:MaxDirectMemorySize=256m -Xss512k -XX:ActiveProcessorCount=1 -XX:+UseG1GC`
+on every frontend/query JVM; instruments record the flags. A `JVM out of memory`
+result is automatically re-queued to one local Docker lane with `--heap-profile vm`
+(2560m), recording `fallback: "vm-oom"` and the AX attempt. `mixed-ax` also uses this
+one Docker lane for overflow, prioritizing pending OOM fallbacks. Thus a mixed run has
+up to two AX tasks plus one Docker container. Use `--docker-image` for the locally
+available VM image; Docker never pulls it implicitly. A STOP file prevents new
+launches; interrupted fallback queues are recovered on resume.
+
+Coordinator commands, from this worktree (the AX binary and kubeconfig remain
+configurable). Build/push the actual uncommitted tree and read its digest:
+
+```bash
+cd /home/mc/Source/OpenUltraSAST/.claude/worktrees/engine-ax
+export PATH=/home/mc/Source/OpenUltraSAST/.venv/bin:$PATH
+export PYTHONPATH="$PWD/src"
+TASK_TAG="ghcr.io/norandom/ousast-engine-task:sha-$(git rev-parse --short=7 HEAD)"
+docker buildx build --platform linux/amd64 -f plane/Dockerfile.engine-task \
+  -t "$TASK_TAG" -t ghcr.io/norandom/ousast-engine-task:main \
+  --push --metadata-file /tmp/ousast-engine-task-build.json .
+export TASK_IMAGE="ghcr.io/norandom/ousast-engine-task@$(python -c 'import json; print(json.load(open("/tmp/ousast-engine-task-build.json"))["containerimage.digest"])')"
+printf '%s\n' "$TASK_IMAGE"
+docker build -f Dockerfile -t openultrasast:engine-ax .
+```
+
+The Engine image workflow also builds/pushes both images and prints each digest;
+CI needs the changes published before dispatch. The local build above includes
+uncommitted changes without publishing a git commit.
+
+With `KUBECONFIG` pointing to the operator's kube-ax config and store settings in
+`.env`, run a one-pin smoke using the default experiment units and local example
+memory (or add `--examples /path/to/frozen-examples.jsonl`):
+
+```bash
+: "${KUBECONFIG:?Set KUBECONFIG to the operator kube-ax configuration}"
+python benchmarks/learn/engine_trace.py --executor ax \
+  --ax-bin "$HOME/.local/ax-v0.3.1/ax" --kubeconfig "$KUBECONFIG" \
+  --atespace default --image "$TASK_IMAGE" --docker-image openultrasast:engine-ax \
+  --parallel 1 --limit 1 --deadline 900 \
+  --out "$HOME/ousast-results/plane/engine-ax-smoke"
+```
+
+Then run the mixed lanes (repeat to resume; defaults re-run failed/timeout pins):
+
+```bash
+python benchmarks/learn/engine_trace.py --executor mixed-ax \
+  --ax-bin "$HOME/.local/ax-v0.3.1/ax" --kubeconfig "$KUBECONFIG" \
+  --atespace default --image "$TASK_IMAGE" --docker-image openultrasast:engine-ax \
+  --parallel 2 --deadline 900 --rerun-status failed,timeout \
+  --out "$HOME/ousast-results/plane/engine-ax-mixed"
+```

benchmarks/learn/build_features.py

@@ -126,8 +126,9 @@ def resolve_scan(scans: Path, case: str, side: str) -> dict[str, Any] | None:
 
 
 def recorded_engine(language: str, findings: list, result: Mapping | None, repository: bool) -> Part:
-    # Recorded findings take precedence over the shared engine language coverage.
-    if language not in ENGINE_LANGUAGES and not findings:
+    # The language list describes standalone scans. Repository scans also analyze
+    # TypeScript; recorded function findings take precedence over that list.
+    if language not in ENGINE_LANGUAGES and not findings and not (repository and language == "typescript" and result is not None):
         return NONE
     state = engine_for(result)
     if state is not None:

benchmarks/learn/engine_task_entry.py

@@ -0,0 +1,173 @@
+"""AX task entry: object-scoped transfers and stdlib only; result PUT is completion."""
+
+from __future__ import annotations
+
+import http.client
+import json
+import os
+import shutil
+import subprocess
+import sys
+import tarfile
+import time
+import urllib.error
+import urllib.request
+from pathlib import Path
+
+
+class TaskEntryError(RuntimeError):
+    """A task-entry diagnostic containing only locally controlled text."""
+
+
+def transfer(request, destination=None, *, readiness_budget=None):
+    """Wait for actor egress on startup; bound later transfers to three attempts."""
+    deadline = time.monotonic() + readiness_budget if readiness_budget is not None else None
+    delay = 0.5
+    attempt = 0
+    while True:
+        remaining = deadline - time.monotonic() if deadline is not None else 60
+        if remaining <= 0:
+            raise TaskEntryError("egress not open after 60 s")
+        attempt += 1
+        try:
+            with urllib.request.urlopen(request, timeout=min(60, remaining)) as response:
+                if destination is not None:
+                    with destination.open("wb") as output:
+                        shutil.copyfileobj(response, output)
+            return
+        except urllib.error.HTTPError as exc:
+            exc.close()
+            if exc.code != 403 and not 500 <= exc.code < 600:
+                raise
+            if deadline is None and attempt >= 3:
+                raise
+        except (urllib.error.URLError, OSError, http.client.HTTPException):
+            if deadline is None and attempt >= 3:
+                raise
+        if deadline is not None:
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                raise TaskEntryError("egress not open after 60 s") from None
+            time.sleep(min(delay, remaining))
+            delay = min(delay * 2, 8)
+        else:
+            time.sleep(attempt)
+
+
+def download(url, destination, *, readiness_budget=None):
+    transfer(url, destination, readiness_budget=readiness_budget)
+
+
+def unpack(source, destination):
+    """Reject links and traversal; source archives contain only regular files/directories."""
+    with tarfile.open(source) as archive:
+        for member in archive.getmembers():
+            target = (destination / member.name).resolve()
+            if not target.is_relative_to(destination.resolve()) or not (member.isfile() or member.isdir()):
+                raise ValueError("unsafe source archive member")
+        archive.extractall(destination, filter="data")
+
+
+def failure_record(pin, reason):
+    return {
+        **pin,
+        "done": True,
+        "status": "failed",
+        "reason": reason,
+        "units": [
+            {
+                **unit,
+                "status": "failed" if unit.get("supported", True) else "unsupported",
+                "reason": reason,
+                "instrument": {"files": [], "bytes": 0, "jvm": []},
+                "questions": {},
+                "questions_asked": [],
+                "questions_completed": [],
+                "witness_rows": [],
+                "traces": [],
+            }
+            for unit in pin.get("units", [])
+        ],
+    }
+
+
+def main(*, work_root=Path("/tmp"), worker_script=None, readiness_budget=60):
+    pin = {}
+    failed = False
+    output = work_root / "out"
+    result = output / "result.json"
+    step = "prepare output"
+    try:
+        output.mkdir(parents=True, exist_ok=True)
+        result.unlink(missing_ok=True)
+        questions = work_root / "questions.json"
+        step = "GET questions"
+        download(os.environ["QUESTIONS_URL"], questions, readiness_budget=readiness_budget)
+        step = "parse questions"
+        pin = json.loads(questions.read_text())
+        if not isinstance(pin, dict):
+            pin = {}
+            raise ValueError("questions object must contain a prepared pin")
+        source = work_root / "source.tar"
+        step = "GET source"
+        download(os.environ["SOURCE_URL"], source)
+        step = "unpack source"
+        case = work_root / "case"
+        case.mkdir(parents=True, exist_ok=True)
+        unpack(source, case)
+        step = "configure worker"
+        deadline = int(os.environ.get("DEADLINE", "900"))
+        command = [
+            sys.executable,
+            str(worker_script or Path(__file__).with_name("engine_trace_worker.py")),
+            "--worker",
+            str(questions),
+            "--root",
+            str(case),
+            "--output",
+            str(result),
+            "--heap-profile",
+            "cluster",
+            "--deadline",
+            str(deadline),
+            "--question-deadline",
+            os.environ.get("QUESTION_DEADLINE", "120"),
+        ]
+        step = "run worker"
+        completed = subprocess.run(command, timeout=deadline + 5, check=False)
+        if completed.returncode:
+            raise TaskEntryError(f"worker exited {completed.returncode}")
+        step = "read worker result"
+        if not result.is_file() or not json.loads(result.read_text()).get("done"):
+            raise TaskEntryError("worker did not write a completed result")
+    except Exception as exc:
+        failed = True
+        # Avoid including exception text containing a presigned URL in durable output.
+        if isinstance(exc, urllib.error.HTTPError):
+            detail = f"HTTP {exc.code}"
+        else:
+            detail = str(exc) if isinstance(exc, TaskEntryError) else type(exc).__name__
+        reason = f"{step}: {detail}"
+        try:
+            output.mkdir(parents=True, exist_ok=True)
+            result.write_text(json.dumps(failure_record(pin, reason)) + "\n")
+        except Exception:
+            print("task entry could not write failure result", file=sys.stderr)
+    step = "pack result"
+    try:
+        archive = work_root / "result.tar"
+        with tarfile.open(archive, "w") as tar:
+            for path in sorted(output.iterdir()):
+                tar.add(path, arcname=path.name)
+        data = archive.read_bytes()
+        step = "PUT result"
+        request = urllib.request.Request(os.environ["RESULT_URL"], data=data, method="PUT", headers={"Content-Type": "application/x-tar"})
+        transfer(request, readiness_budget=readiness_budget if failed else None)
+    except Exception as exc:
+        detail = f"HTTP {exc.code}" if isinstance(exc, urllib.error.HTTPError) else type(exc).__name__
+        print(f"{step}: {detail}; dispatcher deadline remains authoritative", file=sys.stderr)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

benchmarks/learn/engine_trace_ax.py

@@ -0,0 +1,319 @@
+"""Host-only AX Task transport; credentials never enter task manifests."""
+
+from __future__ import annotations
+
+import hashlib
+import io
+import ipaddress
+import json
+import os
+import queue
+import re
+import shutil
+import subprocess
+import tarfile
+import tempfile
+import threading
+import time
+import uuid
+from datetime import timedelta
+from pathlib import Path
+from urllib.parse import urlsplit
+
+import yaml
+from engine_trace_k8s import pack
+
+from openultrasast.config import load_dotenv
+from openultrasast.plane.memory import PRESIGNED_ARCHIVE_SUFFIX, S3Store, open_store
+
+
+class EgressError(ValueError):
+    """Safe diagnostic without presigned URL contents."""
+
+
+class SandboxFailure(Exception):
+    """A terminal AX task without an authoritative result object."""
+
+
+def public_endpoint(url):
+    """Reject endpoints the actor's hostname-based HTTPS egress policy cannot reach."""
+    parts = urlsplit(url)
+    host = parts.hostname or ""
+    try:
+        ipaddress.ip_address(host)
+        is_ip = True
+    except ValueError:
+        is_ip = False
+    if (
+        parts.scheme != "https"
+        or not host
+        or is_ip
+        or "." not in host
+        or host.endswith((".localhost", ".local", ".internal"))
+        or parts.port not in (None, 443)
+        or parts.username
+        or parts.password
+    ):
+        raise EgressError(
+            "AX requires a public HTTPS DNS hostname in S3_ENDPOINT and presigned URLs: "
+            "the egress policy allows the store hostname on port 443 only (no IP or plain HTTP)"
+        )
+    return host.lower()
+
+
+def task_manifest(name, args, urls):
+    if not re.fullmatch(r"[^\s@]+@sha256:[0-9a-f]{64}", args.image):
+        raise ValueError("AX --image must be pinned by sha256 digest")
+    if set(urls) != {"SOURCE_URL", "QUESTIONS_URL", "RESULT_URL"}:
+        raise ValueError("AX Task accepts only source, questions and result presigned URLs")
+    for url in urls.values():
+        public_endpoint(url)
+    env = [{"name": key, "value": value} for key, value in urls.items()]
+    env.extend(
+        {"name": key, "value": str(value)}
+        for key, value in {"DEADLINE": args.deadline, "QUESTION_DEADLINE": args.question_deadline}.items()
+    )
+    if len(json.dumps(env).encode()) >= 32768:
+        raise ValueError("AX Task env exceeds the 32 KB limit")
+    return {
+        "apiVersion": "ax.io/v1alpha1",
+        "kind": "Task",
+        "metadata": {"name": name, "atespace": args.atespace},
+        "spec": {"image": args.image, "env": env, "command": ["python3", "/app/benchmarks/learn/engine_task_entry.py"]},
+    }
+
+
+def task_state(text, name):
+    """Accept AX YAML/JSON output and its human-readable tasks table."""
+    try:
+        doc = yaml.safe_load(text)
+    except yaml.YAMLError:
+        doc = None
+    rows = doc.get("items", doc.get("tasks", [doc])) if isinstance(doc, dict) else doc if isinstance(doc, list) else []
+    for row in rows:
+        if not isinstance(row, dict) or row.get("metadata", {}).get("name", row.get("name")) != name:
+            continue
+        status = row.get("status", {})
+        ip = next(
+            (status.get(key) or row.get(key) for key in ("workerIP", "workerIp", "worker_ip", "ip") if status.get(key) or row.get(key)),
+            None,
+        )
+        worker = status.get("worker")
+        if not ip and isinstance(worker, dict):
+            ip = worker.get("ip") or worker.get("address")
+        return str(status.get("phase", row.get("phase", ""))), ip
+    lines = [line.split() for line in text.splitlines() if line.strip()]
+    if lines:
+        headers = [re.sub(r"[^a-z]", "", cell.lower()) for cell in lines[0]]
+        for cells in lines[1:]:
+            if name not in cells:
+                continue
+            row = dict(zip(headers, cells, strict=False))
+            phase = row.get("phase", row.get("status", ""))
+            ip = row.get("workerip") or row.get("ip")
+            if not ip:
+                for cell in cells:
+                    try:
+                        ipaddress.ip_address(cell)
+                        ip = cell
+                        break
+                    except ValueError:
+                        pass
+            return phase, None if ip in {"<none>", "-", ""} else ip
+    listed = isinstance(doc, (dict, list)) or bool(lines and "name" in headers and {"phase", "status"}.intersection(headers))
+    return "Missing" if listed else "", None
+
+
+def bounded_read(read, timeout):
+    """Do not let SDK retries postpone task teardown beyond the pin deadline.
+
+    Only the read runs in a daemon thread; it cannot publish a late result to disk.
+    No executor shutdown waits for an SDK call whose socket has stopped responding.
+    """
+    if timeout <= 0:
+        raise TimeoutError("AX result deadline")
+    replies = queue.Queue(maxsize=1)
+
+    def fetch():
+        try:
+            replies.put((True, read()))
+        except Exception as exc:
+            replies.put((False, exc))
+
+    threading.Thread(target=fetch, daemon=True, name="engine-ax-result-read").start()
+    try:
+        ok, result = replies.get(timeout=timeout)
+    except queue.Empty:
+        raise TimeoutError("AX result deadline") from None
+    if not ok:
+        raise result
+    return result
+
+
+class AX:
+    def __init__(self, args, store=None, pause=time.sleep, clock=time.monotonic):
+        self.args = args
+        self.pause, self.clock = pause, clock
+        self.run = uuid.uuid4().hex
+        load_dotenv()
+        self.host = public_endpoint(os.environ.get("S3_ENDPOINT", ""))
+        # Validate the image before opening a store, whose constructor probes the bucket.
+        task_manifest("validate", args, {key: f"https://{self.host}/" for key in ("SOURCE_URL", "QUESTIONS_URL", "RESULT_URL")})
+        self.store = store if store is not None else open_store()
+        if store is None and not isinstance(self.store, S3Store):
+            raise ValueError("AX requires OUSAST_MEMORY=s3://bucket[/prefix]")
+
+    def ax(self, *args, manifest=None, timeout=30):
+        env = dict(os.environ)
+        if self.args.kubeconfig:
+            env["KUBECONFIG"] = str(Path(self.args.kubeconfig).expanduser())
+        done = subprocess.run(
+            [str(Path(self.args.ax_bin).expanduser()), *args, *(["--atespace", self.args.atespace] if args[0] != "apply" else [])],
+            input=json.dumps(manifest) if manifest is not None else None,
+            env=env,
+            capture_output=True,
+            text=True,
+            timeout=max(0.01, timeout),
+        )
+        if done.returncode:
+            # AX errors may echo manifest URLs; report only verb and exit code.
+            raise ValueError(f"ax {args[0]} failed (exit {done.returncode})")
+        return done
+
+    def receive(self, data, output, pin):
+        # Validate in isolation: malformed results must never become usable checkpoints.
+        with tempfile.TemporaryDirectory(prefix="engine-ax-result-") as temporary:
+            root = Path(temporary)
+            with tarfile.open(fileobj=io.BytesIO(data)) as archive:
+                archive.extractall(root, filter="data")
+            record = json.loads((root / "result.json").read_text())
+            units = record.get("units", [])
+            if record.get("done") and record.get("status") == "failed" and not units:
+                # Questions download can fail before the actor learns any unit identities.
+                # Keep that explicit failure, filling identities only from the host input.
+                record = {
+                    **pin,
+                    **record,
+                    "units": [
+                        {
+                            **unit,
+                            "status": "failed" if unit["supported"] else "unsupported",
+                            "reason": record.get("reason", "AX task entry failed"),
+                            "instrument": {"files": [], "bytes": 0, "cpg_path": None, "cpg_bytes": 0, "jvm": []},
+                            "questions": {},
+                            "questions_asked": [],
+                            "questions_completed": [],
+                            "witness_rows": [],
+                            "traces": [],
+                        }
+                        for unit in pin["units"]
+                    ],
+                }
+                units = record["units"]
+                (root / "result.json").write_text(json.dumps(record))
+            expected = [u["unit"] for u in pin["units"]]
+            actual = [u["unit"] for u in units]
+            if not record.get("done") or sorted(actual) != sorted(expected):
+                raise ValueError("AX Task returned incomplete or mismatched units")
+            output.mkdir(parents=True, exist_ok=True)
+            # Keep the artifacts without interpreting any remote paths as host destinations.
+            shutil.copytree(root, output, dirs_exist_ok=True)
+
+    def execute(self, checkout, output, pin, pin_id):
+        attempts = []
+        try:
+            for attempt in range(2):
+                try:
+                    return self._execute_attempt(checkout, output, pin, pin_id, attempt, attempts)
+                except SandboxFailure:
+                    if attempt == 0:
+                        self.pause(10)
+            return subprocess.CompletedProcess([], 1, "", "AX sandbox failed twice"), None
+        finally:
+            (output / "ax_attempts.json").write_text(json.dumps(attempts))
+            result_path = output / "result.json"
+            if result_path.exists():
+                record = json.loads(result_path.read_text())
+                record["ax_attempts"] = attempts
+                result_path.write_text(json.dumps(record))
+
+    def _execute_attempt(self, checkout, output, pin, pin_id, attempt, attempts):
+        digest = hashlib.sha256(f"{self.run}/{pin_id}/attempt-{attempt}".encode()).hexdigest()[:12]
+        slug = re.sub(r"[^a-z0-9]+", "-", str(pin_id).lower()).strip("-")[:22]
+        name = f"ousast-engine-{slug or 'pin'}-{digest}"
+        key_id = hashlib.sha256(str(pin_id).encode()).hexdigest()
+        prefix = f"engine-queue/{self.run}/{key_id}/attempt-{attempt}/"
+        keys = [prefix + suffix for suffix in (f"source{PRESIGNED_ARCHIVE_SUFFIX}", "questions.json", f"result{PRESIGNED_ARCHIVE_SUFFIX}")]
+        expires = timedelta(seconds=self.args.deadline + 600)
+        worker_ip = None
+        submitted = False
+        started = self.clock()
+        phase = ""
+        try:
+            self.store._put(keys[0], pack(checkout))
+            self.store._put(keys[1], json.dumps(pin).encode())
+            urls = {
+                "SOURCE_URL": self.store.presign_get(keys[0], expires),
+                "QUESTIONS_URL": self.store.presign_get(keys[1], expires),
+                "RESULT_URL": self.store.presign_put(keys[2], expires),
+            }
+            if any(public_endpoint(url) != self.host for url in urls.values()):
+                raise EgressError("AX presign hostname must match S3_ENDPOINT for the store egress policy")
+            manifest = task_manifest(name, self.args, urls)
+            end = self.clock() + self.args.deadline
+            # An apply timeout can still have created the task; cleanup must cover it.
+            submitted = True
+            self.ax("apply", "-f", "-", manifest=manifest, timeout=min(30, self.args.deadline))
+            while self.clock() < end:
+                result = bounded_read(lambda: self.store._get(keys[2]), end - self.clock())
+                if self.clock() >= end:
+                    return None, worker_ip
+                phase = ""
+                try:
+                    state = self.ax("get", "tasks", timeout=min(30, end - self.clock())).stdout
+                    phase, ip = task_state(state, name)
+                    worker_ip = ip or worker_ip
+                except (ValueError, OSError, subprocess.SubprocessError):
+                    # The object is authoritative even when AX status is temporarily unavailable.
+                    pass
+                if self.clock() >= end:
+                    return None, worker_ip
+                if result is not None:
+                    self.receive(result[0], output, pin)
+                    return subprocess.CompletedProcess([], 0, "", ""), worker_ip
+                if phase.lower() in {"failed", "missing"}:
+                    # The worker may have uploaded while we were reading AX status.
+                    result = bounded_read(lambda: self.store._get(keys[2]), end - self.clock())
+                    if self.clock() >= end:
+                        return None, worker_ip
+                    if result is not None:
+                        self.receive(result[0], output, pin)
+                        return subprocess.CompletedProcess([], 0, "", ""), worker_ip
+                    raise SandboxFailure
+                self.pause(min(2, max(0, end - self.clock())))
+            return None, worker_ip
+        except (subprocess.TimeoutExpired, TimeoutError):
+            return None, worker_ip
+        except SandboxFailure:
+            raise
+        except Exception as exc:
+            # Never expose exception strings which may contain object-scoped bearer URLs.
+            detail = str(exc) if isinstance(exc, EgressError) else type(exc).__name__
+            return subprocess.CompletedProcess([], 1, "", "AX transport failed: " + detail), worker_ip
+        finally:
+            attempts.append({"name": name, "phase": phase, "seconds": self.clock() - started})
+            errors = []
+            if submitted:
+                try:
+                    self.ax("delete", "task", name)
+                except Exception as exc:
+                    # A successful list already confirmed a disappeared task is gone.
+                    if phase != "Missing":
+                        errors.append("task: " + type(exc).__name__)
+            for key in keys:
+                try:
+                    self.store._delete(key)
+                except Exception as exc:
+                    errors.append("object: " + type(exc).__name__)
+            if errors:
+                raise RuntimeError("AX cleanup failed; stop dispatching: " + ", ".join(errors))

benchmarks/learn/engine_trace_k8s.py

@@ -0,0 +1,259 @@
+"""Host Kubernetes transport; the embedded bootstrap uses only the standard library."""
+
+from __future__ import annotations
+
+import hashlib
+import io
+import json
+import re
+import subprocess
+import tarfile
+import time
+import uuid
+from datetime import timedelta
+
+from openultrasast.plane.memory import PRESIGNED_ARCHIVE_SUFFIX, S3Store, open_store
+
+IMAGE = "ghcr.io/norandom/openultrasast:2.0.1"
+BOOTSTRAP = r"""
+import io, json, os, pathlib, subprocess, sys, tarfile, urllib.request
+
+def get(name):
+    with urllib.request.urlopen(os.environ[name], timeout=120) as response:
+        return response.read()
+
+def put(data):
+    request = urllib.request.Request(os.environ['RESULT_URL'], data=data, method='PUT')
+    with urllib.request.urlopen(request, timeout=120) as response:
+        response.read()
+
+try:
+    if os.environ.get('CHECK') == '1':
+        data = get('SOURCE_URL')
+        assert len(data) == 1024
+        subprocess.run([sys.executable, '-c', 'print(42)'], check=True)
+        put(data)
+    else:
+        for name, target in [('SOURCE_URL', '/case'), ('ANALYZER_URL', '/frozen')]:
+            pathlib.Path(target).mkdir(parents=True, exist_ok=True)
+            with tarfile.open(fileobj=io.BytesIO(get(name))) as archive:
+                archive.extractall(target, filter='data')
+        out = pathlib.Path('/out')
+        out.mkdir(exist_ok=True)
+        (out / 'pin.json').write_text(os.environ['PIN'])
+        env = dict(os.environ, PYTHONPATH='/frozen/src')
+        for key in ['SOURCE_URL', 'ANALYZER_URL', 'RESULT_URL']:
+            env.pop(key, None)
+        with (out / 'container.log').open('w') as log:
+            done = subprocess.run([sys.executable, '/frozen/benchmarks/learn/engine_trace_worker.py',
+                '--worker', '/out/pin.json', '--deadline', os.environ['DEADLINE'],
+                '--question-deadline', os.environ['QUESTION_DEADLINE']], env=env, stdout=log, stderr=log)
+        (out / 'exit.json').write_text(json.dumps(done.returncode))
+        buffer = io.BytesIO()
+        with tarfile.open(fileobj=buffer, mode='w') as archive:
+            for path in out.iterdir():
+                archive.add(path, arcname=path.name)
+        put(buffer.getvalue())
+        sys.exit(done.returncode)
+except Exception as exc:
+    # urllib exceptions can contain bearer URLs. Never print exception text.
+    print('engine bootstrap failed: ' + type(exc).__name__, file=sys.stderr)
+    sys.exit(1)
+"""
+
+
+def pack(path):
+    buffer = io.BytesIO()
+    with tarfile.open(fileobj=buffer, mode="w") as archive:
+        for child in sorted(path.iterdir()):
+            archive.add(child, arcname=child.name)
+    return buffer.getvalue()
+
+
+def manifests(name, run, args, urls, pin, check=False):
+    labels = {"engine-trace-run": run, "engine-trace-job": name, "engine-trace": "true"}
+    secret = {
+        "apiVersion": "v1",
+        "kind": "Secret",
+        "metadata": {"name": name, "namespace": args.kube_namespace, "labels": labels},
+        "type": "Opaque",
+        "stringData": urls,
+    }
+    env = [{"name": key, "valueFrom": {"secretKeyRef": {"name": name, "key": key}}} for key in urls]
+    env += [
+        {"name": key, "value": value}
+        for key, value in {
+            "PIN": json.dumps(pin),
+            "DEADLINE": str(args.deadline),
+            "QUESTION_DEADLINE": str(args.question_deadline),
+            "CHECK": "1" if check else "0",
+        }.items()
+    ]
+    job = {
+        "apiVersion": "batch/v1",
+        "kind": "Job",
+        "metadata": {"name": name, "namespace": args.kube_namespace, "labels": labels},
+        "spec": {
+            "activeDeadlineSeconds": args.deadline,
+            "backoffLimit": 0,
+            "ttlSecondsAfterFinished": 86400,
+            "template": {
+                "metadata": {"labels": labels},
+                "spec": {
+                    "restartPolicy": "Never",
+                    "automountServiceAccountToken": False,
+                    "securityContext": {"runAsUser": 1000, "runAsGroup": 1000, "fsGroup": 1000},
+                    "volumes": [{"name": name, "emptyDir": {}} for name in ("case", "frozen", "out")],
+                    "affinity": {
+                        "podAntiAffinity": {
+                            "requiredDuringSchedulingIgnoredDuringExecution": [
+                                {
+                                    "labelSelector": {"matchLabels": {"engine-trace": "true"}},
+                                    "topologyKey": "kubernetes.io/hostname",
+                                }
+                            ]
+                        }
+                    },
+                    "containers": [
+                        {
+                            "name": "engine",
+                            "image": args.image,
+                            "command": ["python", "-c", BOOTSTRAP],
+                            "env": env,
+                            "volumeMounts": [{"name": name, "mountPath": "/" + name} for name in ("case", "frozen", "out")],
+                            "resources": {"requests": {"memory": "3Gi", "cpu": "1"}, "limits": {"memory": "3Gi"}},
+                        }
+                    ],
+                },
+            },
+        },
+    }
+    return job, secret
+
+
+def redact(text):
+    return re.sub(r"https?://[^\s\"'<>]+", "[redacted URL]", text)
+
+
+class Kubernetes:
+    def __init__(self, args, store=None):
+        self.args = args
+        self.run = uuid.uuid4().hex
+        self.store = store if store is not None else open_store()
+        if not isinstance(self.store, S3Store):
+            raise ValueError("Kubernetes requires OUSAST_MEMORY=s3://bucket[/prefix]")
+
+    def kubectl(self, *args, manifest=None, check=True):
+        result = subprocess.run(
+            ["kubectl", "--context", self.args.kube_context, "--namespace", self.args.kube_namespace, *args],
+            input=json.dumps(manifest) if manifest is not None else None,
+            capture_output=True,
+            text=True,
+            timeout=30,
+        )
+        if check and result.returncode:
+            # kubectl can echo submitted Secret data on errors; never expose stderr.
+            raise ValueError(f"kubectl {args[0]} failed (exit {result.returncode})")
+        return result
+
+    def namespace(self):
+        self.kubectl("get", "nodes", "-o", "json")
+        self.kubectl("apply", "-f", "-", manifest={"apiVersion": "v1", "kind": "Namespace", "metadata": {"name": self.args.kube_namespace}})
+
+    def execute(self, source, checkout, output, pin, check=False):
+        name = "engine-" + uuid.uuid4().hex[:24]
+        pin_key = hashlib.sha256(json.dumps([pin.get("repo"), pin.get("pin")]).encode()).hexdigest()
+        prefix = f"engine-jobs/{self.run}/{pin_key}/"
+        keys = [prefix + name + PRESIGNED_ARCHIVE_SUFFIX for name in ("source", "analyzer", "result")]
+        expiry = timedelta(seconds=self.args.deadline + 900)
+        node = None
+        try:
+            self.store._put(keys[0], b"x" * 1024 if check else pack(checkout))
+            self.store._put(keys[1], b"" if check else pack(source))
+            urls = {
+                "SOURCE_URL": self.store.presign_get(keys[0], expiry),
+                "ANALYZER_URL": self.store.presign_get(keys[1], expiry),
+                "RESULT_URL": self.store.presign_put(keys[2], expiry),
+            }
+            job, secret = manifests(name, self.run, self.args, urls, pin, check)
+            # create avoids kubectl's last-applied annotation duplicating Secret values.
+            created = json.loads(self.kubectl("create", "-f", "-", "-o", "json", manifest=job).stdout)
+            secret["metadata"]["ownerReferences"] = [
+                {
+                    "apiVersion": "batch/v1",
+                    "kind": "Job",
+                    "name": name,
+                    "uid": created["metadata"]["uid"],
+                }
+            ]
+            self.kubectl("create", "-f", "-", manifest=secret)
+            end = time.monotonic() + self.args.deadline + 60
+            reason = "host deadline waiting for Job"
+            while time.monotonic() < end:
+                state = json.loads(self.kubectl("get", "job", name, "-o", "json").stdout).get("status", {})
+                pods = json.loads(self.kubectl("get", "pods", "-l", f"engine-trace-job={name}", "-o", "json").stdout)
+                node = next((p.get("spec", {}).get("nodeName") for p in pods.get("items", []) if p.get("spec", {}).get("nodeName")), node)
+                result = self.store._get(keys[2])
+                failed = next((c for c in state.get("conditions", []) if c.get("type") == "Failed" and c.get("status") == "True"), None)
+                if failed or state.get("failed"):
+                    reason = (failed or {}).get("reason", "Job failed") + ": " + (failed or {}).get("message", "")
+                    break
+                if state.get("succeeded") and result is not None:
+                    if check:
+                        if result[0] != b"x" * 1024:
+                            raise ValueError("round-trip bytes mismatch")
+                    else:
+                        with tarfile.open(fileobj=io.BytesIO(result[0])) as archive:
+                            archive.extractall(output, filter="data")
+                        if not (output / "result.json").is_file():
+                            raise ValueError("Job produced no result.json")
+                        record = json.loads((output / "result.json").read_text())
+                        if not record.get("done") or {u["unit"] for u in record["units"]} != {u["unit"] for u in pin["units"]}:
+                            raise ValueError("Job returned incomplete or mismatched units")
+                    return subprocess.CompletedProcess([], 0, "", ""), node
+                time.sleep(2)
+            logs = self.kubectl("logs", f"job/{name}", "--tail=40", check=False)
+            reason = redact(reason + "\n" + logs.stdout + logs.stderr)
+            return subprocess.CompletedProcess([], 1, "", reason), node
+        except Exception as exc:
+            reason = "Kubernetes transport failed: " + type(exc).__name__ + ": " + redact(str(exc))
+            try:
+                logs = self.kubectl("logs", f"job/{name}", "--tail=40", check=False)
+                reason += "\n" + redact(logs.stdout + logs.stderr)
+            except (OSError, ValueError, subprocess.SubprocessError):
+                pass
+            return subprocess.CompletedProcess([], 1, "", reason), node
+        finally:
+            if not self.args.keep_jobs:
+                # Attempt every cleanup even if an earlier deletion fails.
+                errors = []
+                for kind in ("job", "secret"):
+                    try:
+                        self.kubectl("delete", kind, name, "--ignore-not-found", "--wait=true", "--timeout=30s")
+                    except Exception as exc:
+                        errors.append(type(exc).__name__)
+                for key in keys:
+                    try:
+                        self.store._delete(key)
+                    except Exception as exc:
+                        errors.append(type(exc).__name__)
+                if errors:
+                    raise RuntimeError("engine Job cleanup failed: " + ", ".join(errors))
+
+    def check(self):
+        self.namespace()
+        nodes = json.loads(self.kubectl("get", "nodes", "-o", "json").stdout)
+        for node in nodes["items"]:
+            print(json.dumps({"node": node["metadata"]["name"], "allocatable_memory": node["status"]["allocatable"]["memory"]}))
+        done, node = self.execute(None, None, None, {}, check=True)
+        print(
+            json.dumps(
+                {
+                    "k8s_check": "passed" if done.returncode == 0 else "failed",
+                    "node": node,
+                    "reason": done.stderr,
+                    "round_trip_bytes": 1024 if done.returncode == 0 else 0,
+                }
+            )
+        )
+        return done.returncode

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))},
+    }

benchmarks/learn/engine_trace_queue.py

@@ -0,0 +1,52 @@
+"""Host queue transport: source and task metadata only, never analyzer code."""
+
+from __future__ import annotations
+
+import json
+import subprocess
+import time
+import uuid
+
+from engine_worker_service import PENDING, client_from_env, pack, put, read, result_key, status
+
+from openultrasast.plane.memory import PRESIGNED_ARCHIVE_SUFFIX
+
+
+class Dispatcher:
+    def __init__(self, args, client=None, *, pause=time.sleep, clock=time.monotonic):
+        self.client = client if client is not None else client_from_env()
+        self.run = uuid.uuid4().hex
+        self.args, self.pause, self.clock = args, pause, clock
+
+    def execute(self, checkout, output, pin, pin_id):
+        task_id = f"{self.run}-{pin_id}"
+        task = dict(
+            run=self.run,
+            id=pin_id,
+            task_id=task_id,
+            pin=pin,
+            input=f"engine-queue/inputs/{self.run}/{pin_id}/source{PRESIGNED_ARCHIVE_SUFFIX}",
+            image=self.args.image,
+            deadline=self.args.deadline,
+            question_deadline=self.args.question_deadline,
+        )
+        try:
+            self.client.put(task["input"], pack(checkout), {})
+            put(self.client, PENDING + task_id, task)
+            end = self.clock() + self.args.queue_timeout
+            while self.clock() < end:
+                result = read(self.client, result_key(task))
+                if result is not None:
+                    (output / "result.json").write_text(json.dumps(result))
+                    self.client.delete(task["input"])
+                    return subprocess.CompletedProcess([], 0, "", ""), result.get("worker")
+                self.pause(2)
+            # Removing pending cancels work that has not been claimed; retain source for a live lease.
+            self.client.delete(PENDING + task_id)
+            return None, None
+        except Exception as exc:
+            # SDK errors can contain endpoint URLs or credentials. Do not persist their text.
+            raise OSError("queue transport failed") from exc
+
+    def status(self):
+        return status(self.client)

benchmarks/learn/engine_trace_worker.py

@@ -0,0 +1,644 @@
+"""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, extract_payload
+from openultrasast.cpg.session import EngineSession
+from openultrasast.model.contracts import ExecutionBudget
+
+
+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
+CLUSTER_JVM_FLAGS = (
+    "-Xms1700m -Xmx1700m -XX:MaxMetaspaceSize=256m -XX:ReservedCodeCacheSize=128m "
+    "-XX:MaxDirectMemorySize=256m -Xss512k -XX:ActiveProcessorCount=2 -XX:+UseG1GC"
+)
+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" or (stage["command"] == "joern-session" and stage.get("step") == "start" and stage.get("success"))
+        ) 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 process and resident-session watchdogs."""
+
+    def __init__(self, deadline, question_deadline, checkpoint, heap_profile="vm"):
+        if heap_profile not in {"vm", "cluster"}:
+            raise ValueError(f"unknown heap profile: {heap_profile}")
+        self.heap_profile = heap_profile
+        super().__init__(
+            build_timeout=deadline,
+            query_timeout=deadline,
+            session_transport=False,
+            heap_mb=1700 if heap_profile == "cluster" else 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 = []
+        self.session_fallback = ""
+
+    def _jvm_env(self):
+        env = super()._jvm_env()
+        if self.heap_profile == "cluster":
+            # Bound every JVM, including the child spawned by the Joern launcher.
+            env["JAVA_TOOL_OPTIONS"] = CLUSTER_JVM_FLAGS
+            env.pop("_JAVA_OPTIONS", None)
+            env.pop("JDK_JAVA_OPTIONS", None)
+        return env
+
+    def _session_stage(self, name, action):
+        started = time.monotonic()
+        result = None
+        try:
+            result = action()
+            return result
+        finally:
+            self.stages.append(
+                {
+                    "command": "joern-session",
+                    "step": name,
+                    "success": result is not None and result is not False,
+                    "seconds": time.monotonic() - started,
+                    "heap_mb": self._heap_mb(),
+                    "java_tool_options": self._jvm_env()["JAVA_TOOL_OPTIONS"],
+                }
+            )
+            self.checkpoint()
+
+    def _fallback_session(self, reason):
+        self.session_fallback = reason
+        if self._session is not None:
+            detail = self._session.log_tail()
+            if "OutOfMemoryError" in detail:
+                self.fatal = "JVM out of memory"
+                self.stages[-1]["oom"] = True
+            self.stages[-1]["output_tail"] = detail
+        self.close_session()
+        self.checkpoint()
+
+    def _session_for(self, cpg_path):
+        if self.session_fallback or self.fatal or self.timed_out:
+            return None
+        try:
+            if self._session is None:
+                self._session = EngineSession(
+                    cpg_path.parent / "engine-session",
+                    ExecutionBudget(self.deadline, 5),
+                    self._jvm_env(),
+                    startup_allowance_seconds=min(180, max(20, (self.deadline - time.monotonic()) / 3)),
+                )
+                if not self._session_stage("start", self._session.start):
+                    self._fallback_session(self._session.failure or "session_start_failed")
+                    return None
+            if self._session.loaded_graph != str(cpg_path):
+                if not self._session_stage("load", lambda: self._session.load(cpg_path, timeout=self.query_timeout)):
+                    self._fallback_session(self._session.failure or "session_load_failed")
+                    return None
+                self._defined.clear()
+            return self._session
+        except (OSError, RuntimeError) as exc:
+            self._fallback_session(f"session_start_or_load:{exc}")
+            return None
+
+    def _session_payload(self, cpg_path, query, params):
+        session = self._session_for(cpg_path)
+        if session is None:
+            return None
+        end = min(self.deadline, time.monotonic() + self.query_timeout)
+        question_end = end
+        prior_asked = list(self.asked)
+        prior_answers = dict(self.answers)
+        prior_counts = self.completion_counts.copy()
+        consumed = 0
+        streamed = ""
+
+        def observe(output):
+            nonlocal consumed, question_end, streamed
+            streamed = output
+            lines = output[consumed:].splitlines(keepends=True)
+            for line in lines:
+                if not line.endswith("\n"):
+                    break
+                consumed += len(line)
+                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 time.monotonic() >= min(end, question_end):
+                self.timed_out = True
+                return False
+            return True
+
+        def run():
+            source = (self.queries_dir / f"{query}.sc").read_text()
+            # Same shipped script/parameters as JoernBackend, using the graph loaded above.
+            if source.count("importCpg(cpgFile)") != 1:
+                return None
+            symbol = f"ousast_{query}"
+            if query not in self._defined:
+                source = source.replace("importCpg(cpgFile)", "()", 1)
+                source = source.replace("@main def exec(", f"def {symbol}(", 1)
+                if session.define(source, timeout=max(0, end - time.monotonic())) is None:
+                    return None
+                self._defined.add(query)
+            arguments = ", ".join(f"{name} = {json.dumps(_render(value))}" for name, value in sorted(params.items()))
+            session.output_observer = observe
+            try:
+                answer = session.evaluate(f"{symbol}({arguments})", timeout=max(0, end - time.monotonic()))
+            finally:
+                session.output_observer = None
+            if answer is not None:
+                observe(answer.stdout)
+                return answer.stdout
+            return None
+
+        failure = ""
+        try:
+            body = self._session_stage(query, run)
+        except (OSError, RuntimeError) as exc:
+            body = None
+            failure = f"session_{query}:{exc}"
+        if self.timed_out:
+            self.close_session()
+            return streamed
+        parsed = extract_payload(body) if body is not None else None
+        if not isinstance(parsed, dict):
+            # Discard markers from an unverified attempt before replaying the script.
+            self.asked[:] = prior_asked
+            self.answers = prior_answers
+            self.completion_counts = prior_counts
+            self._fallback_session(failure or session.failure or f"session_{query}_payload_missing")
+            return None
+        if query == "census":
+            self.stages[-1]["files"] = int(parsed.get("files", 0))
+            if self.stages[-1]["files"] <= 0:
+                self.fatal = "query census read zero files"
+        return body
+
+    def _apply_overlays(self, cpg_path, scratch):
+        session = self._session_for(cpg_path)
+        result = False
+        if session is not None:
+            body = self._session_payload(cpg_path, "overlay", {"cpgFile": str(cpg_path)})
+            saved = session.scratch / "workspace" / cpg_path.name / "cpg.bin"
+            if body is not None and saved.is_file():
+                import shutil
+
+                shutil.copyfile(saved, cpg_path)
+                result = True
+            elif body is not None:
+                self._fallback_session("session_overlay_saved_graph_missing")
+        if not result:
+            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, heap_profile="vm"):
+    proof = {
+        "files": [],
+        "bytes": 0,
+        "cpg_path": None,
+        "cpg_bytes": 0,
+        "jvm": [],
+        "heap_mb": 1700 if heap_profile == "cluster" else HEAP_MB,
+        "heap_profile": heap_profile,
+        "java_tool_options": CLUSTER_JVM_FLAGS
+        if heap_profile == "cluster"
+        else f"{os.environ.get('JAVA_TOOL_OPTIONS', '').strip()} -Xmx{HEAP_MB}m".strip(),
+        "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"]
+        if getattr(backend, "session_fallback", ""):
+            proof["session_fallback"] = backend.session_fallback
+        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, **({"heap_profile": heap_profile} if heap_profile != "vm" else {}))
+    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 hasattr(backend, "close_session"):
+            backend.close_session()
+        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, heap_profile="vm"):
+    """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, heap_profile=heap_profile)
+    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, heap_profile=heap_profile
+        )
+    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)
+    parser.add_argument("--heap-profile", choices=("cluster", "vm"), default="vm")
+    parser.add_argument("--root", type=Path, default=Path("/case"))
+    parser.add_argument("--output", type=Path, default=Path("/out/result.json"))
+    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()), args.root, args.output, args.deadline, args.question_deadline, args.heap_profile)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

benchmarks/learn/engine_worker_service.py

@@ -0,0 +1,315 @@
+"""Image-resident queue workers. Requires strongly consistent S3 GET/PUT/LIST.
+
+Per-task Lamport bakery tickets use only ordinary object operations. A choosing
+marker precedes ticket selection; contenders wait for choosing peers and lower
+(ticket, worker) pairs. Late arrivals cannot displace a running owner. Leases
+assume bounded clock skew (less than MARGIN / 2) and a worker killed by its deadline.
+"""
+
+from __future__ import annotations
+
+import argparse
+import contextlib
+import ctypes
+import io
+import json
+import logging
+import multiprocessing
+import os
+import signal
+import socket
+import tarfile
+import tempfile
+import threading
+import time
+import uuid
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+
+from engine_trace_worker import unit_record, worker
+
+from openultrasast.plane.memory import S3Client, s3_settings
+
+PENDING = "engine-queue/pending/"
+CLAIMED = "engine-queue/claimed/"
+RESULTS = "engine-results/"
+MARGIN = 60
+LOG = logging.getLogger(__name__)
+
+
+def client_from_env():
+    settings = s3_settings(os.environ)
+    bucket = os.environ.get("S3_BUCKET")
+    if not bucket:
+        raise ValueError("S3_BUCKET is required")
+    return S3Client(bucket=bucket, **settings)
+
+
+def read(client, key):
+    value = client.get(key)
+    return json.loads(value[0]) if value else None
+
+
+def put(client, key, value):
+    client.put(key, json.dumps(value).encode(), {})
+
+
+def result_key(task):
+    return f"{RESULTS}{task['run']}/{task['id']}.json"
+
+
+def pack(root):
+    stream = io.BytesIO()
+    with tarfile.open(fileobj=stream, mode="w") as archive:
+        archive.add(root, arcname=".")
+    return stream.getvalue()
+
+
+def unpack(data, root):
+    with tarfile.open(fileobj=io.BytesIO(data)) as archive:
+        # Source snapshots need only regular files and directories, never links/devices.
+        for member in archive.getmembers():
+            target = (root / member.name).resolve()
+            if not target.is_relative_to(root.resolve()) or not (member.isfile() or member.isdir()):
+                raise ValueError("unsafe source archive")
+        archive.extractall(root, filter="data")
+
+
+class Queue:
+    def __init__(self, client, identity=None, *, now=time.time, pause=time.sleep, stop=None):
+        self.client = client
+        self.identity = identity or f"{socket.gethostname()}-{uuid.uuid4().hex}"
+        self.now, self.pause = now, pause
+        self.stop = stop or threading.Event()
+
+    def markers(self, task_id=None):
+        result = []
+        for key in self.client.keys(CLAIMED):
+            if task_id is None or key.rsplit("/", 1)[-1] == task_id:
+                marker = read(self.client, key)
+                if marker:
+                    result.append((key, marker))
+        return result
+
+    def tasks(self):
+        tasks = {}
+        for key in self.client.keys(PENDING):
+            task = read(self.client, key)
+            if task:
+                tasks[task["task_id"]] = task
+        # A dead worker may have removed pending already. Its task remains in the lease.
+        for _, marker in self.markers():
+            if marker["expires"] < self.now():
+                task = marker["task"]
+                tasks.setdefault(task["task_id"], task)
+        return list(tasks.values())
+
+    def claim(self, task):
+        if read(self.client, result_key(task)) is not None or self.stop.is_set():
+            return None
+        key = f"{CLAIMED}{self.identity}/{task['task_id']}"
+        marker = dict(task=task, ticket=0, expires=self.now() + task["deadline"] + MARGIN)
+        put(self.client, key, marker)
+        try:
+            live = [m for _, m in self.markers(task["task_id"]) if m["expires"] > self.now()]
+            marker["ticket"] = 1 + max((m["ticket"] for m in live), default=0)
+            put(self.client, key, marker)
+            while not self.stop.is_set():
+                # Do not start once the execution budget no longer fits within our lease.
+                if self.now() >= marker["expires"] - task["deadline"] - MARGIN / 2:
+                    break
+                peers = [(k, m) for k, m in self.markers(task["task_id"]) if k != key and m["expires"] > self.now()]
+                if any(m["ticket"] == 0 for _, m in peers):
+                    self.pause(0.1)
+                    continue
+                if any((m["ticket"], k) < (marker["ticket"], key) for k, m in peers):
+                    break  # loser backs off and can try another pending task
+                if read(self.client, key) != marker or read(self.client, result_key(task)) is not None:
+                    break
+                self.client.delete(PENDING + task["task_id"])
+                return key
+        except Exception:
+            self.client.delete(key)
+            raise
+        self.client.delete(key)
+        return None
+
+    def process_one(self, execute=None):
+        for task in self.tasks():
+            key = self.claim(task)
+            if key is None:
+                continue
+            try:
+                lease = read(self.client, key)
+                if lease is None:
+                    continue
+                budget = min(task["deadline"], lease["expires"] - self.now() - MARGIN / 2)
+                if budget <= 0:
+                    continue  # Leave the expired lease's task available to recovery.
+                result = (execute or execute_task)(self.client, {**task, "runtime_budget": budget})
+                result.update(executor="queue", worker=self.identity, pod=socket.gethostname(), image=os.environ.get("ENGINE_IMAGE", ""))
+                put(self.client, result_key(task), result)
+                # Publish the result before removing either the input or the claim.
+                self.client.delete(task["input"])
+                for old_key, old in self.markers(task["task_id"]):
+                    if old_key == key or old["expires"] < self.now():
+                        self.client.delete(old_key)
+                return True
+            except Exception:
+                # Keep the lease and task for recovery, without logging credential-bearing errors.
+                LOG.error("task failed; lease retained for recovery")
+                return False
+        return False
+
+
+def _child(pin, root, out, deadline, question_deadline):
+    os.setsid()
+    signal.signal(signal.SIGTERM, signal.SIG_DFL)
+    worker(pin, root, out, deadline, question_deadline)
+
+
+def adopt_orphans():
+    # Linux PR_SET_CHILD_SUBREAPER: escaped Joern sessions reparent here, not
+    # outside the service. There is exactly one task process tree per worker.
+    if ctypes.CDLL(None, use_errno=True).prctl(36, 1, 0, 0, 0) != 0:
+        raise OSError("cannot enable task descendant cleanup")
+
+
+def kill_children():
+    """Freeze parents before walking children, including new-session JVMs."""
+    seen = set()
+
+    def freeze(parent):
+        for entry in Path("/proc").iterdir():
+            if not entry.name.isdigit() or int(entry.name) in seen:
+                continue
+            try:
+                # comm can contain spaces or parentheses. Fields after it begin at state.
+                fields = (entry / "stat").read_text().rsplit(")", 1)[1].split()
+                if int(fields[1]) != parent:
+                    continue
+                pid = int(entry.name)
+                os.kill(pid, signal.SIGSTOP)
+                seen.add(pid)
+                freeze(pid)
+            except (FileNotFoundError, ProcessLookupError, PermissionError):
+                continue
+
+    # A child that exited during discovery can reparent descendants to this
+    # subreaper. Repeat until every remaining descendant is frozen.
+    while True:
+        count = len(seen)
+        freeze(os.getpid())
+        if len(seen) == count:
+            break
+    for pid in seen:
+        with contextlib.suppress(ProcessLookupError):
+            os.kill(pid, signal.SIGKILL)
+
+
+def execute_task(client, task):
+    adopt_orphans()
+    pin = task["pin"]
+    started = time.monotonic()
+    budget = task.get("runtime_budget", task["deadline"])
+    status, reason, code = "failed", "worker produced no result", None
+    with tempfile.TemporaryDirectory(prefix="engine-task-") as scratch:
+        root, out = Path(scratch) / "case", Path(scratch) / "result.json"
+        root.mkdir()
+        try:
+            if task["deadline"] > 900 or task["deadline"] <= 0:
+                raise ValueError("deadline exceeds deployment grace")
+            if task.get("image") != os.environ.get("ENGINE_IMAGE"):
+                raise ValueError("task image differs from deployed image")
+            source = client.get(task["input"])
+            if source is None:
+                raise ValueError("missing source")
+            unpack(source[0], root)
+            # The shipped worker also checks frontend input and graph output for every language.
+            files = [p for p in root.rglob("*") if p.is_file()]
+            if not files:
+                raise ValueError("empty source")
+            LOG.info("input proof: first file bytes=%d", len(files[0].read_bytes()))
+            if time.monotonic() - started >= budget:
+                status, reason = "timeout", "task input deadline"
+                raise TimeoutError(reason)
+            process = multiprocessing.get_context("fork").Process(target=_child, args=(pin, root, out, budget, task["question_deadline"]))
+            process.start()
+            try:
+                process.join(max(0, budget - (time.monotonic() - started)))
+                if process.is_alive():
+                    status, reason = "timeout", "task deadline"
+                else:
+                    reason = f"worker exit {process.exitcode}"
+                code = process.exitcode
+            finally:
+                # Joern creates separate sessions. Kill the whole descendant tree,
+                # not merely the Python wrapper's process group.
+                kill_children()
+                process.join()
+                # Reap adopted JVMs and their launchers after the direct child.
+                while True:
+                    try:
+                        os.waitpid(-1, 0)
+                    except ChildProcessError:
+                        break
+            record = json.loads(out.read_text()) if out.exists() else {**pin, "units": []}
+        except Exception:
+            record = {**pin, "units": []}
+            reason = "worker input or execution failed"
+        if not record.get("done") or code != 0:
+            previous = {u["unit"]: u for u in record["units"]}
+            record["units"] = [
+                previous[u["unit"]]
+                if code is None and previous.get(u["unit"], {}).get("status") in {"path", "asked-nothing"}
+                else unit_record(u, status if u["supported"] else "unsupported", reason if u["supported"] else u["reason"])
+                for u in pin["units"]
+            ]
+        record.update(done=True, container_exit=code, seconds=time.monotonic() - started)
+        return record
+
+
+def status(client):
+    return {
+        "pending": len(client.keys(PENDING)),
+        "claimed": len(client.keys(CLAIMED)),
+        "done": len([k for k in client.keys(RESULTS) if k.endswith(".json")]),
+    }
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--port", type=int, default=8080)
+    args = parser.parse_args()
+    logging.basicConfig(level=logging.INFO)
+    stop = threading.Event()
+    signal.signal(signal.SIGTERM, lambda *_: stop.set())
+    signal.signal(signal.SIGINT, lambda *_: stop.set())
+    client = client_from_env()
+    queue = Queue(client, stop=stop)
+
+    class Health(BaseHTTPRequestHandler):
+        def do_GET(self):
+            self.send_response(200 if self.path == "/healthz" and not stop.is_set() else 503)
+            self.end_headers()
+
+        def log_message(self, *_):
+            pass
+
+    server = ThreadingHTTPServer(("0.0.0.0", args.port), Health)
+    threading.Thread(target=server.serve_forever, daemon=True).start()
+    try:
+        while not stop.is_set():
+            try:
+                if not queue.process_one():
+                    stop.wait(2)
+            except Exception:
+                LOG.error("queue unavailable; retrying")
+                stop.wait(5)
+    finally:
+        server.shutdown()
+        server.server_close()
+
+
+if __name__ == "__main__":
+    main()

benchmarks/learn/joern_leak_audit.py

@@ -0,0 +1,239 @@
+"""Offline audit of completed Joern results; never launches an analyzer or reads corpora.
+
+Missing results stay in coverage, not signal denominators. Observed non-path rows
+score zero for path features. Sink kinds are one-hot indicators, never ordinal
+category IDs. Both signed and best-direction metrics expose reversed shortcuts.
+"""
+
+from __future__ import annotations
+
+import argparse
+import ast
+import hashlib
+import json
+import re
+from collections import Counter
+from pathlib import Path
+
+from slice_leak_audit import THRESHOLD, complete_pairs, metrics
+
+ROOT = Path(__file__).resolve().parents[2]
+UNITS = ROOT / "plane/experiments/exp-005-graph-slice.units.jsonl"
+RESULTS = Path.home() / "ousast-results/plane/engine-trace"
+OUTPUT = ROOT / "benchmarks/measurements/2026-10-04-joern-slice-leak-audit/record.json"
+STATUSES = ("path", "asked-nothing", "failed", "timeout", "unsupported", "missing")
+ASKED = {"path", "asked-nothing"}
+SIGNALS = ("has_path", "shortest_path_steps", "sanitized", "guard_present", "asked")
+ASYMMETRY = ("vulnerable_only", "fixed_only", "both", "neither", "not_both_asked")
+
+
+def materialization_version():
+    # Read the producer's literal without importing its execution dependencies.
+    tree = ast.parse(Path(__file__).with_name("engine_trace.py").read_bytes())
+    for node in tree.body:
+        if isinstance(node, ast.Assign) and any(isinstance(t, ast.Name) and t.id == "MATERIALIZATION_VERSION" for t in node.targets):
+            return ast.literal_eval(node.value)
+    raise ValueError("Producer has no materialization version")
+
+
+def features(row):
+    path = row["status"] == "path"
+    traces = row.get("traces", []) if path else []
+    candidates = [(i, t) for i, t in enumerate(traces) if t.get("steps")]
+    index, trace = min(candidates, key=lambda it: len(it[1]["steps"])) if candidates else (None, {})
+    witnesses = row.get("witness_rows", [])
+    witness = witnesses[index] if index is not None and index < len(witnesses) else {}
+    guard = trace.get("guard", {})
+    sink = witness.get("sinkKind")
+    if not sink:
+        # Retain only the call target: arguments, literals and code are not kinds.
+        match = re.search(r"([\w:$>.\\-]+)\s*\(", str(witness.get("sink", "")))
+        sink = match[1] if match else "unknown"
+    return {
+        "has_path": int(path),
+        "shortest_path_steps": len(trace["steps"]) if trace else None if path else 0,
+        "sanitized": int(bool(trace.get("sanitized", witness.get("sanitized", False)))) if trace else None if path else 0,
+        "guard_present": int(
+            bool(guard.get("text") or guard.get("bounded") or witness.get("bound") or witness.get("bounded"))
+            or any("guard" in s.get("roles", []) for s in trace.get("steps", []))
+        )
+        if trace
+        else None
+        if path
+        else 0,
+        "sink_kind": str(sink) if path else "no-path",
+        "asked": int(row["status"] in ASKED),
+    }
+
+
+def scored(rows, signal):
+    eligible = [r for r in rows if r[signal] is not None]
+    result = metrics(eligible, signal)
+    result["excluded_rows"] = len(rows) - len(eligible)
+    result["flagged"] = (result["pairs"] >= 10 and (result["best_direction_within_pair_accuracy"] or 0) >= THRESHOLD) or (
+        result["rows"] >= 10 and (result["best_direction_pooled_auc"] or 0) >= THRESHOLD
+    )
+    return result
+
+
+def signal_metrics(rows, kinds):
+    return {
+        **{s: scored(rows, s) for s in SIGNALS},
+        "sink_kind": {kind: scored([{**r, "is_kind": int(r["sink_kind"] == kind)} for r in rows], "is_kind") for kind in kinds},
+    }
+
+
+def scope_report(rows, kinds):
+    pairs = complete_pairs(rows)
+    asymmetry = Counter()
+    honest = []
+    for v, f in pairs:
+        if not (v["asked"] and f["asked"]):
+            asymmetry["not_both_asked"] += 1
+        else:
+            honest.extend((v, f))
+            asymmetry[
+                "both"
+                if v["has_path"] and f["has_path"]
+                else "vulnerable_only"
+                if v["has_path"]
+                else "fixed_only"
+                if f["has_path"]
+                else "neither"
+            ] += 1
+    observed = [r for r in rows if r["status"] != "missing"]
+    return {
+        "rows": len(rows),
+        "coverage": {s: sum(r["status"] == s for r in rows) for s in STATUSES},
+        "complete_pairs": len(pairs),
+        "pair_asymmetry": {s: asymmetry[s] for s in ASYMMETRY},
+        "both_asked_pairs": len(honest) // 2,
+        "signals": signal_metrics(observed, kinds),
+        "both_asked_signals": signal_metrics(honest, kinds),
+    }
+
+
+def audit(units, results, *, expected_rows=501, expected_pairs=195):
+    if len(units) != expected_rows or len({u["unit"] for u in units}) != expected_rows:
+        raise ValueError(f"Expected {expected_rows} unique units")
+    if len(complete_pairs(units)) != expected_pairs:
+        raise ValueError(f"Expected {expected_pairs} complete pairs")
+    if not results.is_dir():
+        raise ValueError(f"Results directory is unreadable: {results}")
+    version = materialization_version()
+    unit_index = {u["unit"]: u for u in units}
+    matched = {}
+    files = []
+    counts = Counter()
+    for path in sorted(results.glob("*.json")):
+        raw = path.read_bytes()
+        record = json.loads(raw)  # Corruption fails loudly; producer uses atomic replace.
+        if not isinstance(record, dict) or "repo" not in record or "pin" not in record:
+            continue
+        stale = record.get("materialization_version") != version
+        unfinished = record.get("done") is not True
+        counts["stale"] += stale
+        counts["unfinished"] += unfinished
+        counts["skipped"] += stale or unfinished
+        files.append((path.name, len(raw)))
+        if stale or unfinished:
+            continue
+        counts["accepted_files"] += 1
+        counts["result_rows_read"] += len(record["units"])
+        for row in record["units"]:
+            if row["unit"] not in unit_index:
+                continue
+            unit = unit_index[row["unit"]]
+            if any(row.get(k) != unit.get(k) for k in ("family", "label", "pair")):
+                raise ValueError(f"Result metadata mismatch: {row['unit']}")
+            if row["status"] not in STATUSES[:-1] or not row.get("source"):
+                raise ValueError(f"Invalid result status/source: {row['unit']}")
+            if row["unit"] in matched:
+                raise ValueError(f"Duplicate current result: {row['unit']}")
+            matched[row["unit"]] = {
+                **unit,
+                "source": row["source"],
+                "status": row["status"],
+                "pin_role": row.get("pin_role"),
+                **features(row),
+            }
+    rows = [
+        matched.get(u["unit"], {**u, "source": u.get("source", "unknown"), "status": "missing", **features({"status": "missing"})})
+        for u in units
+    ]
+    kinds = sorted({r["sink_kind"] for r in rows if r["status"] != "missing"})
+    scopes = {"overall": scope_report(rows, kinds)}
+    for field in ("family", "source"):
+        scopes[f"by_{field}"] = {
+            value: scope_report([r for r in rows if r[field] == value], kinds) for value in sorted({r[field] for r in rows})
+        }
+    return {
+        "cost_usd": 0,
+        "materialization_version": version,
+        "threshold": THRESHOLD,
+        "minimum_samples": 10,
+        "unit_rows_read": len(units),
+        "matched_rows": len(matched),
+        "inputs": {
+            **{k: counts[k] for k in ("accepted_files", "result_rows_read", "stale", "unfinished", "skipped")},
+            "files_read": len(files),
+            "bytes_read": sum(size for _, size in files),
+            # Hash a canonical inventory; filenames and paths may identify repositories.
+            "files_sha256": hashlib.sha256(json.dumps(sorted(files), separators=(",", ":")).encode()).hexdigest(),
+        },
+        "semantics": {
+            "signals": "Shortest nonempty trace (first on ties); its sanitizer, guard and corresponding witness sink call target.",
+            "missing": "Coverage only; absent source metadata is unknown. Skipped result units are never consumed.",
+            "non_path": "Observed non-path rows score zero; path rows lacking trace evidence have null trace features.",
+            "asked": "path or asked-nothing=1; failed, timeout, unsupported=0; missing excluded from metrics.",
+            "asymmetry": "Disjoint categories; not_both_asked takes precedence over path asymmetry.",
+            "flags": "Best of both directions, >=0.70 with >=10 pairs for pair accuracy or >=10 rows for AUC.",
+            "sink_kind": "One binary indicator per observed call target, including no-path/unknown; no ordinal encoding.",
+            "limits": "Exploratory associations, not proof of leakage; running input snapshot, no multiple-testing correction.",
+        },
+        **scopes,
+        "units": [{k: r[k] for k in ("unit", "pair", "family", "label", "fold", "source", "status", *SIGNALS, "sink_kind")} for r in rows],
+    }
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--results", type=Path, default=RESULTS)
+    parser.add_argument("--units", type=Path, default=UNITS)
+    parser.add_argument("--output", type=Path, default=OUTPUT)
+    args = parser.parse_args()
+    raw = args.units.read_bytes()
+    units = [json.loads(line) for line in raw.splitlines() if line.strip()]
+    report = audit(units, args.results)
+    report["units_input"] = {"bytes": len(raw), "sha256": hashlib.sha256(raw).hexdigest()}
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    args.output.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n")
+    overall, inputs = report["overall"], report["inputs"]
+    lines = [
+        f"Joern leak audit: $0; materialization_version={report['materialization_version']}",
+        f"Units read: {len(units)}; bytes={len(raw)}; file={args.units}",
+        f"Result files read: {inputs['files_read']}; bytes={inputs['bytes_read']}; accepted={inputs['accepted_files']}",
+        f"Skipped: {inputs['skipped']} (stale={inputs['stale']}, unfinished={inputs['unfinished']}; counts overlap)",
+        f"Result rows read: {inputs['result_rows_read']}; matched={report['matched_rows']}",
+        f"Coverage: {overall['coverage']}",
+        f"Pairs: {overall['complete_pairs']}; both asked={overall['both_asked_pairs']}",
+        f"Asymmetry: {overall['pair_asymmetry']}",
+    ]
+    for signal in (*SIGNALS, "sink_kind"):
+
+        def brief(value, signal=signal):
+            if signal == "sink_kind":
+                return {k: brief_metric(m) for k, m in value.items()}
+            return brief_metric(value)
+
+        lines.append(f"{signal}: all={brief(overall['signals'][signal])}; both asked={brief(overall['both_asked_signals'][signal])}")
+    lines.append(f"Family/source metrics and input inventory digest: {args.output}")
+    print("\n".join(lines))
+
+
+def brief_metric(metric):
+    return {k: metric[k] for k in ("rows", "pairs", "within_pair_accuracy", "pooled_auc", "flagged")}
+
+
+if __name__ == "__main__":
+    main()

benchmarks/learn/slice_leak_audit.py

@@ -141,7 +141,9 @@ def audit(store: MemoryStore, units: Sequence[Mapping[str, Any]], *, expected_ro
         "limitations": (
             "Flat IR: no dominance/guards, textual branch order, potential parameter inputs, conservative call returns; no alias analysis."
         ),
-        "units": rows,
+        "units": [
+            {k: r[k] for k in ("unit", "pair", "family", "label", "fold", "source", "status", "has_flow", "slice_length")} for r in rows
+        ],
     }
 
 

benchmarks/measurements/2026-10-04-decision-engine-harvest-3/record.json

@@ -0,0 +1,326 @@
+{
+ "created": "2026-10-04",
+ "note": "COUNTS ONLY. Batch 3 of real (vulnerable function, fixed function) pairs whose function survives the fix, mined at zero model cost from reviewed GitHub advisories (pip, npm, maven, composer) and added as pointer-pair corpus benchmarks/pairs/advisory-fixes-3 (source `advisory-fixes-3` in learn/sources.toml); labelled, given static feature records, built into the memory store and embedded. No chat-model call, no Docker, no Joern. Candidate, path, function and repository names stay in the catalog; the pairs the reserved-population guard flagged are withheld (found by bisection with the test output discarded; names never printed).",
+ "method": {
+  "enumeration": "gh api /advisories?type=reviewed&ecosystem=<eco>&cwes=<family CWEs>&published=>2023-01-01 per ecosystem (pip, npm, maven, composer) x six compiled families; Go, Ruby and C# not enumerated; newest first",
+  "advisories_enumerated": 8531,
+  "advisories_published_after_2026_06_01": 1855,
+  "automated_screen": "as batch 2 (single family by CWE; linked fix commit, at most 3, one repository; repository not named anywhere under benchmarks/ incl. batches 1-2; excluded v3 file names never opened; no WordPress plugin; one pair per repository; SPDX licence permissive; one-parent commit; the function holding most changed code lines inside its own matcher span, declared once on both sides, body changed, >= 25% of the file's changed code lines) plus three widenings: (a) an advisory linking only a merged pull request uses its single commit or its squash/rebase merge commit; (b) commits of up to 6 non-test source files / 240 changed lines (batch 2: 4 / 120); (c) a second pass over advisories whose CWEs span two families, the family then chosen from the function's diff on review",
+  "file_reads": "commit metadata and the two versions of each changed source file through the contents API into scratch (no clone, no scanner, engine, plane Run or model on these repositories); scratch deleted after",
+  "manual_review": "every automated candidate's in-function diff read (agent, no model call); kept only when the named function itself holds the fix; a weak inline check replaced by a stronger helper call counts as the fix, a function that only forwards a flag or calls a fix made elsewhere does not",
+  "reserved_population_guard": "pytest -q tests/ -k 'reserved or v3' with output discarded, bisected over catalog subsets (18 runs; baseline with an empty batch green); final catalog green",
+  "rename_check": "canonical full_name of every kept repository from the API: no two kept pairs share a repository, none renamed into a name the benchmarks use",
+  "mechanism": "family default as batch 2 (injection/output_encoding/deserialization source_reaches_sink, path path_join_user_input, access_control missing_auth_guard, untrusted_destination validation_strength for CWE-601 else source_reaches_sink); provenance = agent"
+ },
+ "screened": {
+  "advisories_examined": 8805,
+  "outcomes": {
+   "no fix commit linked": 3480,
+   "repository already in benchmarks/ (populations v1/v2 or a pair corpus)": 1993,
+   "licence not permissive or unrecorded": 1249,
+   "repository already kept (one pair per repository)": 657,
+   "kept": 398,
+   "CWEs span several families": 274,
+   "no function declared on both sides with a changed body": 191,
+   "large commit (refactor or release, not the fix)": 173,
+   "fix spread over several commits or repositories": 167,
+   "merge or root commit": 115,
+   "no source file changed": 42,
+   "language outside the feature vocabulary (go/ruby/c#/...)": 32,
+   "manual review: the chosen function does not hold the fix (comment-only, misattributed, partial or family mismatch)": 19,
+   "repository unavailable": 9,
+   "WordPress plugin (population v3 safety)": 4,
+   "fix commit unavailable": 2
+  },
+  "outcomes_note": "single-family pass (screen 2, every advisory examined, no quota met) plus the two-family pass; `kept` = automated candidates; `manual review: the chosen function does not hold the fix (comment-only, misattributed, partial or family mismatch)` = the screen re-meeting a function the 2026-10-03 review had dropped",
+  "candidates_manually_reviewed": 412,
+  "kept_after_review": 255,
+  "review_drops": {
+   "total": 157,
+   "change in the function unrelated to the advisory (refactor, rename, formatting, another advisory)": 33,
+   "function misattributed (matcher span overrun or an undetected enclosing declaration)": 33,
+   "partial: the function only calls or configures the real fix": 29,
+   "function diff not small (rewrite or refactor around the fix)": 28,
+   "unsure": 17,
+   "family mismatch": 11,
+   "same fix as a kept pair in a fork or sibling repository (would leak across repository folds)": 3,
+   "WordPress plugin (population v3 safety; missed by the automated screen)": 2,
+   "reserved-population guard (flagged by the 2026-10-03 guard run; name not printed)": 1
+  },
+  "one_pair_per_repository_drops": 3,
+  "reserved_population_drops": 2,
+  "kept": 251
+ },
+ "kept": {
+  "total": 251,
+  "by_family": {
+   "access_control": 20,
+   "deserialization": 9,
+   "injection": 56,
+   "output_encoding": 69,
+   "path": 71,
+   "untrusted_destination": 26
+  },
+  "post_cutoff_by_family": {
+   "injection": 5,
+   "output_encoding": 5,
+   "path": 2
+  },
+  "post_cutoff_total": 12,
+  "by_language": {
+   "java": 37,
+   "javascript": 37,
+   "php": 46,
+   "python": 83,
+   "typescript": 48
+  },
+  "post_cutoff_by_language": {
+   "python": 6,
+   "typescript": 6
+  },
+  "by_family_language": {
+   "access_control:java": 5,
+   "access_control:javascript": 3,
+   "access_control:php": 7,
+   "access_control:python": 2,
+   "access_control:typescript": 3,
+   "deserialization:java": 2,
+   "deserialization:php": 1,
+   "deserialization:python": 6,
+   "injection:java": 5,
+   "injection:javascript": 7,
+   "injection:php": 9,
+   "injection:python": 24,
+   "injection:typescript": 11,
+   "output_encoding:java": 7,
+   "output_encoding:javascript": 20,
+   "output_encoding:php": 12,
+   "output_encoding:python": 14,
+   "output_encoding:typescript": 16,
+   "path:java": 10,
+   "path:javascript": 4,
+   "path:php": 10,
+   "path:python": 34,
+   "path:typescript": 13,
+   "untrusted_destination:java": 8,
+   "untrusted_destination:javascript": 3,
+   "untrusted_destination:php": 7,
+   "untrusted_destination:python": 3,
+   "untrusted_destination:typescript": 5
+  },
+  "by_ecosystem": {
+   "composer": 46,
+   "maven": 37,
+   "npm": 81,
+   "pip": 87
+  },
+  "licences": {
+   "Apache-2.0": 80,
+   "BSD-2-Clause": 3,
+   "BSD-3-Clause": 18,
+   "ISC": 1,
+   "MIT": 147,
+   "MPL-2.0": 1,
+   "Unlicense": 1
+  },
+  "split": {
+   "holdout": 124,
+   "train": 127
+  },
+  "published_by_year": {
+   "2023": 50,
+   "2024": 52,
+   "2025": 66,
+   "2026": 83
+  },
+  "published_range": [
+   "2023-01-04",
+   "2026-10-01"
+  ],
+  "via_pull_request": 13,
+  "commit_over_120_source_lines": 10,
+  "two_family_advisory_family_chosen_on_review": 3,
+  "verification": {
+   "verified": 251,
+   "excerpt_bytes_read": 1297516
+  },
+  "targets": {
+   "asked": "injection and untrusted_destination >= 60 each; path, access_control, output_encoding, deserialization >= 30 where advisories allow; >= 250 total",
+   "met": {
+    "access_control": false,
+    "deserialization": false,
+    "injection": false,
+    "output_encoding": true,
+    "path": true,
+    "untrusted_destination": false
+   },
+   "limit": "the reviewed-advisory pool, not the review: every advisory since 2023-01-01 in the four ecosystems was examined; the post-cutoff pool was largely spent by batch 2 (its repositories are now excluded), and untrusted_destination/access_control/deserialization advisories mostly lack a linked single fix commit or sit in repositories the benchmarks already use"
+  }
+ },
+ "labels": {
+  "command": "ousast learn labels --root . --out <scratch> --snapshot <scratch>",
+  "rows": [
+   6981,
+   7985
+  ],
+  "pairs_rows": [
+   1936,
+   2940
+  ],
+  "new_pair_rows": 1004
+ },
+ "features": {
+  "command": "benchmarks/learn/build_features.py --labels <the 1004 new rows> --units <merged units.json of `ousast plane harvest --catalog benchmarks/pairs/advisory-fixes-3/catalog.toml` (manifests written, no Run started)> --catalog benchmarks/pairs/advisory-fixes-3/catalog.toml --kinds pairs",
+  "records": {
+   "plane:access_control": 40,
+   "plane:deserialization": 18,
+   "plane:injection": 112,
+   "plane:output_encoding": 138,
+   "plane:path": 142,
+   "plane:untrusted_destination": 52,
+   "static:access_control": 40,
+   "static:deserialization": 18,
+   "static:injection": 112,
+   "static:output_encoding": 138,
+   "static:path": 142,
+   "static:untrusted_destination": 52
+  },
+  "instrument_states_plane_all_families": {
+   "agree:missing": 164,
+   "agree:none": 338,
+   "delta:not_applicable": 502,
+   "engine:missing": 406,
+   "engine:none": 96,
+   "entry_points:not_applicable": 502,
+   "facts:not_applicable": 502,
+   "language:ran": 502,
+   "model_sinks:missing": 502,
+   "quick:ran": 502,
+   "roles:ran": 502,
+   "source:ran": 502,
+   "verify:missing": 164,
+   "verify:none": 338
+  },
+  "note": "quick rules and deterministic roles ran on each side; engine, verify, agree and model sinks never ran for these pairs (missing/none: null features, never zeros)"
+ },
+ "memory_build": {
+  "command": "ousast learn memory build --labels <labels> --units <merged units.json> --profile static --root . (exit 2: adjudications 9 of 43 anonymous functions, unchanged since 2026-10-01; rows kept)",
+  "examples": [
+   1133,
+   1635
+  ],
+  "pairs_examples": [
+   968,
+   1470
+  ],
+  "pairs_counts_after": {
+   "absent_fixed_side": 0,
+   "examples": 1470,
+   "labels": 2940,
+   "no_excerpt": 0,
+   "no_record": 1470,
+   "recorded": 1470,
+   "unread": 0
+  }
+ },
+ "complete_pairs": {
+  "how": "store example rows (profile static) keyed as learn.experiments.freeze_units pairs them: (repo, candidate, family) holding both labels; whole memory, every family",
+  "before": 459,
+  "after": 710,
+  "by_family": {
+   "access_control": [
+    37,
+    57
+   ],
+   "config_secrets": [
+    12,
+    12
+   ],
+   "deserialization": [
+    32,
+    41
+   ],
+   "injection": [
+    68,
+    124
+   ],
+   "memory": [
+    149,
+    149
+   ],
+   "output_encoding": [
+    77,
+    146
+   ],
+   "path": [
+    38,
+    109
+   ],
+   "unknown": [
+    2,
+    2
+   ],
+   "untrusted_destination": [
+    44,
+    70
+   ]
+  },
+  "exp_004_005_units": "195 complete pairs inside exp-004/005's frozen 501 units (six families, evaluation folds); those units are frozen and were not regenerated here"
+ },
+ "positives_only_groups": {
+  "before": 99,
+  "after": 99,
+  "groups_both_by_family": {
+   "access_control": [
+    37,
+    57
+   ],
+   "config_secrets": [
+    2,
+    2
+   ],
+   "deserialization": [
+    32,
+    41
+   ],
+   "injection": [
+    53,
+    109
+   ],
+   "memory": [
+    5,
+    5
+   ],
+   "output_encoding": [
+    38,
+    107
+   ],
+   "path": [
+    38,
+    109
+   ],
+   "unknown": [
+    2,
+    2
+   ],
+   "untrusted_destination": [
+    42,
+    68
+   ]
+  }
+ },
+ "embeddings": {
+  "command": "ousast learn memory embed --budget-usd 0.20 (dry run first: 144,806 estimated tokens, $0.0029)",
+  "requested": 1479,
+  "cached": 1004,
+  "embedded": 475,
+  "calls": 15,
+  "tokens": 150339,
+  "usd": 0.003007
+ },
+ "spend": {
+  "chat_model_calls": 0,
+  "embedding_usd": 0.003007,
+  "github_api": "metadata and raw contents only"
+ }
+}

docs/deployment.md

@@ -17,6 +17,21 @@ This page uses two words strictly:
 been made.** Section B is guidance, not a record. (kind runs a Kubernetes cluster inside Docker
 containers on one machine.)
 
+The kind profile runs on a 7.7 GB server VM. Observability is off by default
+(`kind_observability = false` in `ops/k8s/profiles/kind.toml`). This saves the measured
+930 MiB used by `otel-system`, about 40% of the cluster's 2.3 GB. The smoke Run still
+completes with those deployments stopped. To enable the stack, run
+`OUSAST_KIND_OBSERVABILITY=1 ops/ax/up.sh` or set `kind_observability = true` in the profile.
+
+`up.sh` stages copied Substrate manifests in a temporary checkout view and installs through
+`hack/install-ate-kind.sh`. The local `ops/ax/substrate-kind-lean/` overlay removes the collector,
+Prometheus, Jaeger and their supporting resources before installation. This preserves the
+installer's secret, CRD and image setup. The upstream checkout is unchanged. The shared
+`ate-otel-config` disables trace, metric and log export and the OTEL SDK; it removes the old
+collector endpoint and metric timing keys. On an existing cluster, `up.sh` scales the three
+observability deployments to zero, replaces the ConfigMap data and rolls its consumers when
+the configuration changes. Opting in restores the upstream ConfigMap and observability manifests.
+
 ## What runs where today (implemented)
 
 ```mermaid
@@ -357,7 +372,7 @@ This section answers the maintainer's questions from the code and the upstream c
 
 The decisions of 2026-10-02 set the target: **two profiles of one code base**.
 
-- `kind` is this laptop. It stays a first-class local execution mode. Every change is proven here first.
+- `kind` is this server VM. It stays a first-class local execution mode. Every change is proven here first.
 - `k3s` is production. (k3s is a small Kubernetes distribution.)
 
 Everything marked **planned** is not built. The rest is what runs on this host today.
@@ -368,7 +383,7 @@ flowchart LR
         hook["ousast pre-push: local, 30 s deadline"]
         cli["ousast plane run, remote mode (planned)"]
     end
-    subgraph k3s ["k3s cluster (the kind profile is the same picture on one laptop)"]
+    subgraph k3s ["k3s cluster (the kind profile is the same picture on one server VM)"]
         subgraph axns ["Namespace ax-system"]
             axs["ax-server Deployment + Redis"]
         end
@@ -403,7 +418,7 @@ This picture has no inbound path:
 - the tasks push their artifacts to the store through the egress gateway;
 - the CLI reads the store.
 
-Today's EndpointSlice back to a laptop (`ops/ax/receiver-service.yaml.tmpl`) belongs to the kind
+Today's EndpointSlice back to the server VM (`ops/ax/receiver-service.yaml.tmpl`) belongs to the kind
 profile only. It does not appear in production.
 
 ### ax, Kubernetes and Agent Substrate in three sentences

docs/plane.md

@@ -159,7 +159,7 @@ flowchart LR
 ax is a Kubernetes application over Agent Substrate. Every Task runs the one runner image as an actor. Each actor
 occupies a whole worker. This code base targets two profiles:
 
-- `kind` on a laptop (today).
+- `kind` on a server VM (today).
 - `k3s` in production (planned). There the same CLI submits Runs remotely, and the tasks deliver to the S3 store.
 
 [Production topology: ax on a Kubernetes cluster](deployment.md#production-topology-ax-on-a-kubernetes-cluster)

docs/rustfs.md

@@ -214,3 +214,42 @@ bucket `OUSAST_MEMORY_TEST_BUCKET`, else `S3_BUCKET`. The tests:
 - never configure the bucket.
 
 More on the host setup: [ax on this host](ops/ax/README.md).
+
+## Scoped account for engine workers
+
+Create a **new account** for the k3s engine Deployment. Attach only this policy.
+Do not give it the existing memory-store account's policies.
+
+```json
+{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Sid": "ListEnginePrefixes",
+      "Effect": "Allow",
+      "Action": "s3:ListBucket",
+      "Resource": "arn:aws:s3:::sast-memory",
+      "Condition": {
+        "StringLike": {
+          "s3:prefix": ["engine-queue/", "engine-queue/*", "engine-results/", "engine-results/*"]
+        }
+      }
+    },
+    {
+      "Sid": "EngineObjectsOnly",
+      "Effect": "Allow",
+      "Action": ["s3:GetObject", "s3:PutObject", "s3:DeleteObject"],
+      "Resource": [
+        "arn:aws:s3:::sast-memory/engine-queue/*",
+        "arn:aws:s3:::sast-memory/engine-results/*"
+      ]
+    }
+  ]
+}
+```
+
+The cluster can read and write only queued engine inputs, claims, and results.
+It cannot list or read the rest of the memory store. It cannot change bucket
+settings. The host queue dispatcher can use the same scoped account. Store its
+keys in the `ousast-engine-s3` Secret; no key values belong in manifests.
+Deployment and coordinator commands are in `ops/k8s/engine/README.md`.

ops/ax/README.md

@@ -17,6 +17,21 @@ in `~/.cache/ousast/ax-src/`. The cluster's context, registry and pool sizes com
 (`ops/k8s/profiles/kind-images.json`), which the templates are re-pinned from and which is committed with them.
 `up.sh` and `smoke-run.sh` run the checkout's code with `OUSAST_PYTHON` (default `.venv/bin/python`).
 
+The kind profile runs on a 7.7 GB server VM. Observability is off by default
+(`kind_observability = false` in `ops/k8s/profiles/kind.toml`). This saves the measured
+930 MiB used by `otel-system`, about 40% of the cluster's 2.3 GB. The smoke Run still
+completes with those deployments stopped. To enable the stack, run
+`OUSAST_KIND_OBSERVABILITY=1 ops/ax/up.sh` or set `kind_observability = true` in the profile.
+
+`up.sh` stages copied Substrate manifests in a temporary checkout view and installs through
+`hack/install-ate-kind.sh`. The local `ops/ax/substrate-kind-lean/` overlay removes the collector,
+Prometheus, Jaeger and their supporting resources before installation. This preserves the
+installer's secret, CRD and image setup. The upstream checkout is unchanged. The shared
+`ate-otel-config` disables trace, metric and log export and the OTEL SDK; it removes the old
+collector endpoint and metric timing keys. On an existing cluster, `up.sh` scales the three
+observability deployments to zero, replaces the ConfigMap data and rolls its consumers when
+the configuration changes. Opting in restores the upstream ConfigMap and observability manifests.
+
 ## How a Run executes
 
 A Run (`plane/runs/*.yaml`, kind `openultrasast.io/v1alpha1 Run`) is a DAG of steps; each step names an ax Task
@@ -191,7 +206,7 @@ configure the bucket.
 No deployment outside this host's kind cluster has been made. The step-by-step guidance, with what is implemented
 and what is planned, is `docs/deployment.md` (the site page "Deployment"). The Model and Workspace manifests under
 `plane/` move as they are and the Task templates need their image reference re-pinned to your registry; these parts
-assume this laptop and must change first (checked 2026-10-02):
+assume this server VM and must change first (checked 2026-10-02):
 
 | Assumption | Where | Needed in a real cluster |
 |---|---|---|

ops/ax/prepare-substrate.py

@@ -0,0 +1,43 @@
+#!/usr/bin/env python3
+"""Stage an installer checkout with lean kind manifests; never modify the source checkout.
+
+Usage: python ops/ax/prepare-substrate.py SOURCE EMPTY_DESTINATION
+The destination lives only for an install/render. Source/build inputs are symlinked;
+manifests are copied because the installer uses fixed paths (including the OTEL ConfigMap).
+"""
+
+from __future__ import annotations
+
+import shutil
+import sys
+from pathlib import Path
+
+import yaml
+
+
+def prepare(source: Path, destination: Path) -> None:
+    source = source.resolve()
+    destination.mkdir(parents=True, exist_ok=True)
+    if any(destination.iterdir()):
+        raise ValueError(f"staging directory must be empty: {destination}")
+    for entry in source.iterdir():
+        if entry.name != "manifests":
+            (destination / entry.name).symlink_to(entry, target_is_directory=entry.is_dir())
+    shutil.copytree(source / "manifests", destination / "manifests")
+    install = destination / "manifests/ate-install"
+    shutil.copytree(install / "kind", install / "kind-upstream")
+    overlay = Path(__file__).resolve().parent / "substrate-kind-lean"
+    shutil.copytree(overlay, install / "kind", dirs_exist_ok=True)
+    # The installer applies this file directly before it renders the bundle.
+    # Keep the strategic-merge directive only in the overlay's patch file.
+    config = yaml.safe_load((overlay / "ate-otel-config.yaml").read_text())
+    config["data"].pop("$patch")
+    (install / "kind/disabled-otel-config.yaml").write_text(yaml.safe_dump(config))
+    (install / "kind/ate-otel-config.yaml").rename(install / "kind/otel-config-patch.yaml")
+    (install / "kind/disabled-otel-config.yaml").rename(install / "kind/ate-otel-config.yaml")
+    kustomization = install / "kind/kustomization.yaml"
+    kustomization.write_text(kustomization.read_text().replace("path: ate-otel-config.yaml", "path: otel-config-patch.yaml"))
+
+
+if __name__ == "__main__":
+    prepare(Path(sys.argv[1]), Path(sys.argv[2]))

ops/ax/substrate-kind-lean/ate-otel-config.yaml

@@ -0,0 +1,11 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: ate-otel-config
+  namespace: ate-system
+data:
+  $patch: replace
+  OTEL_TRACES_EXPORTER: none
+  OTEL_METRICS_EXPORTER: none
+  OTEL_LOGS_EXPORTER: none
+  OTEL_SDK_DISABLED: "true"

ops/ax/substrate-kind-lean/kustomization.yaml

@@ -0,0 +1,44 @@
+apiVersion: kustomize.config.k8s.io/v1beta1
+kind: Kustomization
+# prepare-substrate.py stages the upstream kind overlay beside this directory.
+resources:
+  - ../kind-upstream
+patches:
+  - target:
+      namespace: otel-system
+    patch: |-
+      $patch: delete
+      apiVersion: v1
+      kind: ConfigMap
+      metadata:
+        name: unused
+  - target:
+      kind: Namespace
+      name: otel-system
+    patch: |-
+      $patch: delete
+      apiVersion: v1
+      kind: Namespace
+      metadata:
+        name: otel-system
+  - target:
+      kind: ClusterRole
+      name: ate-prometheus
+    patch: |-
+      $patch: delete
+      apiVersion: rbac.authorization.k8s.io/v1
+      kind: ClusterRole
+      metadata:
+        name: ate-prometheus
+  - target:
+      kind: RoleBinding
+      name: ate-prometheus
+      namespace: ate-system
+    patch: |-
+      $patch: delete
+      apiVersion: rbac.authorization.k8s.io/v1
+      kind: RoleBinding
+      metadata:
+        name: ate-prometheus
+        namespace: ate-system
+  - path: ate-otel-config.yaml

ops/ax/up.sh

@@ -15,6 +15,7 @@ profile() { PYTHONPATH="$REPO_ROOT/src" "$PY" -m openultrasast.plane.profile --p
 CTX="$(profile kube_context)"                       # kind names its context kind-<cluster>
 export KIND_CLUSTER_NAME="${CTX#kind-}"
 export KO_DOCKER_REPO="${KO_DOCKER_REPO:-$(profile registry)}"
+OBSERVABILITY="$(profile kind_observability)"
 IMAGES_FILE="$(profile images)"                     # the digest pins the reconciler and the templates read
 SRC="${OUSAST_AX_SRC:-$HOME/.cache/ousast/ax-src}"
 
@@ -37,9 +38,68 @@ else
 fi
 
 step "agent substrate (ate-system)"
+# The upstream installer has fixed manifest paths and also creates secrets/CRDs.
+# Give it a temporary checkout view so it never installs the observability resources.
+install_substrate() (
+  if [ "$OBSERVABILITY" = true ]; then
+    cd "$SRC/substrate"
+  else
+    # Symlinked build inputs must not make the staged view a different binary version.
+    export VERSION="${VERSION:-$(git -C "$SRC/substrate" describe --tags --always --dirty)}"
+    staged="$(mktemp -d)"
+    trap 'rm -rf "$staged"' EXIT
+    "$PY" "$REPO_ROOT/ops/ax/prepare-substrate.py" "$SRC/substrate" "$staged"
+    cd "$staged"
+  fi
+  KUBECTL_CONTEXT="$CTX" hack/install-ate-kind.sh --deploy-ate-system
+)
 if ready ate-system; then echo "substrate ready; kept"; else
-  (cd "$SRC/substrate" && hack/install-ate-kind.sh --deploy-ate-system)
+  install_substrate
 fi
+# Reconcile the choice even when the existing control plane is already ready.
+if [ "$OBSERVABILITY" = true ]; then
+  for manifest in otel-collector prometheus; do
+    kubectl --context "$CTX" apply -f "$SRC/substrate/manifests/ate-install/kind/$manifest.yaml"
+  done
+  otel_config="$SRC/substrate/manifests/ate-install/kind/ate-otel-config.yaml"
+else
+  otel_config="$REPO_ROOT/ops/ax/substrate-kind-lean/ate-otel-config.yaml"
+  for deployment in opentelemetry-collector prometheus jaeger; do
+    existing="$(kubectl --context "$CTX" -n otel-system get deployment "$deployment" --ignore-not-found -o name)"
+    if [ -n "$existing" ]; then
+      kubectl --context "$CTX" -n otel-system scale "$existing" --replicas=0
+    fi
+  done
+  echo "kind observability off: existing otel-system deployments scaled to 0; opt in with OUSAST_KIND_OBSERVABILITY=1 ops/ax/up.sh"
+fi
+# Replace data to remove stale exporter/endpoint keys in either direction. ConfigMap
+# envFrom is read at pod start; a content annotation triggers only changed rollouts.
+otel_data="$("$PY" - "$otel_config" <<'PYCONFIG'
+import json
+import sys
+import yaml
+config = yaml.safe_load(open(sys.argv[1]))
+config["data"].pop("$patch", None)
+print(json.dumps(config["data"], sort_keys=True))
+PYCONFIG
+)"
+kubectl --context "$CTX" -n ate-system patch configmap ate-otel-config --type=json \
+  -p "[{\"op\":\"replace\",\"path\":\"/data\",\"value\":$otel_data}]"
+otel_hash="$(printf '%s' "$otel_data" | sha256sum | cut -d ' ' -f1)"
+otel_workloads="$(kubectl --context "$CTX" -n ate-system get deployments,daemonsets -o json | "$PY" -c '
+import json, sys
+for item in json.load(sys.stdin)["items"]:
+    spec = item["spec"]["template"]["spec"]
+    containers = spec.get("containers", []) + spec.get("initContainers", [])
+    if any(env.get("configMapRef", {}).get("name") == "ate-otel-config"
+           for container in containers for env in container.get("envFrom", [])):
+        print(item["kind"].lower() + "/" + item["metadata"]["name"])
+')"
+for workload in $otel_workloads; do
+  kubectl --context "$CTX" -n ate-system patch "$workload" --type=merge \
+    -p "{\"spec\":{\"template\":{\"metadata\":{\"annotations\":{\"ousast.io/otel-config\":\"$otel_hash\"}}}}}"
+  kubectl --context "$CTX" -n ate-system rollout status "$workload" --timeout=300s
+done
 kubectl --context "$CTX" -n ate-system get pods
 
 step "egress gateway (atenet-egress, agentgateway variant: one prebuilt image, nothing built)"

ops/k8s/engine/README.md

@@ -0,0 +1,85 @@
+# Engine workers on k3s
+
+Build the new **Engine image** workflow on main (or dispatch it), then copy its
+`ghcr.io/<owner>/openultrasast@sha256:...` job-summary value. The old September
+image does not include boto3 or this service. The release workflow still builds
+version tags. The Dockerfile copies the worker and parser under `/app/benchmarks/learn`
+and installs `.[semantic,s3]`; no analyzer tarball is sent by the dispatcher.
+
+Use two schedulable nodes with at least 3 GiB available each. Required hostname
+anti-affinity places one worker on each node. Updates use Recreate so a third
+worker does not wait forever for a third node. Local scratch uses the container's
+filesystem, with no host mount or Kubernetes API access.
+
+Create a **new scoped account** using the policy in `docs/rustfs.md`. Export its
+`S3_ENDPOINT`, `S3_BUCKET=sast-memory`, `AWS_ACCESS_KEY_ID`, and
+`AWS_SECRET_ACCESS_KEY` in the coordinator shell. Do not use the memory-store
+account. The following commands are for the coordinator; they were not run by the
+implementation agent.
+
+```sh
+export ENGINE_IMAGE='ghcr.io/<owner>/openultrasast@sha256:<digest-from-workflow>'
+# From the repository root, edit the digest parameter without changing the template:
+export ENGINE_OVERLAY="$(mktemp -d /tmp/ousast-engine.XXXXXX)"
+cp ops/k8s/engine/{namespace,deployment,kustomization}.yaml "$ENGINE_OVERLAY/"
+python - <<'PY'
+import os
+from pathlib import Path
+image, digest = os.environ['ENGINE_IMAGE'].split('@', 1)
+p = Path(os.environ['ENGINE_OVERLAY']) / 'kustomization.yaml'
+p.write_text(p.read_text().replace('ghcr.io/OWNER/openultrasast', image)
+             .replace('sha256:REPLACE_WITH_ENGINE_IMAGE_DIGEST', digest))
+PY
+kubectl apply -f ops/k8s/engine/namespace.yaml
+kubectl -n ousast-engine create secret generic ousast-engine-s3 \
+  --from-literal=S3_ENDPOINT="$S3_ENDPOINT" \
+  --from-literal=S3_BUCKET="$S3_BUCKET" \
+  --from-literal=AWS_ACCESS_KEY_ID="$AWS_ACCESS_KEY_ID" \
+  --from-literal=AWS_SECRET_ACCESS_KEY="$AWS_SECRET_ACCESS_KEY"
+kubectl apply -k "$ENGINE_OVERLAY"
+kubectl -n ousast-engine rollout status deployment/ousast-engine-worker
+kubectl -n ousast-engine get pods -o wide
+
+export PATH=/home/mc/Source/OpenUltraSAST/.venv/bin:$PATH
+export PYTHONPATH="$PWD/src"
+python benchmarks/learn/engine_trace.py --queue-status
+python benchmarks/learn/engine_trace.py --executor queue \
+  --image "$ENGINE_IMAGE" --parallel 2 --deadline 900 --queue-timeout 3600 \
+  --units plane/experiments/exp-005-graph-slice.units.jsonl \
+  --out "$HOME/ousast-results/plane/engine-trace-queue"
+```
+
+The dispatcher uses the existing local memory examples by default. Add
+`--examples /path/to/examples.jsonl` for a frozen example snapshot. Materialization
+uses locally cached source pins, just as the Docker executor does. `--executor
+k8s-jobs` preserves the old Job lane (`k8s` is its legacy alias); `queue` requires
+no kubectl access. Resume skips completed local units and retries failed or timed
+out units. `progress.json`, `STOP`, filtering, and `--summary` retain their existing
+meaning. Each invocation uses a new run ID, so an old remote result cannot satisfy
+a new retry. The dispatcher records worker and pod identity in each pin JSON.
+
+Tasks contain the pin, units, prepared questions, source object key, and deadlines.
+Results live at `engine-results/<run>/<pin-id>.json`; inputs are removed after
+publication. The Docker lane retains JSON and a log, not a result tar, so the queue
+also returns JSON. Queue timeout records timeout locally and removes pending;
+an already claimed task finishes remotely and cleans its input. Orphaned inputs
+from a failed submission can be removed under `engine-queue/inputs/` after checking
+that no pending task or claim references them.
+
+Claims use per-worker choosing markers followed by ordered tickets. A worker waits
+for choosing peers and every lower ticket before removing pending and starting.
+This requires strongly consistent GET/PUT/LIST from RustFS, including paginated
+LIST, and synchronized node clocks with skew below 30 seconds. Half of the 60-second lease margin is reserved for clock
+skew; the remaining execution budget is checked again after claiming.
+It does not require conditional writes, object tags, S3 Select, or bucket-admin
+permissions. A dead worker's task is recovered from its expired claim. Execution
+runs in a child process. Linux subreaping and a frozen descendant walk kill its
+entire tree, including Joern's separate sessions, at the task deadline before the
+lease expires. As with any unfenced lease, a node suspended beyond its lease must be
+terminated before it rejoins; the protocol cannot guarantee exclusivity against
+arbitrary process suspension or an inconsistent object store.
+
+SIGTERM stops new claims and lets the current task finish. The 900-second
+termination grace matches the maximum task deadline. Keep these values aligned
+if changing the deployment. `/healthz` serves both probes and becomes unready on
+shutdown. Errors omit SDK exception text to avoid exposing credentials.

ops/k8s/engine/deployment.yaml

@@ -0,0 +1,63 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: ousast-engine-worker
+spec:
+  replicas: 2
+  strategy:
+    type: Recreate
+  selector:
+    matchLabels:
+      app: ousast-engine-worker
+  template:
+    metadata:
+      labels:
+        app: ousast-engine-worker
+    spec:
+      automountServiceAccountToken: false
+      terminationGracePeriodSeconds: 900
+      securityContext:
+        runAsNonRoot: true
+        runAsUser: 1000
+        runAsGroup: 1000
+      affinity:
+        podAntiAffinity:
+          requiredDuringSchedulingIgnoredDuringExecution:
+            - labelSelector:
+                matchLabels:
+                  app: ousast-engine-worker
+              topologyKey: kubernetes.io/hostname
+      containers:
+        - name: worker
+          image: engine-image
+          command: [python, /app/benchmarks/learn/engine_worker_service.py]
+          envFrom:
+            - secretRef:
+                name: ousast-engine-s3
+          env:
+            - name: ENGINE_IMAGE
+              value: engine-image
+          resources:
+            requests:
+              memory: 3Gi
+              cpu: '1'
+            limits:
+              memory: 3Gi
+          securityContext:
+            allowPrivilegeEscalation: false
+            capabilities:
+              drop: [ALL]
+          ports:
+            - name: health
+              containerPort: 8080
+          readinessProbe:
+            httpGet:
+              path: /healthz
+              port: health
+            periodSeconds: 5
+          livenessProbe:
+            httpGet:
+              path: /healthz
+              port: health
+            initialDelaySeconds: 15
+            periodSeconds: 10

ops/k8s/engine/kustomization.yaml

@@ -0,0 +1,22 @@
+apiVersion: kustomize.config.k8s.io/v1beta1
+kind: Kustomization
+namespace: ousast-engine
+resources:
+  - namespace.yaml
+  - deployment.yaml
+# Deliberately non-pullable until the coordinator supplies the workflow's digest.
+images:
+  - name: engine-image
+    newName: ghcr.io/OWNER/openultrasast
+    digest: sha256:REPLACE_WITH_ENGINE_IMAGE_DIGEST
+replacements:
+  - source:
+      kind: Deployment
+      name: ousast-engine-worker
+      fieldPath: spec.template.spec.containers.0.image
+    targets:
+      - select:
+          kind: Deployment
+          name: ousast-engine-worker
+        fieldPaths:
+          - spec.template.spec.containers.0.env.0.value

ops/k8s/engine/namespace.yaml

@@ -0,0 +1,4 @@
+apiVersion: v1
+kind: Namespace
+metadata:
+  name: ousast-engine

ops/k8s/engine/secret.example.yaml

@@ -0,0 +1,9 @@
+# Reference only; excluded from kustomization. Create using the README command.
+apiVersion: v1
+kind: Secret
+metadata:
+  name: ousast-engine-s3
+  namespace: ousast-engine
+type: Opaque
+stringData: {}
+# Required keys: S3_ENDPOINT, S3_BUCKET, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY

ops/k8s/profiles/kind-images.json

@@ -1,3 +1,3 @@
 {
-  "runner": "localhost:5001/ousast-runner@sha256:d2a55293999cc34581ef1f02a1ffb0c2dd8fc0cbe001ac952df81600e868f019"
+  "runner": "localhost:5001/ousast-runner@sha256:784bc6afc869c679c4966d04d3f9653a8d379726b899809590cc83327f4038e3"
 }

ops/k8s/profiles/kind.toml

@@ -1,8 +1,9 @@
-# The kind profile: the plane on this laptop's single-node kind cluster, which ops/ax/up.sh creates and which is
+# The kind profile: the plane on this server VM's single-node kind cluster, which ops/ax/up.sh creates and which is
 # the development default (plane-on-kubernetes design section 1). Values only; a credential never goes here
 # (keys stay in the environment or .env). Every field is overridable by OUSAST_<FIELD> in the environment, a pool
 # by OUSAST_POOL_<NAME>_<FIELD>. This file may name the kind cluster; the code must not (tests/test_plane_literals.py).
 
+kind_observability = false      # opt in with OUSAST_KIND_OBSERVABILITY=1
 exec = "local"                   # the reconciler runs on this host against the cluster below
 kube_context = "kind-ousast"     # kind's context for KIND_CLUSTER_NAME=ousast
 registry = "localhost:5001"      # the kind-local registry up.sh creates (Substrate rewrites it for the node)

plane/Dockerfile.engine-task

@@ -0,0 +1,52 @@
+# Build from the repository root. AX v0.3.1 starts spec.command immediately;
+# preserve its runner as PID 1 and put the engine command in the Task manifest.
+FROM eclipse-temurin:21-jre-jammy AS jre
+
+FROM ghcr.io/norandom/ax-task-runner:v0.3.1
+LABEL org.opencontainers.image.licenses="Apache-2.0" \
+      org.opencontainers.image.source="https://github.com/norandom/OpenUltraSAST"
+
+USER root
+ARG JOERN_VERSION=v4.0.625
+ARG ARCHIVE=joern-cli-linux-x86_64.zip
+
+# The AX runner is Python 3.12 / Debian slim. Copy the Java 21 runtime from
+# Jammy (older glibc than Debian bookworm) instead of relying on distro JRE versions.
+COPY --from=jre /opt/java/openjdk /opt/java/openjdk
+ENV JAVA_HOME=/opt/java/openjdk
+ENV PATH="/venv/bin:/opt/java/openjdk/bin:/opt/joern-cli:${PATH}"
+RUN apt-get update \
+ && apt-get install -y --no-install-recommends \
+      php-cli unzip curl ca-certificates libstdc++6 zlib1g \
+ && rm -rf /var/lib/apt/lists/*
+
+RUN set -eux; \
+    cd /tmp; \
+    curl -fsSL -O "https://github.com/joernio/joern/releases/download/${JOERN_VERSION}/${ARCHIVE}"; \
+    curl -fsSL -O "https://github.com/joernio/joern/releases/download/${JOERN_VERSION}/${ARCHIVE}.sha512"; \
+    sed 's@  target/@  @' "${ARCHIVE}.sha512" > checksum.sha512; \
+    sha512sum -c checksum.sha512; \
+    unzip -q "${ARCHIVE}" -d /opt; \
+    rm -f "${ARCHIVE}" "${ARCHIVE}.sha512" checksum.sha512; \
+    /opt/joern-cli/joern-parse --help > /dev/null
+
+RUN if [ ! -e /opt/joern-cli/jssrc2cpg ] && [ -x /opt/joern-cli/jssrc2cpg.sh ]; then \
+      ln -s jssrc2cpg.sh /opt/joern-cli/jssrc2cpg; \
+    fi
+COPY ops/frontend-retention/install.py /tmp/frontend-retention-install.py
+RUN python3 /tmp/frontend-retention-install.py /opt/joern-cli \
+ && rm /tmp/frontend-retention-install.py
+
+WORKDIR /app
+COPY pyproject.toml README.md LICENSE ./
+COPY src ./src
+COPY benchmarks/learn/engine_task_entry.py \
+     benchmarks/learn/engine_trace_worker.py \
+     benchmarks/learn/engine_trace_parse.py ./benchmarks/learn/
+# Worker imports are standard-library plus engine modules. Isolate them from
+# the AX runner's Python environment and omit PyYAML, boto3 and all extras.
+RUN python3 -m venv /venv \
+ && /venv/bin/pip install --no-cache-dir --no-deps . \
+ && /venv/bin/python -I -c 'import pathlib, runpy, sys; sys.modules.update(yaml=None, boto3=None); sys.path.insert(0, "/app/benchmarks/learn"); p = pathlib.Path("/app/benchmarks/learn/engine_trace_worker.py"); print("Worker import smoke bytes:", len(p.read_bytes())); runpy.run_path(str(p), run_name="engine_worker_import_smoke")'
+
+# Do not set ENTRYPOINT or CMD: both belong to the upstream AX runner.

pyproject.toml

@@ -33,6 +33,7 @@ semantic = [
     "tree-sitter-c>=0.23",
     "tree-sitter-cpp>=0.23",
     "tree-sitter-java>=0.23",
+    "tree-sitter-php>=0.23",
 ]
 
 [project.scripts]

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)
   }

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))

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---")

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)

src/openultrasast/cpg/session.py

@@ -31,10 +31,12 @@ import os
 import signal
 import socket
 import subprocess
+import threading
 import time
 import urllib.error
 import urllib.request
 import uuid
+from collections.abc import Callable
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any
@@ -78,6 +80,7 @@ class EngineSession:
     # with nothing: measured as a 25.3 s "build" of which only 4.5 s was the frontend.
     startup_allowance_seconds: float = 20.0
     work_reserve_seconds: float = 10.0
+    output_observer: Callable[[str], bool] | None = None
     failure: str = ""
     startup_seconds: float = 0.0
     loaded_graph: str = ""
@@ -199,7 +202,34 @@ class EngineSession:
         target = self.scratch / f"answer-{nonce}.txt"
         wrapped = _redirected(code, target, nonce)
         started = time.monotonic()
-        body = self._post(wrapped, timeout=remaining)
+        if self.output_observer is None:
+            body = self._post(wrapped, timeout=remaining)
+        else:
+            # Only HTTP runs in the helper; observation and cancellation stay on the caller.
+            done = threading.Event()
+            response: list[dict[str, object] | None] = []
+
+            def post() -> None:
+                try:
+                    response.append(self._post(wrapped, timeout=remaining))
+                finally:
+                    done.set()
+
+            threading.Thread(target=post, daemon=True).start()
+            while True:
+                finished = done.wait(0.05)
+                try:
+                    first, _, output = target.read_text(errors="replace").partition("\n")
+                except OSError:
+                    first, output = "", ""
+                if not self.output_observer(output if first.strip() == nonce else ""):
+                    self._poison("session_observer_cancelled")
+                    self.close()
+                    target.unlink(missing_ok=True)
+                    return None
+                if finished:
+                    break
+            body = response[0] if response else None
         seconds = time.monotonic() - started
         self.requests += 1
         if body is None:

src/openultrasast/learn/cli.py

@@ -15,6 +15,14 @@ from typing import Any
 
 
 def add_commands(learn_sub: Any) -> None:
+    mechanisms = learn_sub.add_parser("mechanisms", help="deterministic candidate discharge facts (no admission)")
+    mechanisms_sub = mechanisms.add_subparsers(dest="mechanisms_command", required=True)
+    mine = mechanisms_sub.add_parser("mine", help="mine fix pairs; no model or fact-table writes")
+    mine.add_argument("--memory", help="the example store (default OUSAST_MEMORY)")
+    mine.add_argument("--out", type=Path, required=True, help="candidate JSONL under the results root, outside this repository")
+    mine.add_argument("--profile", choices=("static", "plane"), default="static")
+    mine.add_argument("--spec", type=Path, help="decision engine compile spec defining the fold seed/split")
+    mine.add_argument("--root", type=Path, default=Path("."), help="repository root for the counts-only measurement")
     memory = learn_sub.add_parser("memory", help="the decision engine's local memory of labelled examples")
     memory_sub = memory.add_subparsers(dest="memory_command", required=True)
     build = memory_sub.add_parser("build", help="label + feature record + excerpt -> example rows (no model)")
@@ -97,6 +105,8 @@ def add_commands(learn_sub: Any) -> None:
 
 
 def run(args: argparse.Namespace) -> int:
+    if args.learn_command == "mechanisms":
+        return _mechanisms_mine(args)
     if args.learn_command == "audit-leaks":
         return _audit_leaks(args)
     if args.learn_command == "memory" and args.memory_command == "build":
@@ -111,6 +121,25 @@ def run(args: argparse.Namespace) -> int:
     return 2
 
 
+def _mechanisms_mine(args: argparse.Namespace) -> int:
+    from ..plane.memory import MemoryStoreError, open_store
+    from ..plane.reconciler import results_root
+    from .mechanisms import mine_store, write_result
+
+    try:
+        root = args.root.resolve()
+        out = args.out.resolve()
+        if out.is_relative_to(root) or not out.is_relative_to(results_root().resolve()):
+            raise ValueError("--out must be under the results root and outside the repository")
+        result = mine_store(open_store(args.memory), profile=args.profile, spec_path=args.spec)
+        write_result(result, out, root=root, results=results_root())
+    except (ValueError, OSError, MemoryStoreError) as exc:
+        print(f"learn mechanisms mine: {exc}", file=sys.stderr)
+        return 2
+    print(json.dumps(result.record, sort_keys=True))
+    return 0 if result.record["pairs_mined"] else 2
+
+
 def _experiment(args: argparse.Namespace) -> int:
     """register | units | run | analyse. Exit 2 on a refused manifest (unregistered, modified, uncommitted), 3 when a
     ceiling stops the run (``unfinished``: a rerun resumes from the outcome rows)."""

src/openultrasast/learn/mechanisms.py

@@ -0,0 +1,612 @@
+"""Deterministic discharge proposals, never admitted facts.
+
+The shared tree-sitter IR supplies function/call identity. Its CST supplies bounded
+expression containment and structured if branches; loops, exceptions, alias writes,
+and unmodelled expressions fail closed. Parameters and fact-table sources are
+potential taint, as in learn.slice. A return is an output boundary only for
+output_encoding (a proposal, not a claim that every return renders HTML).
+
+Patterns use zero-based argument positions. Only >=2 independent repository
+proposers survive. Each fold carries its own training evidence; global evidence
+must never be used to admit a fact for a held-out fold. No fact table/store writes.
+"""
+
+from __future__ import annotations
+
+import ast
+import hashlib
+import json
+import re
+from collections import Counter, defaultdict
+from collections.abc import Callable, Mapping, Sequence
+from dataclasses import dataclass
+from pathlib import Path
+from textwrap import dedent
+from typing import Any
+
+from ..model.specs import taint_specs
+from ..plane.memory import MemoryStore, repo_key
+from ..semantic.cst import _KIND, _walk, file_ir_from_tree
+from ..semantic.extra import grammar_for
+from ..semantic.facts import load_facts
+from ..semantic.ir import FileIR, trailing_name
+from ..semantic.taint import _call_matches, _source_in
+from .examples import Example
+from .folds import Fold, compile_split, folds_digest, outer_folds
+
+RECORD = Path("benchmarks/measurements/2026-10-04-fix-mechanisms-mining/record.json")
+SUPPORTED = frozenset({"python", "javascript", "typescript", "tsx", "php"})
+_NUMBERED = re.compile(r"^\s*(\d+)(?:  (.*))?$")
+
+
+@dataclass(frozen=True)
+class Pair:
+    id: str
+    repository: str
+    group: str
+    language: str
+    family: str
+    vulnerable: str
+    fixed: str
+    source: str = "pairs"
+    frameworks: tuple[str, ...] = ()
+    package: str = ""
+    example_ids: tuple[str, ...] = ()
+
+
+@dataclass
+class MiningResult:
+    candidates: list[dict[str, Any]]
+    record: dict[str, Any]
+
+
+class Unsupported(ValueError):
+    """A counted skip, never an empty successful analysis."""
+
+
+def _text(node: Any) -> str:
+    return bytes(node.text).decode("utf-8")
+
+
+def _hash(value: str) -> str:
+    return hashlib.sha256(value.encode()).hexdigest()
+
+
+def _canonical(value: Any) -> str:
+    return json.dumps(value, sort_keys=True, separators=(",", ":"))
+
+
+def _parse(text: str, language: str) -> tuple[Any, FileIR]:
+    if language not in SUPPORTED:
+        raise Unsupported("unsupported_language")
+    grammar = grammar_for(language)
+    if grammar is None:
+        raise Unsupported("grammar_unavailable")
+    lines = text.split("\ndiff:\n", 1)[0].splitlines()
+    matches = [_NUMBERED.match(line) for line in lines]
+    if any(matches):
+        if not all(matches) or len({m[1] for m in matches if m}) != len(lines):
+            raise Unsupported("unparseable")
+        lines = [(m[2] or "") for m in matches if m]
+    code = dedent("\n".join(lines))
+    if language == "php" and not code.lstrip().startswith("<?php"):
+        code = "<?php\n" + code
+    from tree_sitter import Parser
+
+    tree = Parser(grammar).parse(code.encode())
+    ir = file_ir_from_tree(path="excerpt", text=code, language=language, tree=tree)
+    if not ir.parse_ok or not code.strip():
+        raise Unsupported("unparseable")
+    return tree, ir
+
+
+def _callee(name: str, variables: set[str]) -> str:
+    name = name.strip().replace("->", ".").replace("::", ".")
+    if "(" in name or not re.fullmatch(r"[\w$\\]+(?:\.[\w$]+)*", name):
+        raise Unsupported("unsupported_shape")
+    if name.split(".")[0] in variables or name.startswith("$"):
+        if "." not in name:
+            raise Unsupported("unsupported_shape")
+        return trailing_name(name)
+    return name.lstrip("\\")
+
+
+@dataclass(frozen=True)
+class Flow:
+    roots: frozenset[str] = frozenset()
+    patterns: frozenset[str] = frozenset()
+
+
+def _union(flows: Sequence[Flow]) -> Flow:
+    return Flow(frozenset().union(*(f.roots for f in flows)), frozenset().union(*(f.patterns for f in flows)))
+
+
+@dataclass
+class Site:
+    callee: str
+    flow: Flow
+    arguments: dict[str, Any]
+    slot: tuple[int, ...]
+    guards: frozenset[str]
+
+
+class Analysis:
+    """Small structured walker over the shipped IR's call sites, not a new parser."""
+
+    def __init__(self, language: str, family: str, function: Any, ir: Any) -> None:
+        self.language, self.family = language, family
+        self.kinds = _KIND[language]
+        self.calls = {call.order: call for call in ir.calls}
+        self.variables = set(ir.params) | {bind.name for bind in ir.binds}
+        self.states = {name: Flow(frozenset({name})) for name in ir.params}
+        self.sites: list[Site] = []
+        self.facts = load_facts().for_language(language)
+        self.spec = taint_specs(language=language).get(family)
+        self.function = function
+
+    def name(self, node: Any) -> str:
+        call = self.calls.get(node.end_byte)
+        if call is None:
+            raise Unsupported("unsupported_shape")
+        return _callee(call.name, self.variables)
+
+    def args(self, node: Any) -> list[Any]:
+        args = node.child_by_field_name("arguments")
+        if args is None:
+            args = next((c for c in node.named_children if c.type in self.kinds["args"]), None)
+        return list(args.named_children) if args else []
+
+    def expr(self, node: Any, slot: tuple[int, ...], guards: frozenset[str]) -> Flow:
+        kind = node.type
+        children = list(node.named_children)
+        if kind in {"identifier", "variable_name"}:
+            return self.states.get(_text(node), Flow())
+        if kind in {"string", "string_literal", "integer", "float", "number", "true", "false", "null", "none", "boolean"}:
+            return Flow()
+        if kind in self.kinds["call"]:
+            name = self.name(node)
+            args = self.args(node)
+            values = [self.expr(arg, slot + (i,), guards) for i, arg in enumerate(args)]
+            flow = _union(values)
+            if _source_in(self.calls[node.end_byte].name + "(", self.facts):
+                flow = Flow(flow.roots | {"<source:" + name + ">"}, flow.patterns)
+            arguments = _arguments(args)
+            self.sites.append(Site(name, flow, arguments, slot, guards))
+            added = {_canonical({"callee": name, "argument": i}) for i, value in enumerate(values) if value.roots}
+            return Flow(flow.roots, flow.patterns | added)
+        if kind in {"attribute", "member_expression", "subscript", "subscript_expression", "member_access_expression"}:
+            flow = self.expr(children[0], slot, guards) if children else Flow()
+            if _source_in(_text(node), self.facts):
+                return Flow(flow.roots | {"<source>"}, flow.patterns)
+            return flow
+        if kind in {"keyword_argument", "argument", "parenthesized_expression", "unary_expression", "not_operator"}:
+            return self.expr(children[-1], slot, guards) if children else Flow()
+        if kind in {
+            "binary_expression",
+            "binary_operator",
+            "comparison_operator",
+            "list",
+            "tuple",
+            "array",
+            "array_creation_expression",
+            "array_element_initializer",
+            "pair",
+            "dictionary",
+            "object",
+        }:
+            return _union([self.expr(child, slot + (i,), guards) for i, child in enumerate(children)])
+        if kind in {"property_identifier", "name", "comment", "string_content", "escape_sequence"}:
+            return Flow()
+        raise Unsupported("unsupported_shape")
+
+    def predicate(self, node: Any, truth: bool) -> frozenset[str]:
+        while node.type == "parenthesized_expression":
+            node = node.named_children[0]
+        if node.type in {"not_operator", "unary_expression"}:
+            if not _text(node).lstrip().startswith(("not ", "!")):
+                raise Unsupported("unsupported_shape")
+            return self.predicate(node.named_children[-1], not truth)
+        if node.type not in self.kinds["call"]:
+            raise Unsupported("unsupported_shape")
+        name = self.name(node)
+        predicates = []
+        for index, arg in enumerate(self.args(node)):
+            # Evaluating a condition is not an output path or a sanitizer.
+            value = self.expr(arg, (), frozenset())
+            if value.roots:
+                predicates.append(
+                    _canonical(
+                        {"callee": name, "argument": index, "predicate": "truthy" if truth else "falsy", "_roots": sorted(value.roots)}
+                    )
+                )
+        return frozenset(predicates)
+
+    def block(self, node: Any, prefix: tuple[int, ...] = (), guards: frozenset[str] = frozenset()) -> bool:
+        children = list(node.named_children) if node.type in {"block", "statement_block", "compound_statement"} else [node]
+        for index, statement in enumerate(children):
+            slot = prefix + (index,)
+            kind = statement.type
+            if kind in self.kinds["function"] or kind in {"comment", "pass_statement"}:
+                continue
+            if kind == "if_statement":
+                condition = statement.child_by_field_name("condition")
+                consequence = statement.child_by_field_name("consequence") or statement.child_by_field_name("body")
+                alternative = statement.child_by_field_name("alternative")
+                if condition is None or consequence is None or alternative is not None:
+                    raise Unsupported("unsupported_shape")
+                before = dict(self.states)
+                yes, no = self.predicate(condition, True), self.predicate(condition, False)
+                exits = self.block(consequence, slot, guards | yes)
+                # Branch writes need SSA/joins; no invented textual-order dataflow.
+                if self.states != before:
+                    raise Unsupported("unsupported_shape")
+                if exits:
+                    guards |= no
+                continue
+            if kind in {"return_statement", "echo_statement", "throw_statement", "raise_statement"}:
+                flow = _union([self.expr(c, slot, guards) for c in statement.named_children])
+                if self.family == "output_encoding" and kind in {"return_statement", "echo_statement"}:
+                    self.sites.append(Site("<return>" if kind == "return_statement" else "echo", flow, {}, slot, guards))
+                if kind != "echo_statement":
+                    return True
+                continue
+            if kind in {"lexical_declaration", "variable_declaration", "expression_statement"}:
+                if self.block(statement.named_children[0], slot, guards):
+                    return True
+                if len(statement.named_children) != 1:
+                    raise Unsupported("unsupported_shape")
+                continue
+            if kind in self.kinds["assign"]:
+                left = statement.child_by_field_name("left") or statement.child_by_field_name("name")
+                right = statement.child_by_field_name("right") or statement.child_by_field_name("value")
+                if left is None or right is None or left.type not in {"identifier", "variable_name"}:
+                    raise Unsupported("unsupported_shape")
+                if kind.startswith("augmented"):
+                    raise Unsupported("unsupported_shape")
+                self.states[_text(left)] = self.expr(right, slot, guards)
+                continue
+            if kind in self.kinds["call"]:
+                self.expr(statement, slot, guards)
+                if self.name(statement) in {"abort", "exit", "die"}:
+                    return True
+                continue
+            raise Unsupported("unsupported_shape")
+        return False
+
+    def run(self) -> list[Site]:
+        body = self.function.child_by_field_name("body")
+        if body is None:
+            raise Unsupported("unsupported_shape")
+        self.block(body)
+        return self.sites
+
+    def sink(self, site: Site) -> bool:
+        return (self.family == "output_encoding" and site.callee in {"<return>", "echo"}) or bool(
+            self.spec and any(_call_matches(site.callee, sink) for sink in self.spec.sinks)
+        )
+
+
+def _literal(node: Any) -> tuple[bool, Any]:
+    text = _text(node)
+    if text.lower() in {"true", "false", "null", "none"}:
+        return True, {"true": True, "false": False, "null": None, "none": None}[text.lower()]
+    if node.type not in {"string", "string_literal", "integer", "number", "float"}:
+        return False, None
+    try:
+        value = ast.literal_eval(text)
+    except (ValueError, SyntaxError):
+        return False, None
+    return (True, value) if isinstance(value, str | int | float | bool) else (False, None)
+
+
+def _arguments(args: list[Any]) -> dict[str, Any]:
+    result: dict[str, Any] = {}
+    for position, original in enumerate(args):
+        arg = original.named_children[-1] if original.type == "argument" else original
+        key = ""
+        if arg.type == "keyword_argument":
+            key = _text(arg.named_children[0])
+            arg = arg.named_children[-1]
+        ok, value = _literal(arg)
+        if ok:
+            result[_canonical({"key": key} if key else {"argument": position, "key": key})] = value
+        elif arg.type in {"array_creation_expression", "dictionary", "object"}:
+            for entry in arg.named_children:
+                if entry.type not in {"array_element_initializer", "pair"} or len(entry.named_children) != 2:
+                    continue
+                valid_key, key_value = _literal(entry.named_children[0])
+                valid_value, value = _literal(entry.named_children[-1])
+                if valid_key and isinstance(key_value, str) and valid_value:
+                    result[_canonical({"argument": position, "key": key_value})] = value
+    return result
+
+
+def _functions(tree: Any, ir: FileIR) -> dict[str, tuple[Any, Any]]:
+    nodes = [n for n in _walk(tree.root_node) if n.type in _KIND[ir.language]["function"]]
+    functions = list(ir.functions[1:])
+    if not functions or len({f.name for f in functions}) != len(functions):
+        raise Unsupported("unsupported_shape")
+    return {f.name: (node, f) for node, f in zip(nodes, functions, strict=True)}
+
+
+def _proposals(before: Analysis, after: Analysis) -> set[tuple[str, str]]:
+    old, new = before.run(), after.run()
+    result: set[tuple[str, str]] = set()
+    old_sinks = [s for s in old if before.sink(s) and s.flow.roots]
+    for sink in old_sinks:
+        # Ambiguous repeated operations are not aligned by textual similarity.
+        matches = [s for s in new if s.callee == sink.callee and s.flow.roots == sink.flow.roots]
+        if len([s for s in old_sinks if s.callee == sink.callee]) != 1:
+            raise Unsupported("ambiguous_sink")
+        if len(matches) == 1:
+            fixed = matches[0]
+            result.update(("sanitizer", p) for p in fixed.flow.patterns - sink.flow.patterns)
+            # Only predicates involving a value on this sink's path are retained below.
+            for encoded in fixed.guards - sink.guards:
+                predicate = json.loads(encoded)
+                roots = set(predicate.pop("_roots"))
+                if roots & fixed.flow.roots:
+                    result.add(("guard", _canonical(predicate)))
+            for key, value in fixed.arguments.items():
+                if key not in sink.arguments or _canonical(sink.arguments[key]) != _canonical(value):
+                    result.add(("safe_argument", _canonical({"callee": fixed.callee, **json.loads(key), "value": value})))
+        elif not matches:
+            replacements = [s for s in new if s.flow.roots == sink.flow.roots]
+            if len(replacements) == 1 and replacements[0].slot == sink.slot and replacements[0].callee != sink.callee:
+                result.add(("safe_alternative", _canonical({"callee": sink.callee, "replacement": replacements[0].callee})))
+            elif replacements:
+                raise Unsupported("ambiguous_sink")
+        else:
+            raise Unsupported("ambiguous_sink")
+    return result
+
+
+def _known(language: str, family: str, form: str, pattern: Mapping[str, Any]) -> bool:
+    spec = taint_specs(language=language).get(family)
+    if spec is None:
+        return False
+    name = str(pattern.get("replacement") if form == "safe_alternative" else pattern.get("callee"))
+    if form == "guard" and pattern.get("predicate") != "truthy":
+        return False
+    names = spec.guards if form == "guard" else spec.sanitizers + spec.quoted_sanitizers
+    # Shape sinks alone do not establish a particular keyword/literal restriction.
+    return form != "safe_argument" and any(_call_matches(name, known) for known in names)
+
+
+def _project_specific(pattern: Mapping[str, Any], pair: Pair, local: set[str]) -> str | None:
+    names = [str(pattern.get(k, "")) for k in ("callee", "replacement")]
+    if any(trailing_name(n) in local for n in names if n):
+        return "local_definition"
+    identities = [pair.repository.rstrip("/").removesuffix(".git").rsplit("/", 1)[-1], pair.package]
+    compact = re.sub(r"[^a-z0-9]", "", _canonical(pattern).lower())
+    if any((token := re.sub(r"[^a-z0-9]", "", identity.lower())) and token in compact for identity in identities):
+        return "project_identifier"
+    return None
+
+
+def mine_pairs(pairs: Sequence[Pair], *, folds: Sequence[Fold] = ()) -> MiningResult:
+    skips: Counter[str] = Counter()
+    parsed: dict[int, tuple[dict[str, tuple[Any, Any]], dict[str, tuple[Any, Any]]]] = {}
+    definitions: dict[tuple[str, str], set[str]] = defaultdict(set)
+    for index, pair in enumerate(pairs):
+        try:
+            before = _functions(*_parse(pair.vulnerable, pair.language))
+            after = _functions(*_parse(pair.fixed, pair.language))
+            parsed[index] = before, after
+            definitions[repo_key(pair.repository), pair.language].update(before.keys() | after.keys())
+        except Unsupported as exc:
+            skips[str(exc)] += 1
+    proposals: dict[tuple[str, str, str, str], list[Pair]] = defaultdict(list)
+    mined = 0
+    for index, (old, new) in parsed.items():
+        pair = pairs[index]
+        try:
+            common = old.keys() & new.keys()
+            if not common:
+                raise Unsupported("unmatched_function")
+            found: set[tuple[str, str]] = set()
+            for name in sorted(common):
+                found |= _proposals(Analysis(pair.language, pair.family, *old[name]), Analysis(pair.language, pair.family, *new[name]))
+            mined += 1
+            if not found:
+                skips["no_supported_change"] += 1
+            for form, pattern in sorted(found):
+                reason = _project_specific(json.loads(pattern), pair, definitions[repo_key(pair.repository), pair.language])
+                if reason:
+                    skips[reason] += 1
+                else:
+                    proposals[pair.language, pair.family, form, pattern].append(pair)
+        except Unsupported as exc:
+            skips[str(exc)] += 1
+    candidates = []
+    cells: dict[str, dict[str, int]] = {}
+    tops: dict[str, list[dict[str, Any]]] = defaultdict(list)
+    for (language, family, form, encoded), evidence in sorted(proposals.items()):
+        repos = {_hash(repo_key(p.repository)) for p in evidence}
+        pattern = json.loads(encoded)
+        known = _known(language, family, form, pattern)
+        cell = cells.setdefault(f"{language}/{family}/{form}", {"total": 0, "at_least_two_repositories": 0, "known": 0})
+        cell["total"] += 1
+        cell["known"] += int(known)
+        if len(repos) < 2:
+            skips["fewer_than_two_repositories"] += 1
+            continue
+        cell["at_least_two_repositories"] += 1
+        candidate = {
+            "language": language,
+            "family": family,
+            "form": form,
+            "pattern": pattern,
+            "known": known,
+            "proposing_pairs": sorted({p.id for p in evidence}),
+            "proposing_examples": {p.id: sorted(p.example_ids) for p in sorted(evidence, key=lambda p: p.id)},
+            "repository_count": len(repos),
+            "repository_ids": sorted(repos),
+        }
+        training = []
+        for fold in folds:
+            held_repos = {repo_key(p.repository) for p in pairs if p.group in fold.eval_groups}
+            train = [
+                p
+                for p in evidence
+                if p.group not in fold.eval_groups
+                and repo_key(p.repository) not in held_repos
+                and p.source != fold.held_source
+                and (not fold.held_framework or fold.held_framework not in p.frameworks)
+            ]
+            train_repos = sorted({_hash(repo_key(p.repository)) for p in train})
+            training.append(
+                {
+                    "fold": fold.name,
+                    "repository_count": len(train_repos),
+                    "repository_ids": train_repos,
+                    "proposing_pairs": sorted({p.id for p in train}),
+                    "supported": len(train_repos) >= 2,
+                    "eligible": len(train_repos) >= 2 and not known and fold.status == "ok",
+                }
+            )
+        candidate["folds"] = training
+        candidates.append(candidate)
+        tops[form].append({"language": language, "family": family, "pattern": pattern, "repository_count": len(repos)})
+    return MiningResult(
+        candidates,
+        {
+            "pairs_read": len(pairs),
+            "pairs_mined": mined,
+            "skips": dict(sorted(skips.items())),
+            "candidates": cells,
+            "top_patterns": {
+                form: sorted(rows, key=lambda r: (-r["repository_count"], _canonical(r)))[:10] for form, rows in sorted(tops.items())
+            },
+        },
+    )
+
+
+def pairs_from_rows(
+    rows: Sequence[Mapping[str, Any]], excerpt_text: Callable[[str], str | None], *, profile: str = "static"
+) -> tuple[list[Pair], Counter[str], int]:
+    """Never pair on repository group alone. Legacy rows use an unambiguous candidate.
+
+    Explicit pair/source_ref keys take precedence; repeated revisions/profiles and
+    missing sides are counted, not guessed. Delta rows duplicate pin evidence.
+    """
+    from .labels import load_sources
+
+    allowed = {source.id for source in load_sources().sources}
+    buckets: dict[tuple[str, str, str, str, str, str], list[Mapping[str, Any]]] = defaultdict(list)
+    skips: Counter[str] = Counter()
+    for row in rows:
+        if row.get("profile", "static") != profile or row.get("unit", "pin") != "pin":
+            continue
+        source = str(row.get("source", ""))
+        if source not in allowed or row.get("origin") == "user" or row.get("conditional"):
+            skips["excluded_source"] += 1
+            continue
+        repo, group = str(row.get("repo") or ""), str(row.get("group") or "")
+        candidate = str(row.get("candidate") or "")
+        if not repo or not group or "/" not in repo or not candidate:
+            skips["missing_pair_identity"] += 1
+            continue
+        explicit = str(row.get("pair_id") or row.get("pair") or row.get("source_ref") or "")
+        identity = explicit or candidate
+        key = (repo_key(repo), group, source, str(row.get("language", "")), str(row.get("family", "")), identity)
+        buckets[key].append(row)
+    pairs: list[Pair] = []
+    for key, members in sorted(buckets.items()):
+        sides = {label: [r for r in members if r.get("label") == label] for label in (0, 1)}
+        if len(members) != 2 or any(len(side) != 1 for side in sides.values()):
+            skips["incomplete_pair" if any(not s for s in sides.values()) else "ambiguous_pair"] += 1
+            continue
+        fixed, vulnerable = sides[0][0], sides[1][0]
+        if any(r.get("pin_role", role) != role for r, role in ((fixed, "fixed"), (vulnerable, "vulnerable"))):
+            skips["label_side_mismatch"] += 1
+            continue
+        texts = [excerpt_text(str(r.get("excerpt_sha") or "")) for r in (vulnerable, fixed)]
+        if any(not text for text in texts):
+            skips["unreadable_excerpt"] += 1
+            continue
+        repo, group, source, language, family, identity = key
+        if not any(vulnerable.get(k) for k in ("pair_id", "pair", "source_ref")):
+            identity = "legacy-" + _hash(_canonical(key))
+        pairs.append(
+            Pair(
+                identity,
+                repo,
+                group,
+                language,
+                family,
+                texts[0] or "",
+                texts[1] or "",
+                source,
+                tuple(sorted(set(vulnerable.get("frameworks") or ()) | set(fixed.get("frameworks") or ()))),
+                str(vulnerable.get("package") or ""),
+                tuple(sorted(str(r["id"]) for r in members if r.get("id"))),
+            )
+        )
+    return pairs, skips, len(buckets)
+
+
+def mine_store(store: MemoryStore, *, profile: str = "static", spec_path: Path | None = None) -> MiningResult:
+    from .compile import load_spec
+    from .labels import load_sources
+
+    rows = [r.row for r in store.rows(kind="example")]
+    allowed = {source.id for source in load_sources().sources}
+    examples = [
+        Example.from_row(row)
+        for row in rows
+        if row.get("profile") == profile and row.get("source") in allowed and row.get("origin") != "user" and not row.get("conditional")
+    ]
+    spec = load_spec(spec_path) if spec_path else load_spec()
+    split = compile_split(examples, fraction=float(spec["split"].get("compile_fraction", 0.25)), seed=spec.seed)
+    folds = outer_folds(examples, seed=spec.seed, exclude=split.groups)
+    read_bytes = 0
+    blobs = 0
+
+    def excerpt_text(sha: str) -> str | None:
+        nonlocal read_bytes, blobs
+        if not re.fullmatch(r"[0-9a-f]{64}", sha):
+            return None
+        data = store.get_blob("excerpts", sha)
+        if data is None:
+            return None
+        read_bytes += len(data)
+        blobs += 1
+        try:
+            return data.decode("utf-8")
+        except UnicodeDecodeError:
+            return None
+
+    pairs, skips, pairs_read = pairs_from_rows(rows, excerpt_text, profile=profile)
+    result = mine_pairs(pairs, folds=folds)
+    combined = Counter(result.record["skips"]) + skips
+    result.record.update(
+        {
+            "rows_read": len(rows),
+            "pairs_read": pairs_read,
+            "complete_pairs": len(pairs),
+            "excerpt_blobs_read": blobs,
+            "excerpt_bytes_read": read_bytes,
+            "skips": dict(sorted(combined.items())),
+            "folds_digest": folds_digest(split, folds),
+            "compile_spec_sha256": spec.sha256,
+            "fold_seed": spec.seed,
+            "profile": profile,
+            "status": "mined" if result.record["pairs_mined"] else "no_mineable_pairs",
+        }
+    )
+    return result
+
+
+def write_result(result: MiningResult, out: Path, *, root: Path, results: Path) -> None:
+    """Counts-only checked-in record; candidate evidence must stay under results outside the repo."""
+    out, root, results = out.resolve(), root.resolve(), results.resolve()
+    if out.is_relative_to(root) or not out.is_relative_to(results):
+        raise ValueError("--out must be under the results root and outside the repository")
+    record = root / RECORD
+    if not record.resolve().is_relative_to(root):
+        raise ValueError("measurement record must stay inside the repository")
+    out.parent.mkdir(parents=True, exist_ok=True)
+    out.write_text("".join(_canonical(candidate) + "\n" for candidate in result.candidates), encoding="utf-8")
+    record.parent.mkdir(parents=True, exist_ok=True)
+    record.write_text(json.dumps(result.record, indent=2, sort_keys=True) + "\n", encoding="utf-8")

src/openultrasast/learn/sources.toml

@@ -82,6 +82,18 @@ recorded = "2026-10-03"
 status = "development"
 exclude_tiers = ["title"]
 
+# Batch 3: more advisory fixes whose function survives the fix, weighted toward injection and untrusted_destination.
+# Same slice, cache layout and checks; screening record
+# benchmarks/measurements/2026-10-04-decision-engine-harvest-3/record.json.
+[[source]]
+id = "advisory-fixes-3"
+kind = "pairs"
+file = "benchmarks/pairs/advisory-fixes-3/catalog.toml"
+evaluation = ["benchmarks/pairs/advisory-fixes-3/verification.json"]
+recorded = "2026-10-04"
+status = "development"
+exclude_tiers = ["title"]
+
 [[source]]
 id = "adjudications"
 kind = "adjudications"

src/openultrasast/model/partitions.py

@@ -19,7 +19,7 @@ from .layout import is_test_path, is_vendored, layout_facts
 FRONTENDS = {
     "php": "php",
     "javascript": "javascript",
-    "typescript": "javascript",
+    "typescript": "typescript",
     "python": "python",
     "java": "java",
     "c": "c",
@@ -109,19 +109,14 @@ class PartitionGraph:
 
     def _dispatch(self, kind: str, requests: Mapping[str, Mapping[str, object]], method: str) -> dict[str, list[object]]:
         answers: dict[str, list[object]] = {}
-
-        def graph_language(path: str) -> str:
-            language = self.path_languages.get(path, "")
-            return "javascript" if language == "typescript" else language
-
-        languages = dict.fromkeys(graph_language(str(params.get("file", ""))) for params in requests.values())
+        languages = dict.fromkeys(self.path_languages.get(str(params.get("file", "")), "") for params in requests.values())
         for language in languages:
             graph = self.graphs.get(language)
             if graph is None:
                 continue
             if self.budget is not None and time.monotonic() >= self.budget.deadline_monotonic:
                 break
-            selected = {rid: params for rid, params in requests.items() if graph_language(str(params.get("file", ""))) == language}
+            selected = {rid: params for rid, params in requests.items() if self.path_languages.get(str(params.get("file", ""))) == language}
             if not selected:
                 continue
             batch = getattr(graph, method, None) or getattr(graph, "run_batch", None)
@@ -138,12 +133,7 @@ class PartitionGraph:
                     # Carried, not normalized: it names request ids and costs, never a path.
                     answers.setdefault(TIMING, []).extend(rows)
                 elif rid == "__census__" and isinstance(rows, list):
-                    expected = {
-                        path
-                        for p in self.partitions
-                        if p.language == language or (language == "javascript" and p.language == "typescript")
-                        for path in p.paths
-                    }
+                    expected = set(next(p.paths for p in self.partitions if p.language == language))
                     normalized = self._normalize(rows)
                     names = [r.get("file_names") for r in normalized if isinstance(r, Mapping)]
                     named = bool(names) and all(isinstance(n, list) and all(isinstance(p, str) for p in n) for n in names)
@@ -244,17 +234,10 @@ def build_partitions(
                     graph.path_languages[relative] = languages[0]
                 elif languages:
                     boundaries.add("ambiguous_frontend_path")
-        # JavaScript and TypeScript share one frontend input tree and graph. Keep
-        # original per-file languages for scoped coverage and property limitations.
-        groups: dict[str, list[str]] = {}
-        for language in paths:
-            key = "javascript" if language == "typescript" else language
-            groups.setdefault(key, []).append(language)
-        if len(groups) > 1:
+        if len(paths) > 1:
             boundaries.add("cross_partition_semantics_unresolved")
         records: list[PartitionCoverage] = []
-        for language, members in sorted(groups.items()):
-            sources = [source for member in members for source in paths[member]]
+        for language, sources in sorted(paths.items()):
             relative_paths = tuple(sorted(p.relative_to(root).as_posix() for p in sources))
             frontend = FRONTENDS.get(language)
             if frontend is None:
@@ -264,7 +247,7 @@ def build_partitions(
                 continue
             scratch = Path(tempfile.mkdtemp(prefix="ousast-input-"))
             graph.roots[language] = scratch
-            sizes: dict[str, int] = {}
+            size = 0
             complete = True
             for source in sources:
                 if budget is not None and time.monotonic() >= budget.deadline_monotonic:
@@ -285,31 +268,25 @@ def build_partitions(
                             complete = False
                             break
                         output.write(chunk)
-                        relative = source.relative_to(root).as_posix()
-                        sizes[relative] = sizes.get(relative, 0) + len(chunk)
+                        size += len(chunk)
                 if not complete:
                     break
             built = build(scratch, frontend) if complete else None
             if built is not None:
                 graph.graphs[language] = built
                 graph.unparsed += tuple(graph._normalize(list(getattr(built, "unparsed", ()))))
-            for language in sorted(members):
-                relative_paths = tuple(sorted(p.relative_to(root).as_posix() for p in paths[language]))
-                source_size = sum(sizes.get(p, 0) for p in relative_paths)
-                record_limits: tuple[str, ...] = ("runtime_unspecified",) if language in {"javascript", "typescript"} else ()
-                if language == "typescript":
-                    record_limits += ("typescript_property_support_unvalidated",)
-                    boundaries.add("typescript_property_support_unvalidated")
-                    # About TypeScript answers. YesWiki is a PHP application that ships a few `.ts` files, and this
-                    # boundary made all 1,128 of its PHP questions unresolved.
-                    owned_languages.setdefault("typescript_property_support_unvalidated", set()).add("typescript")
-                if language in {"c", "cpp"}:
-                    record_limits += ("c_bounds_arithmetic_unmodeled",)
-                records.append(
-                    PartitionCoverage(
-                        language, frontend, relative_paths, source_size, "built" if built is not None else "build_failed", record_limits
-                    )
-                )
+            record_limits: tuple[str, ...] = ("runtime_unspecified",) if language in {"javascript", "typescript"} else ()
+            if language == "typescript":
+                record_limits += ("typescript_property_support_unvalidated",)
+                boundaries.add("typescript_property_support_unvalidated")
+                # About TypeScript answers. YesWiki is a PHP application that ships a few `.ts` files, and this
+                # boundary made all 1,128 of its PHP questions unresolved.
+                owned_languages.setdefault("typescript_property_support_unvalidated", set()).add("typescript")
+            if language in {"c", "cpp"}:
+                record_limits += ("c_bounds_arithmetic_unmodeled",)
+            records.append(
+                PartitionCoverage(language, frontend, relative_paths, size, "built" if built is not None else "build_failed", record_limits)
+            )
         from dataclasses import replace
 
         graph.partitions = tuple(

src/openultrasast/model/regions.py

@@ -24,7 +24,6 @@ from collections.abc import Mapping, Sequence
 from dataclasses import dataclass, replace
 from pathlib import Path
 
-from ..semantic.facts import TAINT_LANGUAGE_ALIAS
 from .contracts import AffectedRelationship, ChangeContext, ExecutionBudget, QuestionIdentity
 from .specs import config_specs, dominance_specs, taint_specs
 
@@ -42,7 +41,7 @@ _ACCESS_RANK: dict[str, float] = {
 # Languages whose facts the model layer carries. Anything else yields no region rather than an empty scan
 # that looks like a clean bill of health.
 _LANGUAGES = {"python", "javascript", "typescript", "java", "c", "c_cpp", "php"}
-_NORMALISE = {**TAINT_LANGUAGE_ALIAS, "c_cpp": "c"}
+_NORMALISE = {"typescript": "javascript", "c_cpp": "c"}
 
 # What a CPG calls a file's top-level code. A module body is a method like any other to the engine, and
 # naming it here is what lets a region ask about the code that is not inside any function.

src/openultrasast/model/specs.py

@@ -21,7 +21,7 @@ import re
 from collections.abc import Mapping
 from dataclasses import dataclass, field
 
-from ..semantic.facts import TAINT_LANGUAGE_ALIAS, SemanticFacts, load_facts
+from ..semantic.facts import SemanticFacts, load_facts
 from ..semantic.obligations.facts import ObligationFacts, load_obligation_facts
 from .taxonomy import FamilyTaxonomy, load_families
 
@@ -245,10 +245,7 @@ def taint_specs(
     Sources and sanitizers are language-wide: a family does not get its own notion of what untrusted input is,
     and a sanitizer that breaks one flow breaks it whatever the sink's family.
     """
-    facts = facts if facts is not None else load_facts()
-    if language in TAINT_LANGUAGE_ALIAS:
-        language = facts.taint_language(language)
-    scoped = facts.for_language(language)
+    scoped = (facts if facts is not None else load_facts()).for_language(language)
     families = taxonomy if taxonomy is not None else load_families()
     sources = tuple(sorted({pattern for fact in scoped.sources for pattern in fact.patterns}))
 

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,

src/openultrasast/pairs.py

@@ -39,7 +39,7 @@ from .preprocess import FileTarget, preprocess_repository
 from .rank import rank_targets
 from .ruleset import PatternRule
 from .semantic import OverlayRecord, adjudicate
-from .semantic.facts import TAINT_LANGUAGE_ALIAS, FactLoadError, load_facts
+from .semantic.facts import FactLoadError, load_facts
 from .semantic.functions import RESERVED_FUNCTION_LABELS, function_at, named_function_ranges, spans_named
 from .semantic.obligations.facts import OPERATION_KINDS
 
@@ -1071,7 +1071,7 @@ def _sink_ids_by_rule() -> dict[tuple[str, str], frozenset[str]]:
 
 
 def _fact_language(language: str) -> str:
-    return {"c_cpp": "c", "cpp": "c", **TAINT_LANGUAGE_ALIAS}.get(language, language)
+    return {"c_cpp": "c", "cpp": "c", "typescript": "javascript"}.get(language, language)
 
 
 def _in_named_function(record: OverlayRecord, function: str, ranges: FunctionRanges | None) -> bool:

src/openultrasast/plane/memory.py

@@ -586,6 +586,10 @@ class ObjectClient(Protocol):
 
 _VERIFIED: set[str] = set()
 PROBE_KEY = "_probe/select.jsonl"
+# Live finding: files.because-security.com denies presigned GETs ending in .tar, .gz,
+# .zip or .bin (HTTP 403 AccessDenied). Use .dat for archive objects; bytes stay tar.
+PRESIGNED_ARCHIVE_SUFFIX = ".dat"  # engine executors only; the plane keeps output.tar
+
 RUNS_PREFIX = "runs/"
 RUNS_RULE_ID = "ousast-runs-expiry"
 RUNS_EXPIRE_DAYS = 30

src/openultrasast/plane/profile.py

@@ -1,6 +1,6 @@
 """The plane's configuration surface: one ``PlaneProfile`` per cluster (plane-on-kubernetes design section 1).
 
-A profile is a TOML file under ``ops/k8s/profiles/`` (``kind.toml`` for the laptop's kind cluster, ``k3s.toml`` for
+A profile is a TOML file under ``ops/k8s/profiles/`` (``kind.toml`` for the server VM's kind cluster, ``k3s.toml`` for
 the production cluster) holding every value that differs between clusters: the execution mode, the kube context,
 the registry and the digest-pinned images file, the router and ax-server addresses (empty: tunnel through the
 Kubernetes API), the memory store, the atespace, ax's snapshot bucket, the worker pools and the image pull secret.
@@ -59,6 +59,7 @@ class PlaneProfile:
     name: str
     path: Path
     exec: str = "local"
+    kind_observability: bool = False
     kube_context: str = ""
     registry: str = ""
     images: Path = Path("images.json")
@@ -135,6 +136,10 @@ def _reject_secret(where: str, key: str, value: object, declared: bool = False)
 
 
 def _coerce(where: str, key: str, value: object, kind: str) -> Any:
+    if kind == "bool":
+        if not isinstance(value, bool):
+            raise ProfileError(f"{where}: {key} must be a boolean, got {value!r}")
+        return value
     if kind == "int":
         if isinstance(value, bool) or not isinstance(value, int):
             raise ProfileError(f"{where}: {key} must be an integer, got {value!r}")
@@ -172,6 +177,10 @@ def _env_overrides(environ: Mapping[str, str], where: str, values: dict[str, Any
         raw = environ.get(_ENV_PREFIX + name.upper())
         if raw is not None and raw != "":
             value: Any = raw
+            if kind == "bool":
+                if raw.lower() not in ("true", "false", "1", "0"):
+                    raise ProfileError(f"{where}: {_ENV_PREFIX}{name.upper()} must be true/false or 1/0, got {raw!r}")
+                value = raw.lower() in ("true", "1")
             if kind == "int":
                 try:
                     value = int(raw)
@@ -241,7 +250,8 @@ def _lookup(profile: PlaneProfile, dotted: str) -> str:
         return str(getattr(profile.pools[pool_name], key))
     if head not in _SCALARS or rest:
         raise ProfileError(f"{profile.path}: no such field {dotted!r}")
-    return str(getattr(profile, head))
+    value = getattr(profile, head)
+    return str(value).lower() if isinstance(value, bool) else str(value)
 
 
 def main(argv: list[str] | None = None) -> int:

src/openultrasast/plane/tasks/alerts.py

@@ -44,7 +44,6 @@ from ...mapping import analyze_entry_points, attach_reachability_hints
 from ...preprocess import detect_language, preprocess_repository
 from ...rank import rank_targets
 from ...ruleset import DEFAULT_RULESET_DIR, load_ruleset
-from ...semantic.facts import TAINT_LANGUAGE_ALIAS
 from .repo_facts import DECLARATION, GLOBAL, enclosing
 
 ROLES = ("vulnerable", "fixed")
@@ -65,9 +64,8 @@ def quick_languages(directory: Path = DEFAULT_RULESET_DIR) -> frozenset[str]:
 
 
 def engine_languages(directory: Path = DEFAULT_RULESET_DIR) -> frozenset[str]:
-    """Languages with a semantic fact table, including aliases of available tables."""
-    languages = {path.stem for path in (directory / "semantic").glob("*.toml")}
-    return frozenset(languages | {alias for alias, target in TAINT_LANGUAGE_ALIAS.items() if target in languages})
+    """The languages the deterministic engine has a semantic model for: one ``semantic/<language>.toml`` each."""
+    return frozenset(path.stem for path in (directory / "semantic").glob("*.toml"))
 
 
 def covers(covered: Iterable[str], language: str) -> bool:

src/openultrasast/preprocess.py

@@ -22,8 +22,6 @@ LANGUAGE_BY_EXTENSION = {
     ".cjs": "javascript",
     ".ts": "typescript",
     ".tsx": "typescript",
-    ".mts": "typescript",
-    ".cts": "typescript",
     ".go": "go",
     ".rs": "rust",
     ".java": "java",

src/openultrasast/semantic/cst.py

@@ -43,6 +43,14 @@ _KIND = {
 _KIND["cpp"] = _KIND["c"]
 _KIND["typescript"] = _KIND["javascript"]
 _KIND["tsx"] = _KIND["javascript"]
+_KIND["php"] = {
+    "function": frozenset({"function_definition", "method_declaration"}),
+    "assign": frozenset({"assignment_expression", "augmented_assignment_expression"}),
+    "call": frozenset({"function_call_expression", "member_call_expression", "scoped_call_expression"}),
+    "args": frozenset({"arguments"}),
+    "string": frozenset({"string", "encapsed_string"}),
+    "params": frozenset({"formal_parameters"}),
+}
 
 _CONST = frozenset(
     {
@@ -183,6 +191,11 @@ def _call(node: Any, kinds: dict[str, frozenset[str]]) -> CallSite:
 
 
 def _callee_name(node: Any, args_node: Any) -> str:
+    if node.type in {"member_call_expression", "scoped_call_expression"}:
+        receiver = node.child_by_field_name("object") or node.child_by_field_name("scope")
+        name = node.child_by_field_name("name")
+        if receiver is not None and name is not None:
+            return f"{_text(receiver)}->{_text(name)}"
     idents = [
         _text(child)
         for child in getattr(node, "named_children", ()) or ()
@@ -207,7 +220,7 @@ def _call_name(node: Any, kinds: dict[str, frozenset[str]]) -> str | None:
 
 def _function_name(node: Any) -> str:
     for child in getattr(node, "named_children", ()) or ():
-        if child.type == "identifier":
+        if child.type in {"identifier", "name"}:
             return _text(child)
         if child.type == "function_declarator":
             for grandchild in getattr(child, "named_children", ()) or ():
@@ -240,7 +253,7 @@ def _left_name(node: Any) -> str:
 def _identifiers(node: Any) -> tuple[str, ...]:
     names: list[str] = []
     for child in _walk(node):
-        if child.type == "identifier":
+        if child.type in {"identifier", "variable_name"}:
             names.append(_text(child))
     return tuple(names)
 

src/openultrasast/semantic/extra.py

@@ -17,6 +17,7 @@ _GRAMMAR_MODULES = {
     "c": ("tree_sitter_c", "language"),
     "cpp": ("tree_sitter_cpp", "language"),
     "java": ("tree_sitter_java", "language"),
+    "php": ("tree_sitter_php", "language_php"),
 }
 _PROBE_ENV = "OPENULTRASAST_TREE_SITTER_PROBE"
 _PROBE_OFF = frozenset({"0", "off", "false", "unavailable", "no"})

src/openultrasast/semantic/facts.py

@@ -15,14 +15,12 @@ from ..ruleset.frameworks import Priors, kept, normalize_priors, read_tag, tag_o
 
 DEFAULT_FACTS_DIR = Path(__file__).resolve().parents[1] / "ruleset" / "semantic"
 
-TAINT_LANGUAGE_ALIAS = {"typescript": "javascript"}
-
 _LANGUAGE_ALIASES = {
     "c": "c",
     "cpp": "c",
     "python": "python",
     "javascript": "javascript",
-    **TAINT_LANGUAGE_ALIAS,
+    "typescript": "javascript",
     "java": "java",
 }
 
@@ -147,18 +145,8 @@ class SemanticFacts:
     dispatches: tuple[DispatchFact, ...] = ()
     layouts: tuple[LayoutFact, ...] = ()
 
-    def taint_language(self, language: str) -> str:
-        """Prefer native facts when supplied; otherwise use the shared language aliases."""
-        if language in TAINT_LANGUAGE_ALIAS and any(
-            item.language == language
-            for items in (self.sources, self.sinks, self.sanitizers, self.dispatches, self.layouts)
-            for item in items
-        ):
-            return language
-        return _LANGUAGE_ALIASES.get(language, language)
-
     def for_language(self, language: str) -> SemanticFacts:
-        key = self.taint_language(language)
+        key = _LANGUAGE_ALIASES.get(language, language)
         return SemanticFacts(
             version=self.version,
             sources=tuple(item for item in self.sources if item.language == key),

tests/test_advisory_fix_pairs.py

@@ -21,8 +21,9 @@ from openultrasast.pairs import load_pair_catalog
 
 ROOT = Path(__file__).resolve().parents[1]
 PAIRS = ROOT / "benchmarks" / "pairs"
-# batch 1 (2026-10-01) and batch 2 (2026-10-03, functions that survive the fix): one source each, same rules
-BATCHES = {"advisory-fixes": PAIRS / "advisory-fixes", "advisory-fixes-2": PAIRS / "advisory-fixes-2"}
+# batch 1 (2026-10-01), batches 2 (2026-10-03) and 3 (2026-10-04, functions that survive the fix): one source each,
+# same rules
+BATCHES = {name: PAIRS / name for name in ("advisory-fixes", "advisory-fixes-2", "advisory-fixes-3")}
 SHA = re.compile(r"[0-9a-f]{40}")
 PERMISSIVE_OR_COPYLEFT = re.compile(
     r"^(MIT|BSD-[23]-Clause|Apache-2\.0|ISC|Zlib|0BSD|Unlicense|MPL-[12]\.[01]"
@@ -106,10 +107,18 @@ def test_no_pair_repeats_a_repository_the_benchmarks_already_name(source_id: str
 PERMISSIVE = {"MIT", "BSD-2-Clause", "BSD-3-Clause", "Apache-2.0", "ISC", "MPL-2.0", "Unlicense"}
 
 
-def test_batch_two_is_permissive_and_dated_against_the_cutoff() -> None:
-    """Batch 2 publishes permissive licences only and records each advisory's publication against the 2026-06-01
+@pytest.mark.parametrize(("source_id", "batch"), [("advisory-fixes-2", "2026-10-03"), ("advisory-fixes-3", "2026-10-04")])
+def test_later_batches_are_permissive_and_dated_against_the_cutoff(source_id: str, batch: str) -> None:
+    """Batches 2 and 3 publish permissive licences only and record each advisory's publication against the 2026-06-01
     cutoff, so memorisation can be measured on the post-cutoff pairs."""
-    for item in _raw(BATCHES["advisory-fixes-2"]):
+    for item in _raw(BATCHES[source_id]):
         assert item["license"] in PERMISSIVE, item["name"]
-        assert item["batch"] == "2026-10-03", item["name"]
+        assert item["batch"] == batch, item["name"]
         assert item["post_cutoff"] is (item["published_at"][:10] >= "2026-06-01"), item["name"]
+
+
+def test_no_repository_repeats_across_batches() -> None:
+    seen: Counter[str] = Counter()
+    for folder in BATCHES.values():
+        seen.update({repo_name(item["repo"]) for item in _raw(folder)})
+    assert [repo for repo, n in seen.items() if n > 1] == []

tests/test_ax_executor.py

@@ -0,0 +1,406 @@
+"""AX transport controls using an executable shim and an in-memory object store."""
+
+import io
+import json
+import subprocess
+import sys
+import tarfile
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+with pytest.MonkeyPatch.context() as patch:
+    patch.syspath_prepend(str(Path(__file__).resolve().parents[1] / "benchmarks/learn"))
+    import engine_trace_ax as ax
+
+
+IMAGE = "ghcr.io/norandom/ousast-engine-task@sha256:" + "a" * 64
+PIN = {"repo": "owner/repo", "pin": "123", "units": [{"unit": "a", "supported": True}], "questions": {"a": {}}}
+
+
+def archive(record):
+    buffer = io.BytesIO()
+    with tarfile.open(fileobj=buffer, mode="w") as tar:
+        data = json.dumps(record).encode()
+        info = tarfile.TarInfo("result.json")
+        info.size = len(data)
+        tar.addfile(info, io.BytesIO(data))
+    return buffer.getvalue()
+
+
+class Store:
+    def __init__(self, result=None):
+        self.result = result
+        self.objects = {}
+        self.deleted = []
+        self.expiries = []
+
+    def _put(self, key, data):
+        self.objects[key] = data
+
+    def _get(self, key):
+        return (self.result, None) if self.result else None
+
+    def _delete(self, key):
+        self.deleted.append(key)
+        self.objects.pop(key, None)
+
+    def presign_get(self, key, expiry):
+        self.expiries.append(expiry.total_seconds())
+        return "https://files.because-security.com/bucket/" + key + "?signature=secret"
+
+    presign_put = presign_get
+
+
+@pytest.fixture
+def setup(tmp_path, monkeypatch):
+    log = tmp_path / "calls.jsonl"
+    shim = tmp_path / "fake-ax"
+    shim.write_text("#!/bin/sh\nexec " + sys.executable + " " + str(tmp_path / "shim.py") + ' "$@"\n')
+    shim.chmod(0o755)
+    (tmp_path / "shim.py").write_text("""import json, os, sys
+from pathlib import Path
+args = sys.argv[1:]
+with Path(os.environ["AX_LOG"]).open("a") as log:
+    log.write(json.dumps({"args": args, "input": sys.stdin.read() if args[0] == "apply" else "",
+        "kubeconfig": os.environ.get("KUBECONFIG")}) + "\\n")
+if args[0] == "get" and os.environ.get("AX_GET_FAIL"):
+    sys.exit(1)
+if args[0] == "get":
+    applies = [json.loads(line) for line in Path(os.environ["AX_LOG"]).read_text().splitlines()
+        if json.loads(line)["args"][0] == "apply"]
+    manifest = json.loads(applies[-1]["input"])
+    phases = os.environ.get("AX_PHASE", "Running").split(",")
+    phase = phases[min(len(applies) - 1, len(phases) - 1)]
+    print(json.dumps({"items": [{"metadata": manifest["metadata"], "status": {
+        "phase": phase, "workerIP": "10.0.0.2"}}] if phase != "Missing" else []}))
+if args[0] == "delete" and os.environ.get("AX_DELETE_FAIL"):
+    sys.exit(1)
+""")
+    monkeypatch.setenv("AX_LOG", str(log))
+    monkeypatch.setenv("S3_ENDPOINT", "https://files.because-security.com")
+    args = SimpleNamespace(
+        ax_bin=str(shim), kubeconfig="/operator/kubeconfig", atespace="default", image=IMAGE, deadline=4, question_deadline=2
+    )
+    source = tmp_path / "case"
+    source.mkdir()
+    (source / "app.py").write_text("print(42)\n")
+    output = tmp_path / "out"
+    output.mkdir()
+    now = [0]
+
+    def pause(seconds):
+        now[0] += seconds
+
+    return args, source, output, log, pause, lambda: now[0]
+
+
+def test_ax_executor_success_while_running_and_cleanup(setup):
+    args, source, output, log, pause, clock = setup
+    store = Store(archive({**PIN, "done": True}))
+    executor = ax.AX(args, store, pause=pause, clock=clock)
+    done, ip = executor.execute(source, output, PIN, "owner/repo/123")
+    assert done.returncode == 0
+    assert ip == "10.0.0.2"
+    assert json.loads((output / "result.json").read_text())["done"]
+    calls = [json.loads(line) for line in log.read_text().splitlines()]
+    assert calls[0]["args"] == ["apply", "-f", "-"]
+    assert calls[-1]["args"][:2] == ["delete", "task"]
+    assert calls[-1]["args"][-2:] == ["--atespace", args.atespace]
+    assert all(c["kubeconfig"] == args.kubeconfig for c in calls)
+    manifest = json.loads(calls[0]["input"])
+    assert manifest["apiVersion"] == "ax.io/v1alpha1"
+    assert manifest["kind"] == "Task"
+    assert len(manifest["metadata"]["name"]) <= 49
+    assert manifest["spec"]["image"] == IMAGE
+    env = {e["name"]: e["value"] for e in manifest["spec"]["env"]}
+    assert set(env) == {"SOURCE_URL", "QUESTIONS_URL", "RESULT_URL", "DEADLINE", "QUESTION_DEADLINE"}
+    assert len(json.dumps(manifest["spec"]["env"]).encode()) < 32768
+    assert "AWS" not in calls[0]["input"]
+    assert env["SOURCE_URL"].split("?")[0].endswith("/source.dat")
+    assert env["RESULT_URL"].split("?")[0].endswith("/result.dat")
+    assert env["QUESTIONS_URL"].split("?")[0].endswith("/questions.json")
+    assert {key.rsplit("/", 1)[-1] for key in store.deleted} == {"source.dat", "questions.json", "result.dat"}
+    assert store.expiries == [604] * 3
+    assert len(store.deleted) == 3 and not store.objects
+
+
+def test_ax_executor_timeout_deletes_task(setup):
+    args, source, output, log, pause, clock = setup
+    store = Store()
+    done, _ = ax.AX(args, store, pause=pause, clock=clock).execute(source, output, PIN, "pin")
+    assert done is None
+    assert json.loads(log.read_text().splitlines()[-1])["args"][:2] == ["delete", "task"]
+    assert len(store.deleted) == 3
+
+
+def test_ax_executor_failed_phase_deletes_task(setup, monkeypatch):
+    args, source, output, log, pause, clock = setup
+    monkeypatch.setenv("AX_PHASE", "Failed")
+    done, _ = ax.AX(args, Store(), pause=pause, clock=clock).execute(source, output, PIN, "pin")
+    assert done.returncode == 1 and done.stderr == "AX sandbox failed twice"
+    attempts = json.loads((output / "ax_attempts.json").read_text())
+    assert [a["phase"] for a in attempts] == ["Failed", "Failed"]
+    assert json.loads(log.read_text().splitlines()[-1])["args"][:2] == ["delete", "task"]
+
+
+@pytest.mark.parametrize("phase", ["Failed", "Missing"])
+def test_ax_executor_retries_with_new_name_links_and_backoff(setup, monkeypatch, phase):
+    args, source, output, log, pause, clock = setup
+    monkeypatch.setenv("AX_PHASE", phase + ",Succeeded")
+    store = Store()
+    store._get = lambda key: (archive({**PIN, "done": True}), None) if "/attempt-1/" in key else None
+    pauses = []
+
+    def backoff(seconds):
+        calls = [json.loads(line) for line in log.read_text().splitlines()]
+        assert calls[-1]["args"][:2] == ["delete", "task"]
+        pauses.append(seconds)
+        pause(seconds)
+
+    done, _ = ax.AX(args, store, pause=backoff, clock=clock).execute(source, output, PIN, "x" * 100)
+    assert done.returncode == 0
+    record = json.loads((output / "result.json").read_text())
+    attempts = record["ax_attempts"]
+    assert record["done"] and [a["phase"] for a in attempts] == [phase, "Succeeded"]
+    assert len({a["name"] for a in attempts}) == 2
+    assert all(a["name"].startswith("ousast-engine-") and len(a["name"]) <= 49 and a["seconds"] >= 0 for a in attempts)
+    assert pauses == [10]
+    calls = [json.loads(line) for line in log.read_text().splitlines()]
+    manifests = [json.loads(c["input"]) for c in calls if c["args"][0] == "apply"]
+    links = [{e["name"]: e["value"] for e in m["spec"]["env"]} for m in manifests]
+    assert all(links[0][key] != links[1][key] for key in ("SOURCE_URL", "QUESTIONS_URL", "RESULT_URL"))
+    assert len(store.expiries) == 6 and len(store.deleted) == 6
+
+
+@pytest.mark.parametrize("late", [False, True])
+def test_ax_executor_failed_result_is_never_retried(setup, monkeypatch, late):
+    args, source, output, log, pause, clock = setup
+    monkeypatch.setenv("AX_PHASE", "Failed")
+    store = Store(archive({**PIN, "done": True, "status": "failed"}))
+    original_get = store._get
+    reads = []
+
+    def get(key):
+        reads.append(key)
+        return None if late and len(reads) == 1 else original_get(key)
+
+    store._get = get
+    done, _ = ax.AX(args, store, pause=pause, clock=clock).execute(source, output, PIN, "pin")
+    record = json.loads((output / "result.json").read_text())
+    assert done.returncode == 0 and record["status"] == "failed"
+    assert len(record["ax_attempts"]) == 1
+    assert sum(json.loads(line)["args"][0] == "apply" for line in log.read_text().splitlines()) == 1
+
+
+@pytest.mark.parametrize("mode", ["ax", "mixed-ax"])
+def test_ax_sandbox_failed_twice_scheduler(setup, monkeypatch, mode):
+    with monkeypatch.context() as patch:
+        patch.syspath_prepend(str(Path(__file__).resolve().parents[1] / "benchmarks/learn"))
+        import engine_trace as trace
+
+    args, source, output, log, pause, clock = setup
+    monkeypatch.setenv("AX_PHASE", "Failed")
+    args.executor, args.parallel, args.out, args.docker_image = mode, 1, output, "vm-image"
+    args.ax_dispatcher = ax.AX(args, Store(), pause=pause, clock=clock)
+    monkeypatch.setattr(trace, "prepare_questions", lambda *_: PIN["questions"])
+
+    class Inputs:
+        def materialize(self, pin, target):
+            target.mkdir()
+            data = (source / "app.py").read_bytes()
+            (target / "app.py").write_bytes(data)
+            return {"files": 1, "bytes": len(data)}
+
+    original_run = subprocess.run
+    docker_calls = []
+
+    def run(command, **kwargs):
+        if command[0] == args.ax_bin:
+            return original_run(command, **kwargs)
+        assert command[:2] in (["docker", "run"], ["docker", "rm"])
+        docker_calls.append(command)
+        if command[1] == "run":
+            assert command[command.index("--heap-profile") + 1] == "vm"
+            assert "vm-image" in command
+            target = Path(next(arg.removesuffix(":/out") for arg in command if arg.endswith(":/out")))
+            prepared = json.loads((target / "pin.json").read_text())
+            trace.write_json(
+                target / "result.json",
+                {**prepared, "done": True, "units": [trace.unit_record(u, "asked-nothing") for u in prepared["units"]]},
+            )
+        return subprocess.CompletedProcess(command, 0, "", "")
+
+    monkeypatch.setattr(subprocess, "run", run)
+    assert trace.run_plan([PIN], source, {}, Inputs(), args) is (mode == "ax")
+    record = json.loads((output / trace.pin_name(PIN)).read_text())
+    assert [a["phase"] for a in record["ax_attempts"]] == ["Failed", "Failed"]
+    assert sum(json.loads(line)["args"][0] == "apply" for line in log.read_text().splitlines()) == 2
+    if mode == "mixed-ax":
+        assert [command[1] for command in docker_calls] == ["run", "rm"]
+        assert record["executor"] == "docker" and record["fallback"] == "vm-sandbox"
+        assert record["units"][0]["status"] == "asked-nothing"
+        assert record["ax_attempt"]["ax_attempts"] == record["ax_attempts"]
+    else:
+        assert not docker_calls
+        assert record["units"][0]["status"] == "failed"
+        assert record["units"][0]["reason"] == "AX sandbox failed twice"
+
+
+def test_ax_executor_delete_failure_is_fatal_and_still_cleans_objects(setup, monkeypatch):
+    args, source, output, _, pause, clock = setup
+    monkeypatch.setenv("AX_DELETE_FAIL", "1")
+    store = Store(archive({**PIN, "done": True}))
+    with pytest.raises(RuntimeError, match="cleanup"):
+        ax.AX(args, store, pause=pause, clock=clock).execute(source, output, PIN, "pin")
+    assert len(store.deleted) == 3
+
+
+@pytest.mark.parametrize("endpoint", ["http://files.because-security.com", "https://127.0.0.1", "https://[::1]", "https://localhost"])
+def test_ax_executor_endpoint_rejected_before_store_open(setup, monkeypatch, endpoint):
+    args = setup[0]
+    monkeypatch.setenv("S3_ENDPOINT", endpoint)
+    monkeypatch.setattr(ax, "open_store", lambda: pytest.fail("store opened before validation"))
+    with pytest.raises(ValueError, match="egress policy"):
+        ax.AX(args)
+
+
+def test_ax_executor_rejects_tag_and_large_env(setup):
+    args = setup[0]
+    urls = {key: "https://files.because-security.com/" for key in ("SOURCE_URL", "QUESTIONS_URL", "RESULT_URL")}
+    args.image = "image:latest"
+    with pytest.raises(ValueError, match="digest"):
+        ax.task_manifest("name", args, urls)
+    args.image = IMAGE
+    urls["SOURCE_URL"] += "x" * 32768
+    with pytest.raises(ValueError, match="32 KB"):
+        ax.task_manifest("name", args, urls)
+
+
+def test_ax_executor_rejects_mismatched_result(setup):
+    args, source, output, _, pause, clock = setup
+    store = Store(archive({"done": True, "units": []}))
+    done, _ = ax.AX(args, store, pause=pause, clock=clock).execute(source, output, PIN, "pin")
+    assert done.returncode == 1
+    assert not (output / "result.json").exists()
+
+
+def test_ax_executor_table_worker_ip():
+    assert ax.task_state("NAME ATESPACE PHASE WORKER-IP\nours default Running 10.2.3.4\n", "ours") == ("Running", "10.2.3.4")
+
+
+def test_ax_executor_missing_requires_readable_task_list():
+    assert ax.task_state('{"items": []}', "ours") == ("Missing", None)
+    assert ax.task_state("NAME ATESPACE PHASE WORKER-IP\n", "ours") == ("Missing", None)
+    assert ax.task_state("", "ours") == ("", None)
+    assert ax.task_state("temporarily unavailable", "ours") == ("", None)
+
+
+def test_ax_executor_status_query_failure_is_not_disappearance(setup, monkeypatch):
+    args, source, output, log, pause, clock = setup
+    monkeypatch.setenv("AX_GET_FAIL", "1")
+    done, _ = ax.AX(args, Store(), pause=pause, clock=clock).execute(source, output, PIN, "pin")
+    assert done is None
+    assert sum(json.loads(line)["args"][0] == "apply" for line in log.read_text().splitlines()) == 1
+
+
+def test_ax_executor_entry_download_failure_restores_host_units(setup):
+    args, source, output, _, pause, clock = setup
+    store = Store(archive({"done": True, "status": "failed", "reason": "questions download failed", "units": []}))
+    done, _ = ax.AX(args, store, pause=pause, clock=clock).execute(source, output, PIN, "pin")
+    assert done.returncode == 0
+    unit = json.loads((output / "result.json").read_text())["units"][0]
+    assert unit["unit"] == "a" and unit["status"] == "failed"
+    assert unit["reason"] == "questions download failed"
+
+
+def test_ax_executor_presign_host_must_match_egress_policy(setup):
+    args, source, output, log, pause, clock = setup
+    store = Store()
+    store.presign_put = lambda key, expiry: "https://another-store.example/result"
+    done, _ = ax.AX(args, store, pause=pause, clock=clock).execute(source, output, PIN, "pin")
+    assert done.returncode == 1 and "egress policy" in done.stderr
+    assert not log.exists()
+    assert len(store.deleted) == 3
+
+
+def test_ax_executor_get_failure_does_not_lose_result(setup, monkeypatch):
+    args, source, output, _, pause, clock = setup
+    monkeypatch.setenv("AX_GET_FAIL", "1")
+    done, _ = ax.AX(args, Store(archive({**PIN, "done": True})), pause=pause, clock=clock).execute(source, output, PIN, "pin")
+    assert done.returncode == 0
+
+
+def test_ax_executor_absent_worker_ip():
+    assert ax.task_state("NAME ATESPACE PHASE ACTOR WORKER-IP AGE\nours other Pending actor <none> 1s\n", "ours") == ("Pending", None)
+
+
+def test_ax_executor_late_store_result_times_out(setup):
+    args, source, output, log, pause, clock = setup
+    store = Store(archive({**PIN, "done": True}))
+    original_get = store._get
+
+    def late_get(key):
+        pause(args.deadline)
+        return original_get(key)
+
+    store._get = late_get
+    done, _ = ax.AX(args, store, pause=pause, clock=clock).execute(source, output, PIN, "pin")
+    assert done is None
+    assert not (output / "result.json").exists()
+    assert json.loads(log.read_text().splitlines()[-1])["args"][:2] == ["delete", "task"]
+
+
+def test_ax_executor_late_status_times_out(setup):
+    args, source, output, _, pause, clock = setup
+    executor = ax.AX(args, Store(archive({**PIN, "done": True})), pause=pause, clock=clock)
+    original_ax = executor.ax
+
+    def late_ax(*command, **kwargs):
+        result = original_ax(*command, **kwargs)
+        if command[0] == "get":
+            pause(args.deadline)
+        return result
+
+    executor.ax = late_ax
+    done, _ = executor.execute(source, output, PIN, "pin")
+    assert done is None
+    assert not (output / "result.json").exists()
+
+
+def test_ax_executor_blocking_store_read_cannot_delay_task_deletion(setup):
+    import threading
+    import time
+
+    args, source, output, log, _, _ = setup
+    args.deadline = 0.15
+    release = threading.Event()
+    store = Store()
+    store._get = lambda key: release.wait(5)
+    started = time.monotonic()
+    try:
+        done, _ = ax.AX(args, store).execute(source, output, PIN, "pin")
+        assert done is None
+        assert time.monotonic() - started < 2
+        assert json.loads(log.read_text().splitlines()[-1])["args"][:2] == ["delete", "task"]
+    finally:
+        release.set()
+
+
+def test_ax_task_names_reserve_watcher_prefix_and_fit_actor_limit(setup):
+    args, source, output, log, pause, clock = setup
+    executor = ax.AX(args, Store(archive({**PIN, "done": True})), pause=pause, clock=clock)
+    # Sanitization and truncation must not collapse distinct pins.
+    for pin_id in ["A/B", "a-b", "x" * 100 + "1", "x" * 100 + "2", "☃"]:
+        done, _ = executor.execute(source, output, PIN, pin_id)
+        assert done.returncode == 0
+    manifests = [json.loads(call["input"]) for line in log.read_text().splitlines() if (call := json.loads(line))["args"][0] == "apply"]
+    names = [manifest["metadata"]["name"] for manifest in manifests]
+    assert len(set(names)) == 5
+    for manifest, name in zip(manifests, names, strict=True):
+        assert manifest["kind"] == "Task"
+        assert manifest["metadata"]["atespace"] == "default"
+        assert name.startswith("ousast-engine-")
+        assert len(name.encode()) <= 49
+        assert len((name + "-tmpl-12345678").encode()) <= 63

tests/test_ax_lean.py

@@ -0,0 +1,130 @@
+"""Offline overlay contract and installer staging; no Kubernetes executable needed."""
+
+from __future__ import annotations
+
+import runpy
+from pathlib import Path
+
+import pytest
+import yaml
+
+ROOT = Path(__file__).resolve().parents[1]
+OVERLAY = ROOT / "ops/ax/substrate-kind-lean"
+DISABLED = {
+    "OTEL_TRACES_EXPORTER": "none",
+    "OTEL_METRICS_EXPORTER": "none",
+    "OTEL_LOGS_EXPORTER": "none",
+    "OTEL_SDK_DISABLED": "true",
+}
+
+
+def test_ax_lean_overlay_removes_observability_and_replaces_export_config() -> None:
+    overlay = yaml.safe_load((OVERLAY / "kustomization.yaml").read_text())
+    assert overlay["resources"] == ["../kind-upstream"]
+    deletes = [p for p in overlay["patches"] if "target" in p]
+    assert [p["target"] for p in deletes] == [
+        {"namespace": "otel-system"},
+        {"kind": "Namespace", "name": "otel-system"},
+        {"kind": "ClusterRole", "name": "ate-prometheus"},
+        {"kind": "RoleBinding", "name": "ate-prometheus", "namespace": "ate-system"},
+    ]
+    assert all(yaml.safe_load(p["patch"])["$patch"] == "delete" for p in deletes)
+    config = yaml.safe_load((OVERLAY / overlay["patches"][-1]["path"]).read_text())
+    assert config["metadata"] == {"name": "ate-otel-config", "namespace": "ate-system"}
+    # Whole-data replacement drops the upstream endpoint, interval and timeout,
+    # including any future keys that would otherwise survive a merge.
+    assert config["data"] == {"$patch": "replace", **DISABLED}
+
+
+def test_ax_staging_preserves_upstream_and_installer_prerequisites(tmp_path: Path) -> None:
+    prepare = runpy.run_path(str(ROOT / "ops/ax/prepare-substrate.py"))["prepare"]
+    source = tmp_path / "source"
+    kind = source / "manifests/ate-install/kind"
+    kind.mkdir(parents=True)
+    original = "resources: [otel-collector.yaml, prometheus.yaml, ate-otel-config.yaml]\n"
+    (kind / "kustomization.yaml").write_text(original)
+    (kind / "ate-otel-config.yaml").write_text("upstream config")
+    (kind / "postgres").mkdir()
+    (kind / "postgres/kustomization.yaml").write_text("resources: [../../postgres]\n")
+    (source / "go.mod").write_text("module test\n")
+    (source / "hack").mkdir()
+    (source / "hack/install-ate-kind.sh").write_text("upstream installer")
+    stage = tmp_path / "stage"
+    prepare(source, stage)
+    install = stage / "manifests/ate-install"
+    assert (install / "kind-upstream/kustomization.yaml").read_text() == original
+    assert (install / "kind/postgres/kustomization.yaml").read_text() == "resources: [../../postgres]\n"
+    assert (kind / "kustomization.yaml").read_text() == original
+    assert (kind / "ate-otel-config.yaml").read_text() == "upstream config"
+    assert (stage / "hack/install-ate-kind.sh").read_text() == "upstream installer"
+    assert (stage / "go.mod").is_symlink()
+    config = yaml.safe_load((install / "kind/ate-otel-config.yaml").read_text())
+    assert config["data"] == DISABLED  # direct installer apply must not contain $patch
+    overlay = yaml.safe_load((install / "kind/kustomization.yaml").read_text())
+    patch = yaml.safe_load((install / "kind" / overlay["patches"][-1]["path"]).read_text())
+    assert patch["data"] == {"$patch": "replace", **DISABLED}
+    with pytest.raises(ValueError, match="must be empty"):
+        prepare(source, stage)
+
+
+@pytest.mark.parametrize("enabled", [False, True])
+@pytest.mark.parametrize("existing", [False, True])
+def test_ax_reconciles_observability_on_ready_clusters(tmp_path: Path, enabled: bool, existing: bool) -> None:
+    """Execute the reconciliation shell with a function double, never real kubectl."""
+    import json
+    import os
+    import subprocess
+    import sys
+
+    script = (ROOT / "ops/ax/up.sh").read_text()
+    section = script.split("# Reconcile the choice", 1)[1].split('kubectl --context "$CTX" -n ate-system get pods', 1)[0]
+    section = "# Reconcile the choice" + section
+    source = tmp_path / "substrate/manifests/ate-install/kind"
+    source.mkdir(parents=True)
+    upstream_data = {"OTEL_EXPORTER_OTLP_ENDPOINT": "http://collector:4317", "OTEL_LOGS_EXPORTER": "otlp"}
+    (source / "ate-otel-config.yaml").write_text(yaml.safe_dump({"data": upstream_data}))
+    log = tmp_path / "calls"
+    consumer = {
+        "kind": "Deployment",
+        "metadata": {"name": "ate-controller"},
+        "spec": {"template": {"spec": {"containers": [{"envFrom": [{"configMapRef": {"name": "ate-otel-config"}}]}]}}},
+    }
+    nonconsumer = {"kind": "Deployment", "metadata": {"name": "unrelated"}, "spec": {"template": {"spec": {"containers": []}}}}
+    mock = """
+set -euo pipefail
+kubectl() {
+  printf '%s\\t' "$@" >> "$CALL_LOG"
+  printf '\\n' >> "$CALL_LOG"
+  case " $* " in
+    *" get deployment "*)
+      if [ "$EXISTING" = 1 ]; then printf 'deployment.apps/%s\\n' "$7"; fi ;;
+    *" get deployments,daemonsets "*) printf '%s' "$WORKLOADS" ;;
+  esac
+  return 0
+}
+"""
+    env = {
+        **os.environ,
+        "OBSERVABILITY": str(enabled).lower(),
+        "EXISTING": str(int(existing)),
+        "PY": sys.executable,
+        "CTX": "test-context",
+        "SRC": str(tmp_path),
+        "REPO_ROOT": str(ROOT),
+        "CALL_LOG": str(log),
+        "WORKLOADS": json.dumps({"items": [consumer, nonconsumer]}),
+    }
+    result = subprocess.run(["bash", "-c", mock + section], env=env, text=True, capture_output=True)
+    assert result.returncode == 0, result.stderr
+    calls = [line.rstrip("\t").split("\t") for line in log.read_text().splitlines()]
+    scales = [call for call in calls if "scale" in call]
+    assert len(scales) == (3 if existing and not enabled else 0)
+    assert all(call[-1] == "--replicas=0" for call in scales)
+    applies = [call for call in calls if "apply" in call]
+    assert len(applies) == (2 if enabled else 0)
+    config_patch = next(call[-1] for call in calls if "configmap" in call)
+    assert json.loads(config_patch) == [{"op": "replace", "path": "/data", "value": upstream_data if enabled else DISABLED}]
+    assert not any("deployment/unrelated" in call for call in calls)
+    assert any("deployment/ate-controller" in call for call in calls)
+    if not enabled:
+        assert "OUSAST_KIND_OBSERVABILITY=1" in result.stdout

tests/test_cpg_backend.py

@@ -1368,33 +1368,3 @@ def test_requests_held_for_arbitration_are_reported_as_unasked() -> None:
     merged = scan._batched(once, "taint", {rid: {} for rid in weights}, weights=weights, within=lambda: next(budget, False), unasked=held)
     assert merged is not None and len(calls) == 1
     assert sorted(held + calls[0]) == sorted(weights), (held, calls)
-
-
-@pytest.mark.parametrize("extension", [".ts", ".tsx", ".mts", ".cts"])
-def test_typescript_uses_javascript_frontend(tmp_path, monkeypatch, extension):
-    from openultrasast.cpg.backend import JoernBackend
-
-    (tmp_path / "tests").mkdir()
-    (tmp_path / f"tests/probe{extension}").write_bytes(b"function first_party_test(x) { return x; }")
-    from openultrasast.preprocess import detect_language
-
-    assert detect_language(Path(f"probe{extension}")) == "typescript"
-    commands = []
-
-    def runner(command, **kwargs):
-        if _is_overlay(command):
-            return _overlaid(command, kwargs.get("cwd"))
-        commands.append(command)
-        assert Path(command[0]).name == "jssrc2cpg"
-        Path(command[command.index("-o") + 1]).write_bytes(b"graph")
-        return subprocess.CompletedProcess(command, 0, "", "")
-
-    monkeypatch.setattr("shutil.which", lambda name: f"/opt/joern/{name}")
-    monkeypatch.setenv("OPENULTRASAST_JOERN_PROBE", "on")
-    result = JoernBackend(runner=runner).build(tmp_path, language="typescript")
-    try:
-        assert result is not None
-        assert len(commands) == 1
-    finally:
-        if result is not None:
-            result.cleanup()

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
 

tests/test_cpg_session.py

@@ -343,3 +343,42 @@ def test_a_startup_wait_is_bounded_by_its_allowance_not_the_deadline(tmp_path, f
     # Bounded by the allowance, nowhere near the 600-second deadline.
     assert time.monotonic() - started < 10.0
     live.close()
+
+
+@pytest.mark.parametrize("cancel", [False, True])
+def test_live_receipt_observation_without_socket_or_jvm(tmp_path, monkeypatch, cancel):
+    import threading
+    from types import SimpleNamespace
+
+    live = EngineSession(tmp_path, ExecutionBudget(time.monotonic() + 10, 1), {})
+    live._process = SimpleNamespace(poll=lambda: None)
+    stopped = threading.Event()
+    observed = []
+
+    def close():
+        live._process = None
+        stopped.set()
+
+    def post(code, **kwargs):
+        target = code.split('java.nio.file.Path.of("', 1)[1].split('"', 1)[0]
+        nonce = code.split('__ousastPath, "', 1)[1].split("\\n", 1)[0]
+        Path(target).write_text(nonce + '\n{"__question__":"a"}\n')
+        if cancel:
+            assert stopped.wait(2)
+        return {"success": True}
+
+    def observe(output):
+        observed.append(output)
+        return not cancel
+
+    monkeypatch.setattr(live, "_post", post)
+    monkeypatch.setattr(live, "close", close)
+    live.output_observer = observe
+    answer = live.evaluate("work()", timeout=5)
+    assert any("__question__" in output for output in observed)
+    if cancel:
+        assert answer is None and live.failure == "session_observer_cancelled"
+        assert stopped.is_set()
+    else:
+        assert answer.stdout == '{"__question__":"a"}\n'
+    assert not list(tmp_path.glob("answer-*.txt"))

tests/test_engine_queue.py

@@ -0,0 +1,291 @@
+"""Offline queue protocol, image imports and deployment contract controls."""
+
+from __future__ import annotations
+
+import importlib
+import json
+import os
+import subprocess
+import sys
+import threading
+import time
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+import yaml
+
+ROOT = Path(__file__).resolve().parents[1]
+with pytest.MonkeyPatch.context() as patch:
+    patch.syspath_prepend(str(ROOT / "benchmarks/learn"))
+    service = importlib.import_module("engine_worker_service")
+    transport = importlib.import_module("engine_trace_queue")
+
+
+class FakeS3:
+    def __init__(self):
+        self.objects = {}
+        self.lock = threading.Lock()
+        self.choosing = None
+
+    def get(self, key):
+        with self.lock:
+            return (self.objects[key], None) if key in self.objects else None
+
+    def put(self, key, data, labels):
+        with self.lock:
+            self.objects[key] = data
+        if self.choosing and key.startswith(service.CLAIMED) and json.loads(data)["ticket"] == 0:
+            self.choosing.wait(timeout=5)
+
+    def delete(self, key):
+        with self.lock:
+            self.objects.pop(key, None)
+
+    def keys(self, prefix):
+        with self.lock:
+            return sorted(k for k in self.objects if k.startswith(prefix))
+
+
+def task():
+    return dict(
+        run="run",
+        id="pin",
+        task_id="run-pin",
+        input="engine-queue/inputs/run/pin/source.dat",
+        deadline=10,
+        question_deadline=2,
+        pin={"repo": "owner/repo", "pin": "commit", "units": []},
+    )
+
+
+def test_engine_worker_racing_claimants_run_exactly_once():
+    store = FakeS3()
+    item = task()
+    service.put(store, service.PENDING + item["task_id"], item)
+    store.choosing = threading.Barrier(2)
+    called = []
+
+    def execute(client, entry):
+        called.append(entry["task_id"])
+        time.sleep(0.05)
+        return {**entry["pin"], "done": True, "instrument": {"bytes": 11}}
+
+    with ThreadPoolExecutor(2) as pool:
+        futures = [pool.submit(service.Queue(store, identity).process_one, execute) for identity in ["worker-b", "worker-a"]]
+        results = [f.result(timeout=10) for f in futures]
+    assert sum(results) == 1
+    assert called == ["run-pin"]
+    result = service.read(store, service.result_key(item))
+    assert result["worker"] in {"worker-a", "worker-b"}
+    assert not store.keys(service.CLAIMED)
+    assert not store.keys(service.PENDING)
+
+
+def test_engine_worker_expired_lease_recovered_without_pending():
+    store = FakeS3()
+    item = task()
+    service.put(store, "engine-queue/claimed/dead/run-pin", dict(task=item, ticket=1, expires=99))
+    queue = service.Queue(store, "new", now=lambda: 100)
+    assert queue.process_one(lambda *_: {**item["pin"], "done": True})
+    assert service.read(store, service.result_key(item))["worker"] == "new"
+    assert not store.keys(service.CLAIMED)
+
+
+def test_engine_worker_late_arrival_cannot_displace_running_owner():
+    store = FakeS3()
+    item = task()
+    first = service.Queue(store, "z-first")
+    key = first.claim(item)
+    assert key
+    stop = threading.Event()
+    later = service.Queue(store, "a-later", stop=stop)
+    with ThreadPoolExecutor(1) as pool:
+        future = pool.submit(later.claim, item)
+        assert future.result(timeout=3) is None
+        assert service.read(store, key) is not None
+        service.put(store, service.result_key(item), {"done": True})
+        store.delete(key)
+
+
+def test_engine_queue_dispatch_roundtrip_and_timeout(tmp_path):
+    store = FakeS3()
+    checkout, output = tmp_path / "case", tmp_path / "out"
+    checkout.mkdir()
+    output.mkdir()
+    (checkout / "app.py").write_text("print(1)\n")
+    args = SimpleNamespace(deadline=10, question_deadline=2, queue_timeout=2, image="image@sha256:abc")
+    unit = {"unit": "u", "supported": True, "reason": "", "family": "injection"}
+    pin = {"repo": "owner/repo", "pin": "sha", "units": [unit], "questions": []}
+
+    def execute(client, item):
+        assert item["input"].endswith("/source.dat")
+        unpacked = tmp_path / "unpacked"
+        unpacked.mkdir()
+        service.unpack(client.get(item["input"])[0], unpacked)
+        assert (unpacked / "app.py").read_bytes() == b"print(1)\n"
+        return {
+            **item["pin"],
+            "done": True,
+            "container_exit": 0,
+            "units": [service.unit_record(unit, "asked-nothing")],
+            "instrument": {"bytes": 9, "heap_mb": 2560},
+        }
+
+    queue = service.Queue(store, "pod-1")
+    dispatcher = transport.Dispatcher(args, store, pause=lambda _: queue.process_one(execute))
+    completed, worker = dispatcher.execute(checkout, output, pin, "pin")
+    assert completed.returncode == 0
+    assert worker == "pod-1"
+    result = json.loads((output / "result.json").read_text())
+    assert result["executor"] == "queue"
+    assert result["units"][0]["status"] == "asked-nothing"
+    assert result["instrument"]["heap_mb"] == 2560
+    assert dispatcher.status() == {"pending": 0, "claimed": 0, "done": 1}
+    assert not store.keys("engine-queue/inputs/")
+    ticks = iter([0, 3])
+    dispatcher = transport.Dispatcher(args, store, clock=lambda: next(ticks))
+    assert dispatcher.execute(checkout, output, pin, "other") == (None, None)
+    assert not store.keys(service.PENDING)
+
+
+def test_engine_worker_imports_without_yaml():
+    code = "import sys; sys.modules['yaml'] = None; import engine_worker_service; import engine_trace_worker"
+    done = subprocess.run(
+        [sys.executable, "-c", code],
+        capture_output=True,
+        text=True,
+        env={**os.environ, "PYTHONPATH": f"{ROOT / 'src'}:{ROOT / 'benchmarks/learn'}"},
+    )
+    assert done.returncode == 0, done.stderr
+
+
+def test_engine_k8s_render_contract():
+    base = ROOT / "ops/k8s/engine"
+    config = yaml.safe_load((base / "kustomization.yaml").read_text())
+    # Render this base's resource inclusion, namespace, image and replacement transforms offline.
+    resources = [yaml.safe_load((base / name).read_text()) for name in config["resources"]]
+    deployment = next(r for r in resources if r["kind"] == "Deployment")
+    deployment["metadata"]["namespace"] = config["namespace"]
+    image = config["images"][0]
+    pod = deployment["spec"]["template"]["spec"]
+    container = pod["containers"][0]
+    assert container["image"] == image["name"]
+    container["image"] = image["newName"] + "@" + image["digest"]
+    replacement = config["replacements"][0]
+    assert replacement["source"]["fieldPath"] == "spec.template.spec.containers.0.image"
+    assert replacement["targets"][0]["fieldPaths"] == ["spec.template.spec.containers.0.env.0.value"]
+    container["env"][0]["value"] = container["image"]
+    assert "@sha256:" in container["image"]
+    assert deployment["spec"]["replicas"] == 2
+    assert container["resources"] == {"requests": {"memory": "3Gi", "cpu": "1"}, "limits": {"memory": "3Gi"}}
+    affinity = pod["affinity"]["podAntiAffinity"]["requiredDuringSchedulingIgnoredDuringExecution"][0]
+    assert affinity["topologyKey"] == "kubernetes.io/hostname"
+    assert affinity["labelSelector"]["matchLabels"] == deployment["spec"]["template"]["metadata"]["labels"]
+    assert pod["terminationGracePeriodSeconds"] == 900
+    assert container["envFrom"] == [{"secretRef": {"name": "ousast-engine-s3"}}]
+    for probe in ["readinessProbe", "livenessProbe"]:
+        assert container[probe]["httpGet"] == {"path": "/healthz", "port": "health"}
+    assert not pod["automountServiceAccountToken"]
+    assert pod["securityContext"]["runAsNonRoot"]
+    assert "hostPath" not in json.dumps(resources)
+    assert "privileged" not in json.dumps(resources)
+    assert all(r["kind"] != "Secret" for r in resources)
+    secret = yaml.safe_load((base / "secret.example.yaml").read_text())
+    assert secret["stringData"] == {}
+    assert "AWS_SECRET_ACCESS_KEY" not in json.dumps(resources)
+
+
+def test_engine_image_and_rustfs_policy():
+    dockerfile = (ROOT / "Dockerfile").read_text()
+    assert "COPY benchmarks ./benchmarks" in dockerfile
+    assert ".[semantic,s3]" in dockerfile
+    for name in ["engine_trace_worker.py", "engine_trace_parse.py", "engine_worker_service.py"]:
+        assert (ROOT / "benchmarks/learn" / name).stat().st_size > 0
+    doc = (ROOT / "docs/rustfs.md").read_text().split("## Scoped account for engine workers")[1]
+    policy = json.loads(doc.split("```json\n")[1].split("```")[0])
+    listing, objects = policy["Statement"]
+    assert listing["Action"] == "s3:ListBucket"
+    assert set(listing["Condition"]["StringLike"]["s3:prefix"]) == {
+        "engine-queue/",
+        "engine-queue/*",
+        "engine-results/",
+        "engine-results/*",
+    }
+    assert set(objects["Action"]) == {"s3:GetObject", "s3:PutObject", "s3:DeleteObject"}
+    assert set(objects["Resource"]) == {"arn:aws:s3:::sast-memory/engine-queue/*", "arn:aws:s3:::sast-memory/engine-results/*"}
+
+
+@pytest.mark.parametrize("hang", [False, True])
+def test_engine_worker_process_deadline_and_shipped_schema(tmp_path, monkeypatch, hang):
+    store = FakeS3()
+    item = task()
+    item["image"] = "image@sha256:abc"
+    item["deadline"] = 0.1 if hang else 5
+    item["pin"]["units"] = [{"unit": "u", "supported": True, "reason": ""}]
+    (tmp_path / "app.py").write_text("print(1)\n")
+    store.put(item["input"], service.pack(tmp_path), {})
+    monkeypatch.setenv("ENGINE_IMAGE", item["image"])
+
+    def fake_worker(pin, root, out, deadline, question_deadline):
+        assert (root / "app.py").read_bytes() == b"print(1)\n"
+        if hang:
+            time.sleep(10)
+        out.write_text(json.dumps({**pin, "done": True, "units": [service.unit_record(pin["units"][0], "asked-nothing")]}))
+
+    monkeypatch.setattr(service, "worker", fake_worker)
+    result = service.execute_task(store, item)
+    assert result["done"]
+    assert result["units"][0]["status"] == ("timeout" if hang else "asked-nothing")
+    assert result["container_exit"] == (None if hang else 0)
+    assert result["seconds"] < 5
+
+
+def test_engine_worker_stopping_does_not_claim():
+    store = FakeS3()
+    item = task()
+    service.put(store, service.PENDING + item["task_id"], item)
+    stop = threading.Event()
+    stop.set()
+    assert not service.Queue(store, stop=stop).process_one(lambda *_: pytest.fail("claimed after SIGTERM"))
+    assert store.keys(service.PENDING)
+    assert not store.keys(service.CLAIMED)
+
+
+def test_engine_worker_rejects_archive_links_and_escape(tmp_path):
+    import io
+    import tarfile
+
+    for name, kind in [("../escape", tarfile.REGTYPE), ("link", tarfile.SYMTYPE)]:
+        stream = io.BytesIO()
+        with tarfile.open(fileobj=stream, mode="w") as archive:
+            member = tarfile.TarInfo(name)
+            member.type = kind
+            member.linkname = "/etc/passwd"
+            archive.addfile(member)
+        with pytest.raises(ValueError, match="unsafe source archive"):
+            service.unpack(stream.getvalue(), tmp_path)
+
+
+def test_engine_worker_timeout_kills_separate_session_descendant(tmp_path, monkeypatch):
+    store = FakeS3()
+    item = task()
+    item.update(image="image@sha256:abc", deadline=0.5)
+    item["pin"]["units"] = [{"unit": "u", "supported": True, "reason": ""}]
+    (tmp_path / "app.py").write_text("print(1)\n")
+    store.put(item["input"], service.pack(tmp_path), {})
+    monkeypatch.setenv("ENGINE_IMAGE", item["image"])
+    pid_file = tmp_path / "escaped-pid"
+
+    def fake_worker(*_):
+        child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"], start_new_session=True)
+        pid_file.write_text(str(child.pid))
+        time.sleep(30)
+
+    monkeypatch.setattr(service, "worker", fake_worker)
+    result = service.execute_task(store, item)
+    assert result["units"][0]["status"] == "timeout"
+    assert pid_file.exists(), "instrument: fake worker never launched its descendant"
+    pid = int(pid_file.read_text())
+    assert not Path(f"/proc/{pid}").exists(), "escaped session survived the task deadline"

tests/test_engine_task_entry.py

@@ -0,0 +1,234 @@
+"""Offline AX worker entry controls; HTTP is loopback only."""
+
+import importlib.util
+import io
+import json
+import tarfile
+import threading
+import time
+import urllib.error
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+
+import pytest
+
+SCRIPTS = Path(__file__).resolve().parents[1] / "benchmarks/learn"
+
+
+def load(name):
+    with pytest.MonkeyPatch.context() as patch:
+        patch.syspath_prepend(str(SCRIPTS))
+        spec = importlib.util.spec_from_file_location(name, SCRIPTS / f"{name}.py")
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+        return module
+
+
+def test_engine_cluster_heap_flags(monkeypatch):
+    worker = load("engine_trace_worker")
+    monkeypatch.setenv("JAVA_TOOL_OPTIONS", "-Xmx9999m")
+    backend = worker.MeasuredBackend(30, 10, lambda: None, heap_profile="cluster")
+    flags = (
+        "-Xms1700m -Xmx1700m -XX:MaxMetaspaceSize=256m -XX:ReservedCodeCacheSize=128m "
+        "-XX:MaxDirectMemorySize=256m -Xss512k -XX:ActiveProcessorCount=2 -XX:+UseG1GC"
+    )
+    assert backend._jvm_env()["JAVA_TOOL_OPTIONS"] == flags
+    assert backend._heap_flag() == "-J-Xmx1700m"
+    assert worker.MeasuredBackend(30, 10, lambda: None)._heap_flag() == "-J-Xmx2560m"
+
+
+@pytest.mark.parametrize("readiness", ["ready", "delayed", "closed", "source_denied"])
+@pytest.mark.parametrize("failure", [False, True])
+@pytest.mark.parametrize("transport", ["memory", "loopback"])
+def test_task_entry_http_roundtrip(tmp_path, monkeypatch, failure, transport, readiness):
+    entry = load("engine_task_entry")
+    archive = io.BytesIO()
+    with tarfile.open(fileobj=archive, mode="w") as tar:
+        data = b"print('input proof')\n"
+        info = tarfile.TarInfo("app.py")
+        info.size = len(data)
+        tar.addfile(info, io.BytesIO(data))
+    pin = {"pin": "abc", "units": [{"unit": "u", "supported": True}], "questions": {"u": {}}}
+    uploads = []
+    gets = []
+    puts = []
+    started = time.monotonic()
+
+    class Handler(BaseHTTPRequestHandler):
+        def do_GET(self):
+            gets.append((self.path, time.monotonic() - started))
+            if readiness == "source_denied" and self.path == "/source":
+                self.send_response(403)
+                self.end_headers()
+                return
+            if readiness == "closed" or (readiness == "delayed" and gets[-1][1] < 3):
+                self.send_response(403 if len(gets) % 2 else 503)
+                self.end_headers()
+                return
+            data = archive.getvalue() if self.path == "/source" else json.dumps(pin).encode()
+            self.send_response(200)
+            self.end_headers()
+            self.wfile.write(data)
+
+        def do_PUT(self):
+            body = self.rfile.read(int(self.headers["Content-Length"]))
+            puts.append(time.monotonic())
+            if readiness == "closed" and len(puts) == 1:
+                self.send_response(403)
+                self.end_headers()
+                return
+            uploads.append(body)
+            self.send_response(200)
+            self.end_headers()
+
+        def log_message(self, *args):
+            pass
+
+    fake_worker = tmp_path / "worker.py"
+    fake_worker.write_text(
+        "import argparse,json,pathlib,sys\np=argparse.ArgumentParser()\n"
+        "for key in ['worker','root','output','deadline','question-deadline','heap-profile']: p.add_argument('--'+key)\n"
+        "a=p.parse_args()\nassert a.heap_profile == 'cluster'\n"
+        "assert (pathlib.Path(a.root)/'app.py').stat().st_size > 0\n"
+        + ("sys.exit(7)\n" if failure else "pathlib.Path(a.output).write_text(json.dumps({'done':True,'units':[]}))\n")
+    )
+    if transport == "loopback":
+        try:
+            server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+        except PermissionError:
+            pytest.skip("sandbox prohibits loopback sockets; in-memory HTTP roundtrip still runs")
+        thread = threading.Thread(target=server.serve_forever, daemon=True)
+        thread.start()
+        base = f"http://127.0.0.1:{server.server_port}"
+    else:
+        base = "https://fake-store.invalid"
+
+        def urlopen(request, timeout):
+            method = request.get_method() if hasattr(request, "get_method") else "GET"
+            url = request.full_url if hasattr(request, "full_url") else request
+            body = request.data or b"" if method == "PUT" else b""
+            raw = f"{method} {url.removeprefix(base)} HTTP/1.0\r\nContent-Length: {len(body)}\r\n\r\n".encode() + body
+
+            class Socket:
+                def __init__(self):
+                    self.sent = bytearray()
+
+                def makefile(self, *args):
+                    return io.BytesIO(raw)
+
+                def sendall(self, data):
+                    self.sent.extend(data)
+
+            connection = Socket()
+            Handler(connection, ("127.0.0.1", 0), None)
+            headers, payload = bytes(connection.sent).split(b"\r\n\r\n", 1)
+            status = int(headers.split()[1])
+            if status != 200:
+                raise urllib.error.HTTPError(url, status, "denied", {}, io.BytesIO(payload))
+            return io.BytesIO(payload)
+
+        monkeypatch.setattr(entry.urllib.request, "urlopen", urlopen)
+    try:
+        for name, suffix in [("SOURCE_URL", "source"), ("QUESTIONS_URL", "questions"), ("RESULT_URL", "result")]:
+            monkeypatch.setenv(name, base + "/" + suffix)
+        budget = 0.7 if readiness == "closed" else 60
+        assert entry.main(work_root=tmp_path / "work", worker_script=fake_worker, readiness_budget=budget) == 0
+        if readiness == "delayed":
+            assert len(gets) >= 5
+            assert gets[-1][1] >= 3
+        if readiness == "closed":
+            assert 0.7 <= puts[0] - started < 1.5
+            assert len(puts) == 2
+            assert puts[1] - puts[0] >= 0.5
+            assert all(path == "/questions" for path, _ in gets)
+    finally:
+        if transport == "loopback":
+            server.shutdown()
+            thread.join()
+            server.server_close()
+    assert len(uploads) == 1
+    with tarfile.open(fileobj=io.BytesIO(uploads[0])) as tar:
+        result = json.load(tar.extractfile("result.json"))
+    assert result["done"]
+    if readiness == "closed":
+        assert result["status"] == "failed"
+        assert result["reason"] == "GET questions: egress not open after 60 s"
+        assert json.loads((tmp_path / "work/out/result.json").read_text()) == result
+    elif readiness == "source_denied":
+        assert result["status"] == "failed"
+        assert result["reason"] == "GET source: HTTP 403"
+        assert result["units"][0]["reason"] == "GET source: HTTP 403"
+        assert base not in json.dumps(result)
+        assert json.loads((tmp_path / "work/out/result.json").read_text()) == result
+    elif failure:
+        assert result["units"][0]["status"] == "failed"
+        assert "7" in result["units"][0]["reason"]
+
+
+def test_engine_cluster_profile_recorded_on_input_failure(tmp_path):
+    worker = load("engine_trace_worker")
+    pin = {
+        "units": [{"unit": "u", "file": "missing.py", "language": "python", "supported": True, "reason": ""}],
+        "questions": {"u": {}},
+    }
+    result = worker.worker(pin, tmp_path, tmp_path / "result.json", 30, 10, heap_profile="cluster")
+    assert result["units"][0]["status"] == "failed"
+    assert result["instrument"]["heap_profile"] == "cluster"
+    assert result["instrument"]["heap_mb"] == 1700
+    assert result["instrument"]["java_tool_options"] == worker.CLUSTER_JVM_FLAGS
+
+
+def test_task_entry_rejects_tar_traversal(tmp_path):
+    entry = load("engine_task_entry")
+    source = tmp_path / "source.tar"
+    with tarfile.open(source, "w") as archive:
+        member = tarfile.TarInfo("../escape")
+        member.size = 1
+        archive.addfile(member, io.BytesIO(b"x"))
+    with pytest.raises(ValueError, match="unsafe"):
+        entry.unpack(source, tmp_path / "case")
+    assert not (tmp_path / "escape").exists()
+
+
+@pytest.mark.parametrize("error", [urllib.error.URLError("refused"), ConnectionResetError("reset")])
+def test_readiness_connection_errors_back_off_to_cap(tmp_path, monkeypatch, error):
+    entry = load("engine_task_entry")
+    now = [0.0]
+    pauses = []
+    timeouts = []
+
+    def pause(seconds):
+        pauses.append(seconds)
+        now[0] += seconds
+
+    def refused(request, timeout):
+        timeouts.append(timeout)
+        raise error
+
+    monkeypatch.setattr(entry.time, "monotonic", lambda: now[0])
+    monkeypatch.setattr(entry.time, "sleep", pause)
+    monkeypatch.setattr(entry.urllib.request, "urlopen", refused)
+    with pytest.raises(RuntimeError, match="^egress not open after 60 s$"):
+        entry.download("https://fake.invalid", tmp_path / "input", readiness_budget=60)
+    assert now[0] == 60
+    assert pauses[:5] == [0.5, 1, 2, 4, 8]
+    assert max(pauses) == 8
+    assert timeouts[-1] == pauses[-1]
+
+
+@pytest.mark.parametrize("status, attempts", [(403, 3), (503, 3), (404, 1)])
+def test_later_transfers_keep_short_retry(monkeypatch, status, attempts):
+    entry = load("engine_task_entry")
+    calls = []
+    pauses = []
+
+    def refused(request, timeout):
+        calls.append(timeout)
+        raise urllib.error.HTTPError(request, status, "denied", {}, io.BytesIO())
+
+    monkeypatch.setattr(entry.time, "sleep", pauses.append)
+    monkeypatch.setattr(entry.urllib.request, "urlopen", refused)
+    with pytest.raises(urllib.error.HTTPError):
+        entry.transfer("https://fake.invalid")
+    assert len(calls) == attempts
+    assert pauses == list(range(1, attempts))

tests/test_joern_leak_audit.py

@@ -0,0 +1,190 @@
+"""Synthetic, offline result snapshots; no analyzer, model or corpus access."""
+
+import hashlib
+import importlib
+import json
+from pathlib import Path
+
+import pytest
+
+
+@pytest.fixture
+def audit_module(monkeypatch):
+    monkeypatch.syspath_prepend(str(Path(__file__).resolve().parents[1] / "benchmarks/learn"))
+    return importlib.import_module("joern_leak_audit")
+
+
+def units(count=12):
+    return [
+        {"unit": f"{pair}-{label}", "pair": str(pair), "label": label, "family": "injection", "group": "g", "fold": "f"}
+        for pair in range(count)
+        for label in (1, 0)
+    ]
+
+
+def write_result(root, module, rows, name="pin", *, done=True, version=None):
+    path = root / f"{name}.json"
+    path.write_text(
+        json.dumps(
+            {
+                "repo": "repo",
+                "pin": name,
+                "done": done,
+                "materialization_version": module.materialization_version() if version is None else version,
+                "units": rows,
+            }
+        )
+    )
+    return path
+
+
+def result(unit, status):
+    return {
+        **unit,
+        "source": "authored",
+        "status": status,
+        "pin_role": "vulnerable" if unit["label"] else "fixed",
+        "traces": [{"steps": [{"roles": ["source"]}, {"roles": ["sink"]}], "sanitized": False, "guard": {"text": "", "bounded": False}}]
+        if status == "path"
+        else [],
+        "witness_rows": [{"sink": "execute(x)"}] if status == "path" else [],
+    }
+
+
+def run(module, root, population):
+    return module.audit(population, root, expected_rows=len(population), expected_pairs=len(population) // 2)
+
+
+def test_skips_stale_unfinished_and_proves_read(audit_module, tmp_path):
+    population = units(2)
+    paths = [
+        write_result(tmp_path, audit_module, [result(population[0], "path")], "valid"),
+        write_result(tmp_path, audit_module, [result(population[1], "path")], "stale", version=-1),
+        write_result(tmp_path, audit_module, [result(population[2], "path")], "unfinished", done=False),
+        write_result(tmp_path, audit_module, [result(population[3], "path")], "both", version=-1, done=False),
+    ]
+    report = run(audit_module, tmp_path, population)
+    evidence = report["inputs"]
+    assert (evidence["stale"], evidence["unfinished"], evidence["skipped"]) == (2, 2, 3)
+    assert evidence["files_read"] == 4 and evidence["accepted_files"] == evidence["result_rows_read"] == 1
+    assert evidence["bytes_read"] == sum(p.stat().st_size for p in paths)
+    assert report["matched_rows"] == 1
+    assert report["overall"]["coverage"]["missing"] == 3
+    assert report["by_source"]["unknown"]["rows"] == 3
+    assert report["overall"]["signals"]["has_path"]["rows"] == 1
+    assert report["overall"]["both_asked_signals"]["has_path"]["pooled_auc"] is None
+
+
+def test_asymmetry_and_honest_denominator(audit_module, tmp_path):
+    population = units(6)
+    statuses = [
+        "path",
+        "asked-nothing",
+        "asked-nothing",
+        "path",
+        "path",
+        "path",
+        "asked-nothing",
+        "asked-nothing",
+        "path",
+        "failed",
+        "unsupported",
+        "timeout",
+    ]
+    write_result(tmp_path, audit_module, [result(u, s) for u, s in zip(population, statuses, strict=True)])
+    report = run(audit_module, tmp_path, population)
+    overall = report["overall"]
+    assert overall["pair_asymmetry"] == {"vulnerable_only": 1, "fixed_only": 1, "both": 1, "neither": 1, "not_both_asked": 2}
+    assert overall["both_asked_pairs"] == 4
+    assert overall["both_asked_signals"]["asked"]["within_pair_accuracy"] == 0.5
+    assert overall["coverage"] == {"path": 5, "asked-nothing": 4, "failed": 1, "timeout": 1, "unsupported": 1, "missing": 0}
+    assert report["by_family"]["injection"] == overall == report["by_source"]["authored"]
+
+
+@pytest.mark.parametrize("balanced", [False, True])
+def test_constructed_leak_and_balanced_set(audit_module, tmp_path, balanced):
+    population = units()
+    rows = [result(u, "path" if u["label"] == (int(u["pair"]) % 2 if balanced else 1) else "asked-nothing") for u in population]
+    write_result(tmp_path, audit_module, rows)
+    overall = run(audit_module, tmp_path, population)["overall"]
+    for denominator in ("signals", "both_asked_signals"):
+        signals = overall[denominator]
+        for name in ("has_path", "shortest_path_steps"):
+            assert signals[name]["flagged"] is not balanced
+            assert signals[name]["within_pair_accuracy"] == signals[name]["pooled_auc"] == (0.5 if balanced else 1)
+        assert signals["sink_kind"]["execute"]["flagged"] is not balanced
+        assert not signals["asked"]["flagged"]
+        assert not signals["sanitized"]["flagged"]
+        assert not signals["guard_present"]["flagged"]
+
+
+def test_minimum_samples_reverse_and_source_proxy(audit_module, tmp_path):
+    population = units(2)
+    write_result(tmp_path, audit_module, [result(u, "path" if not u["label"] else "asked-nothing") for u in population])
+    assert not run(audit_module, tmp_path, population)["overall"]["signals"]["has_path"]["flagged"]
+    population = units(12)
+    rows = []
+    for u in population:
+        row = result(u, "path" if not u["label"] else "unsupported")
+        row["source"] = "positive" if u["label"] else "negative"
+        rows.append(row)
+    write_result(tmp_path, audit_module, rows)
+    report = run(audit_module, tmp_path, population)
+    assert report["overall"]["signals"]["asked"]["flagged"]
+    assert report["overall"]["signals"]["asked"]["pooled_auc"] == 0
+    assert report["overall"]["both_asked_pairs"] == 0
+    assert report["by_source"]["positive"]["signals"]["asked"]["pooled_auc"] is None
+
+
+def test_shortest_trace_evidence_and_missing_trace(audit_module):
+    row = result(units(1)[0], "path")
+    row["traces"].insert(0, {"steps": [{}] * 5, "sanitized": False})
+    row["witness_rows"].insert(0, {"sink": "longer(x)"})
+    row["traces"][1].update(sanitized=True, guard={"bounded": True, "text": "off-path bound"})
+    features = audit_module.features(row)
+    assert (features["shortest_path_steps"], features["sanitized"], features["guard_present"], features["sink_kind"]) == (
+        2,
+        1,
+        1,
+        "execute",
+    )
+    row["traces"] = []
+    assert audit_module.features(row)["shortest_path_steps"] is None
+
+
+def test_bad_inputs_fail_loudly(audit_module, tmp_path):
+    population = units(1)
+    write_result(tmp_path, audit_module, [result(population[0], "path")] * 2)
+    with pytest.raises(ValueError, match="Duplicate"):
+        run(audit_module, tmp_path, population)
+    write_result(tmp_path, audit_module, [{**result(population[0], "path"), "label": 0}])
+    with pytest.raises(ValueError, match="metadata mismatch"):
+        run(audit_module, tmp_path, population)
+
+
+@pytest.mark.parametrize("done", [False, True])
+def test_record_uses_hashes_without_repository_identity(audit_module, tmp_path, done):
+    repo = "https://github.com/owner/name"
+    population = [{**u, "group": "owner/name", "repo": repo, "source": "population-v1"} for u in units(1)]
+    root = tmp_path / "github.com" / "owner" / "name"
+    root.mkdir(parents=True)
+    paths = []
+    for i, u in enumerate(population):
+        row = {**result(u, "path"), "source": "population-v1"}
+        path = write_result(root, audit_module, [row], f"name-{1 - i}", done=done)
+        record = json.loads(path.read_text())
+        record.update(repo=repo, pin="owner/name")
+        path.write_text(json.dumps(record))
+        paths.append(path)
+    report = run(audit_module, root, population)
+    serialized = json.dumps(report)
+    assert repo not in serialized and "owner/name" not in serialized
+    assert "files" not in report["inputs"]
+    inventory = sorted((p.name, p.stat().st_size) for p in paths)
+    assert report["inputs"]["files_sha256"] == hashlib.sha256(json.dumps(inventory, separators=(",", ":")).encode()).hexdigest()
+    assert report["inputs"]["files_read"] == 2
+    assert report["inputs"]["bytes_read"] == sum(size for _, size in inventory)
+    assert report["inputs"]["result_rows_read"] == (2 if done else 0)
+    assert report["by_source"]["population-v1"]["rows"] == 2
+    assert [r["unit"] for r in report["units"]] == [u["unit"] for u in population]
+    assert [r["pair"] for r in report["units"]] == [u["pair"] for u in population]

tests/test_learn_build_features.py

@@ -87,8 +87,8 @@ def test_feature_typescript_repository_witness_reaches_static_record(tmp_path: P
     assert parts["engine"].state == "ran"
     assert parts["engine"].values["eng.witness_steps_min"] == 5
     assert parts["engine"].values["eng.findings"] == 1
-    assert builder.recorded_engine("typescript", [], None, True).state == "failed"
-    assert builder.recorded_engine("typescript", [], result, False).state == "ran"
+    assert builder.recorded_engine("typescript", [], None, True).state == "none"
+    assert builder.recorded_engine("typescript", [], result, False).state == "none"
     assert builder.recorded_engine("typescript", [], result, True).values["eng.findings"] == 0
 
 

tests/test_learn_mechanisms.py

@@ -0,0 +1,289 @@
+"""Authored fix pairs only: no store, models, containers, or population files."""
+
+from __future__ import annotations
+
+import importlib.util
+import json
+from dataclasses import replace
+
+import pytest
+
+from openultrasast.learn.folds import Fold
+from openultrasast.learn.mechanisms import Pair, mine_pairs
+
+
+def pair(before: str, after: str, *, language: str = "python", family: str = "output_encoding", repo: str = "owner/alpha") -> Pair:
+    return Pair("pair-" + repo, repo, repo, language, family, before, after)
+
+
+def repeated(p: Pair, **kwargs):
+    return mine_pairs([p, replace(p, id="pair-beta", repository="owner/beta", group="owner/beta")], **kwargs)
+
+
+def test_learn_mechanisms_sanitizer_return_and_def_use():
+    p = pair("def f(x):\n    return x", "def f(x):\n    y = html.escape(x)\n    return y")
+    result = repeated(p)
+    assert [(c["form"], c["pattern"]) for c in result.candidates] == [("sanitizer", {"callee": "html.escape", "argument": 0})]
+    assert result.candidates[0]["repository_count"] == 2
+    assert "owner/alpha" not in json.dumps(result.record)
+    assert result.record["pairs_mined"] == 2
+
+
+def test_learn_mechanisms_javascript():
+    p = pair("function f(x) { res.send(x); }", "function f(x) { const y = escapeHtml(x); res.send(y); }", language="javascript")
+    assert repeated(p).candidates[0]["pattern"] == {"callee": "escapeHtml", "argument": 0}
+
+
+def test_learn_mechanisms_guard_and_known():
+    p = pair(
+        "def f(u):\n    redirect(u)",
+        "def f(u):\n    if not is_safe_url(u):\n        abort()\n    redirect(u)",
+        family="untrusted_destination",
+    )
+    candidate = repeated(p).candidates[0]
+    assert candidate["form"] == "guard"
+    assert candidate["pattern"] == {"callee": "is_safe_url", "argument": 0, "predicate": "truthy"}
+    assert candidate["known"] is True
+    assert not any(f["eligible"] for f in candidate["folds"])
+
+
+@pytest.mark.skipif(importlib.util.find_spec("tree_sitter_php") is None, reason="PHP tree-sitter wheel unavailable")
+def test_learn_mechanisms_php_safe_argument_and_echo():
+    p = pair(
+        "function f($d) { return unserialize($d); }",
+        "function f($d) { return unserialize($d, ['allowed_classes' => false]); }",
+        language="php",
+        family="deserialization",
+    )
+    assert repeated(p).candidates[0]["pattern"] == {"callee": "unserialize", "argument": 1, "key": "allowed_classes", "value": False}
+    assert repeated(p).candidates[0]["form"] == "safe_argument"
+    p = pair("function f($x) { echo $x; }", "function f($x) { echo htmlspecialchars($x); }", language="php")
+    assert repeated(p).candidates[0]["form"] == "sanitizer"
+
+
+def test_learn_mechanisms_safe_alternative():
+    p = pair("def f(x):\n    return yaml.load(x)", "def f(x):\n    return yaml.safe_load(x)", family="deserialization")
+    candidate = repeated(p).candidates[0]
+    assert candidate["form"] == "safe_alternative"
+    assert candidate["pattern"] == {"callee": "yaml.load", "replacement": "yaml.safe_load"}
+    assert candidate["known"] is True
+
+
+def test_learn_mechanisms_keyword():
+    p = pair("def f(x):\n    subprocess.run(x, shell=True)", "def f(x):\n    subprocess.run(x, shell=False)", family="injection")
+    result = repeated(p)
+    assert any(c["pattern"] == {"callee": "subprocess.run", "key": "shell", "value": False} for c in result.candidates)
+
+
+def test_learn_mechanisms_one_repository_and_local_helpers_rejected():
+    p = pair("def f(x):\n    return x", "def f(x):\n    return cleanse(x)")
+    assert mine_pairs([p]).candidates == []
+    local = replace(p, fixed="def cleanse(x):\n    return x\ndef f(x):\n    return cleanse(x)")
+    result = repeated(local)
+    assert result.candidates == []
+    assert result.record["skips"]["local_definition"] == 2
+    assert repeated(replace(p, fixed="def f(x):\n    return alpha_escape(x)")).candidates == []
+
+
+def test_learn_mechanisms_folds_exclude_proposers():
+    p = pair("def f(x):\n    return x", "def f(x):\n    return cleanse(x)")
+    folds = [Fold("outer", "outer-0", frozenset({p.group})), Fold("outer", "outer-1", frozenset({"elsewhere"}))]
+    candidate = repeated(p, folds=folds).candidates[0]
+    assert [(f["repository_count"], f["eligible"]) for f in candidate["folds"]] == [(1, False), (2, True)]
+    assert candidate["folds"][0]["proposing_pairs"] == ["pair-beta"]
+
+
+@pytest.mark.parametrize(
+    "fixed",
+    [
+        "def f(x):\n    cleanse(x)\n    return x",
+        "def f(x):\n    y = cleanse(x)\n    y = 'constant'\n    return y",
+        "def f(x):\n    if is_safe_url(x):\n        pass\n    redirect(x)",
+    ],
+)
+def test_learn_mechanisms_unrelated_calls_are_not_candidates(fixed):
+    before = "def f(x):\n    redirect(x)" if "redirect" in fixed else "def f(x):\n    return x"
+    family = "untrusted_destination" if "redirect" in fixed else "output_encoding"
+    assert repeated(pair(before, fixed, family=family)).candidates == []
+
+
+def test_learn_mechanisms_skips_and_determinism():
+    p = pair("def f(x):\n    return x", "def f(x):\n    return cleanse(x)")
+    assert repeated(p) == repeated(p)
+    assert repeated(replace(p, fixed="def broken(")).record["skips"]["unparseable"] == 2
+    assert repeated(replace(p, language="ruby")).record["skips"]["unsupported_language"] == 2
+    assert (
+        repeated(replace(p, fixed="def f(x):\n    for y in x:\n        cleanse(y)\n    return x")).record["skips"]["unsupported_shape"] == 2
+    )
+
+
+def test_learn_mechanisms_guard_must_test_sink_value():
+    p = pair(
+        "def f(x, u):\n    redirect(u)",
+        "def f(x, u):\n    if not check(x):\n        abort()\n    redirect(u)",
+        family="untrusted_destination",
+    )
+    assert repeated(p).candidates == []
+
+
+def test_learn_mechanisms_existing_path_call_is_not_new():
+    p = pair("def f(x):\n    return html.escape(x)", "def f(x):\n    y = html.escape(x)\n    return y")
+    assert repeated(p).candidates == []
+
+
+def test_learn_mechanisms_missing_grammar_is_counted(monkeypatch):
+    monkeypatch.setattr("openultrasast.learn.mechanisms.grammar_for", lambda language: None)
+    result = repeated(pair("def f(x):\n    return x", "def f(x):\n    return cleanse(x)"))
+    assert result.candidates == []
+    assert result.record["skips"] == {"grammar_unavailable": 2}
+
+
+def test_learn_mechanisms_pair_identity_and_unread_inputs():
+    from openultrasast.learn.mechanisms import pairs_from_rows
+
+    base = {
+        "repo": "owner/alpha",
+        "group": "owner/alpha",
+        "source": "pairs",
+        "candidate": "app.py::f",
+        "language": "python",
+        "family": "output_encoding",
+        "profile": "static",
+        "unit": "pin",
+    }
+    rows = [{**base, "label": label, "excerpt_sha": str(label)} for label in (0, 1)]
+    pairs, skips, count = pairs_from_rows(rows, lambda sha: "def f(x):\n    return x")
+    assert len(pairs) == count == 1 and not skips
+    assert pairs_from_rows(rows + [rows[0]], lambda sha: "code")[1] == {"ambiguous_pair": 1}
+    assert pairs_from_rows(rows[:1], lambda sha: "code")[1] == {"incomplete_pair": 1}
+    assert pairs_from_rows(rows, lambda sha: None)[1] == {"unreadable_excerpt": 1}
+    forbidden = [{**row, "source": "population-v3"} for row in rows]
+    assert pairs_from_rows(forbidden, lambda sha: pytest.fail("must not read excluded excerpts"))[1] == {"excluded_source": 2}
+
+
+def memory_store(tmp_path):
+    from openultrasast.plane.memory import FileStore, row_id
+
+    store = FileStore(tmp_path / "memory")
+    rows = []
+    for repo in ("owner/alpha", "owner/beta", "owner/gamma"):
+        for label in (0, 1):
+            code = "def f(x):\n    return x" if label else "def f(x):\n    return cleanse(x)"
+            sha = store.put_blob("excerpts", code.encode())
+            rows.append(
+                {
+                    "id": row_id("example", "test", "test", f"{repo}:{label}"),
+                    "kind": "example",
+                    "repo": repo,
+                    "pin": str(label) * 40,
+                    "run": "test",
+                    "task": "test",
+                    "population": "pairs",
+                    "split": "pairs",
+                    "image": "host",
+                    "candidate": "app.py::f",
+                    "family": "output_encoding",
+                    "language": "python",
+                    "profile": "static",
+                    "label": label,
+                    "source": "pairs",
+                    "group": repo,
+                    "unit": "pin",
+                    "x": {},
+                    "instruments": {},
+                    "excerpt_sha": sha,
+                }
+            )
+    store.put_rows(rows)
+    return store
+
+
+def test_learn_mechanisms_store_cli_and_privacy(tmp_path, monkeypatch):
+    from openultrasast.cli import main
+    from openultrasast.learn.mechanisms import RECORD, mine_store
+
+    store = memory_store(tmp_path)
+    result = mine_store(store)
+    assert result.record["complete_pairs"] == 3
+    assert result.record["excerpt_bytes_read"] > 0
+    assert len(result.candidates) == 1
+    assert len(result.candidates[0]["folds"]) == 5
+    monkeypatch.setenv("OUSAST_RESULTS", str(tmp_path / "results"))
+    root = tmp_path / "repo"
+    out = tmp_path / "results" / "plane" / "mining" / "candidates.jsonl"
+    assert main(["learn", "mechanisms", "mine", "--memory", f"file://{tmp_path / 'memory'}", "--out", str(out), "--root", str(root)]) == 0
+    record = (root / RECORD).read_text()
+    assert all(repo not in record for repo in ("alpha", "beta", "gamma"))
+    assert json.loads(out.read_text())["repository_count"] == 3
+    assert main(["learn", "mechanisms", "mine", "--memory", f"file://{tmp_path / 'empty'}", "--out", str(out), "--root", str(root)]) == 2
+    assert json.loads((root / RECORD).read_text())["status"] == "no_mineable_pairs"
+
+
+def test_learn_mechanisms_output_location_rejected(tmp_path):
+    from openultrasast.learn.mechanisms import MiningResult, write_result
+
+    root = tmp_path / "repo"
+    with pytest.raises(ValueError, match="outside"):
+        write_result(MiningResult([], {}), root / "candidate.jsonl", root=root, results=tmp_path)
+    assert not root.exists()
+
+
+@pytest.mark.parametrize("signature,source", [("", "input()"), ("request", "request.get_json()")])
+def test_learn_mechanisms_source_calls(signature, source):
+    p = pair(f"def f({signature}):\n    return {source}", f"def f({signature}):\n    return html.escape({source})")
+    assert repeated(p).candidates[0]["pattern"] == {"callee": "html.escape", "argument": 0}
+
+
+def test_learn_mechanisms_inserted_log_is_not_sink_replacement():
+    p = pair("def f(x):\n    yaml.load(x)", "def f(x):\n    logger.info(x)\n    yaml.safe_load(x)", family="deserialization")
+    result = repeated(p)
+    assert result.candidates == []
+    assert result.record["skips"]["ambiguous_sink"] == 2
+
+
+def test_learn_mechanisms_keyword_reordering_is_not_change():
+    p = pair(
+        "def f(x):\n    subprocess.run(x, shell=False, timeout=10)",
+        "def f(x):\n    subprocess.run(x, timeout=10, shell=False)",
+        family="injection",
+    )
+    assert repeated(p).candidates == []
+
+
+def test_learn_mechanisms_legacy_pair_ids_are_repository_scoped(tmp_path):
+    from openultrasast.learn.mechanisms import mine_store
+
+    result = mine_store(memory_store(tmp_path))
+    candidate = result.candidates[0]
+    assert len(candidate["proposing_pairs"]) == 3
+    assert all(len(ids) == 2 for ids in candidate["proposing_examples"].values())
+    assert len({i for ids in candidate["proposing_examples"].values() for i in ids}) == 6
+
+
+def test_learn_mechanisms_runtime_callable_is_unsupported():
+    p = pair("def f(x, clean):\n    return x", "def f(x, clean):\n    return clean(x)")
+    result = repeated(p)
+    assert result.candidates == []
+    assert result.record["skips"]["unsupported_shape"] == 2
+
+
+def test_learn_mechanisms_language_support_is_not_pooled():
+    p = pair("def f(x):\n    return x", "def f(x):\n    return cleanse(x)")
+    js = pair("function f(x) { return x; }", "function f(x) { return cleanse(x); }", language="javascript", repo="owner/beta")
+    assert mine_pairs([p, js]).candidates == []
+
+
+def test_learn_mechanisms_guard_inside_branch_and_numbered_excerpt():
+    p = pair(
+        "  10  def f(x):\n  11      redirect(x)",
+        "  20  def f(x):\n  21      if is_safe_url(x):\n  22          redirect(x)",
+        family="untrusted_destination",
+    )
+    candidate = repeated(p).candidates[0]
+    assert candidate["form"] == "guard" and candidate["pattern"]["predicate"] == "truthy"
+
+
+@pytest.mark.skipif(importlib.util.find_spec("tree_sitter_php") is None, reason="PHP tree-sitter wheel unavailable")
+def test_learn_mechanisms_php_receiver_normalization():
+    p = pair("function f($x, $helper) { echo $x; }", "function f($x, $helper) { echo $helper->escape($x); }", language="php")
+    assert repeated(p).candidates[0]["pattern"] == {"callee": "escape", "argument": 0}

tests/test_learn_slice.py

@@ -150,7 +150,9 @@ def test_learn_slice_audit_reads_and_never_writes(audit_module: ModuleType, monk
 
     raw = b"   10  def f(x):\n   11      execute(x)\n"
     digest = hashlib.sha256(raw).hexdigest()
-    example = replace(corpus(groups=1)[0], excerpt_sha=digest, source="pairs", language="python", family="injection")
+    example = replace(
+        corpus(groups=1)[0], excerpt_sha=digest, source="pairs", language="python", family="injection", group="github.com/owner/name"
+    )
     monkeypatch.setattr(audit_module, "load_examples", lambda store: [example])
 
     class ReadOnly:
@@ -161,6 +163,8 @@ def test_learn_slice_audit_reads_and_never_writes(audit_module: ModuleType, monk
     units = [{"family": example.family, "label": example.label, "group": example.group, "unit": example.id, "pair": None, "fold": "f"}]
     report = audit_module.audit(ReadOnly(), units, expected_rows=1, expected_pairs=0)
     assert report["example_rows_read"] == report["matched_rows"] == report["excerpts_read"] == 1
+    assert "owner/name" not in json.dumps(report)
+    assert report["units"][0]["unit"] == example.id
     assert report["excerpt_bytes"] == len(raw)
     assert report["coverage"]["injection"]["flow"] == 1
 

tests/test_model_partitions.py

@@ -344,37 +344,3 @@ def test_unnamed_sharded_census_remains_explicitly_incomplete(tmp_path):
         assert "partition_file_census_unavailable" in graph.execution_diagnostics()
     finally:
         graph.cleanup()
-
-
-def test_mixed_javascript_typescript_pin_builds_and_queries_one_graph(tmp_path):
-    from openultrasast.model.partitions import build_partitions
-
-    sources = {"app.js": "function run(x) { return x; }", "handler.ts": "const x: number = 1;"}
-    for path, body in sources.items():
-        (tmp_path / path).write_text(body)
-    builds, queries = [], []
-
-    class Graph:
-        def run_batch(self, kind, requests):
-            queries.append(requests)
-            return {**{key: [] for key in requests}, "__census__": [{"file_names": list(sources)}]}
-
-    def build(root, language):
-        assert {p.name: p.read_text() for p in root.iterdir()} == sources
-        builds.append(language)
-        return Graph()
-
-    graph = build_partitions(tmp_path, build, None)
-    try:
-        assert builds == ["javascript"]
-        assert {p.language for p in graph.partitions} == {"javascript", "typescript"}
-        assert sum(p.source_bytes for p in graph.partitions) == sum(len(body.encode()) for body in sources.values())
-        assert "cross_partition_semantics_unresolved" not in graph.boundaries
-        assert graph.boundary_languages["typescript_property_support_unvalidated"] == ("typescript",)
-        requests = {path: {"file": path} for path in sources}
-        answers = graph.run_batch("taint", requests)
-        assert all(answers[path] == [] for path in sources)
-        assert queries == [requests]
-        assert not graph.execution_diagnostics()
-    finally:
-        graph.cleanup()

tests/test_model_specs.py

@@ -322,26 +322,3 @@ def test_a_sanitizer_that_does_not_stop_sql_injection_is_not_listed() -> None:
     assert "esc_html" not in sanitizers
     assert "$wpdb->prepare" in sanitizers, "the parameterized form does stop it"
     assert "esc_sql" in sanitizers
-
-
-def test_typescript_taint_specs_share_javascript_facts():
-    from openultrasast.model.specs import taint_specs
-
-    javascript = taint_specs(language="javascript")
-    assert javascript
-    assert taint_specs(language="typescript") == javascript
-
-
-def test_typescript_native_facts_take_precedence_over_alias():
-    from dataclasses import replace
-
-    from openultrasast.model.specs import taint_specs
-    from openultrasast.semantic.facts import load_facts
-
-    facts = load_facts()
-    native = replace(next(sink for sink in facts.sinks if sink.language == "javascript"), language="typescript", calls=("tsOnly",))
-    facts = replace(facts, sinks=(*facts.sinks, native))
-    specs = taint_specs(language="typescript", facts=facts)
-    assert specs
-    assert {spec.language for spec in specs.values()} == {"typescript"}
-    assert {sink for spec in specs.values() for sink in spec.sinks} == {"tsOnly"}

tests/test_plane_loop.py

@@ -311,15 +311,3 @@ def test_the_loop_runs_as_one_plane_run_and_the_gate_rejects_a_demotion_that_cos
     assert set(features) == {"lmdeploy/vl/media/connection.py::_load_http_url"}, "no fixed-pin alert became a candidate"
     assert all(r["pin"] == VULNERABLE and "in_fix_range" not in json.dumps(r["x"]) for r in features.values())
     assert fixed_alert.row["in_fix_range"] is True and fixed_alert.row["function"] == "load"
-
-
-def test_engine_languages_include_aliases_only_for_available_tables(tmp_path):
-    from openultrasast.plane.tasks.alerts import engine_languages
-
-    assert "typescript" in engine_languages()
-    semantic = tmp_path / "semantic"
-    semantic.mkdir()
-    (semantic / "python.toml").write_text("")
-    assert engine_languages(tmp_path) == {"python"}
-    (semantic / "javascript.toml").write_text("")
-    assert engine_languages(tmp_path) == {"python", "javascript", "typescript"}

tests/test_plane_profile.py

@@ -156,3 +156,27 @@ def test_profile_is_frozen_and_typed() -> None:
     assert isinstance(loaded, PlaneProfile)
     with pytest.raises(AttributeError):
         loaded.exec = "remote"  # type: ignore[misc]
+
+
+def test_kind_observability_defaults_off() -> None:
+    assert load_profile("kind", environ={}).kind_observability is False
+
+
+@pytest.mark.parametrize(("raw", "expected"), [("1", True), ("true", True), ("0", False), ("FALSE", False)])
+def test_kind_observability_environment_override(tmp_path: Path, raw: str, expected: bool) -> None:
+    path = write_profile(tmp_path, MINIMAL + f"kind_observability = {str(not expected).lower()}\n")
+    assert load_profile(str(path), environ={"OUSAST_KIND_OBSERVABILITY": raw}).kind_observability is expected
+
+
+def test_kind_observability_validation_and_print(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+) -> None:
+    path = write_profile(tmp_path, MINIMAL + 'kind_observability = "false"\n')
+    with pytest.raises(ProfileError, match="kind_observability must be a boolean"):
+        load_profile(str(path), environ={})
+    path = write_profile(tmp_path)
+    with pytest.raises(ProfileError, match="OUSAST_KIND_OBSERVABILITY must be"):
+        load_profile(str(path), environ={"OUSAST_KIND_OBSERVABILITY": "maybe"})
+    monkeypatch.setenv("OUSAST_KIND_OBSERVABILITY", "1")
+    assert profile_module.main(["--profile", str(path), "--print", "kind_observability"]) == 0
+    assert capsys.readouterr().out == "true\n"