engine: admit TypeScript (jssrc2cpg, JavaScript taint specs via one alias), .mts/.cts, mixed JS/TS builds one CPG

Implemented with Codex. Not merged: wall-clock deadline tests failed at host
load 47; to be re-checked against main at normal load.

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

.github/workflows/engine-image.yml

@@ -1,63 +0,0 @@
-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

@@ -1,80 +0,0 @@
-# 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,s3]"
+ && /venv/bin/pip install --no-cache-dir -e ".[semantic]"
 ENV PATH="/venv/bin:${PATH}"
 
 # Analysis never needs root.

README.md

@@ -363,81 +363,3 @@ 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,9 +126,8 @@ 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:
-    # 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):
+    # Recorded findings take precedence over the shared engine language coverage.
+    if language not in ENGINE_LANGUAGES and not findings:
         return NONE
     state = engine_for(result)
     if state is not None:

benchmarks/learn/engine_task_entry.py

@@ -1,173 +0,0 @@
-"""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

@@ -1,319 +0,0 @@
-"""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

@@ -1,259 +0,0 @@
-"""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

@@ -1,51 +0,0 @@
-"""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

@@ -1,52 +0,0 @@
-"""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

@@ -1,644 +0,0 @@
-"""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

@@ -1,315 +0,0 @@
-"""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

@@ -1,239 +0,0 @@
-"""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,9 +141,7 @@ 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": [
-            {k: r[k] for k in ("unit", "pair", "family", "label", "fold", "source", "status", "has_flow", "slice_length")} for r in rows
-        ],
+        "units": rows,
     }
 
 

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

@@ -1,326 +0,0 @@
-{
- "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,21 +17,6 @@ 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
@@ -372,7 +357,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 server VM. It stays a first-class local execution mode. Every change is proven here first.
+- `kind` is this laptop. 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.
@@ -383,7 +368,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 server VM)"]
+    subgraph k3s ["k3s cluster (the kind profile is the same picture on one laptop)"]
         subgraph axns ["Namespace ax-system"]
             axs["ax-server Deployment + Redis"]
         end
@@ -418,7 +403,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 the server VM (`ops/ax/receiver-service.yaml.tmpl`) belongs to the kind
+Today's EndpointSlice back to a laptop (`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 server VM (today).
+- `kind` on a laptop (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,42 +214,3 @@ 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,21 +17,6 @@ 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
@@ -206,7 +191,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 server VM and must change first (checked 2026-10-02):
+assume this laptop and must change first (checked 2026-10-02):
 
 | Assumption | Where | Needed in a real cluster |
 |---|---|---|

ops/ax/prepare-substrate.py

@@ -1,43 +0,0 @@
-#!/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

@@ -1,11 +0,0 @@
-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

@@ -1,44 +0,0 @@
-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,7 +15,6 @@ 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}"
 
@@ -38,68 +37,9 @@ 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
-  install_substrate
+  (cd "$SRC/substrate" && hack/install-ate-kind.sh --deploy-ate-system)
 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

@@ -1,85 +0,0 @@
-# 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

@@ -1,63 +0,0 @@
-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

@@ -1,22 +0,0 @@
-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

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

ops/k8s/engine/secret.example.yaml

@@ -1,9 +0,0 @@
-# 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:784bc6afc869c679c4966d04d3f9653a8d379726b899809590cc83327f4038e3"
+  "runner": "localhost:5001/ousast-runner@sha256:d2a55293999cc34581ef1f02a1ffb0c2dd8fc0cbe001ac952df81600e868f019"
 }

ops/k8s/profiles/kind.toml

@@ -1,9 +1,8 @@
-# The kind profile: the plane on this server VM's single-node kind cluster, which ops/ax/up.sh creates and which is
+# The kind profile: the plane on this laptop'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

@@ -1,52 +0,0 @@
-# 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,7 +33,6 @@ 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,8 +53,7 @@
     }
 
   val calls = {
-    // 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 all = cpg.call.filter(c => 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,8 +172,7 @@
   // 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.
-  // 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 allOperations = cpg.call.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 --------------------------
@@ -237,7 +236,7 @@
     if (fileS.isEmpty) Nil
     else
       fileMethods(fileS).flatMap { m =>
-        m.call.l.filter(c => Option(c.method).nonEmpty).flatMap { via =>
+        m.call.l.flatMap { via =>
           via.callee.l
             .filter(_.isExternal)
             .flatMap(stub => stubTarget(stub.fullName))

src/openultrasast/cpg/queries/taint.sc

@@ -56,8 +56,7 @@
     contextFilter: String = "",
     contextPaths: String = "",
     requests: String = "",
-    requestsFile: String = "",
-    trace: String = ""
+    requestsFile: String = ""
 ) = {
   importCpg(cpgFile)
 
@@ -226,7 +225,6 @@
   // 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
@@ -497,7 +495,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 if i.method != null =>
+      case i: io.shiftleft.codepropertygraph.generated.nodes.Identifier =>
         // 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
@@ -791,7 +789,7 @@
         case _ => false
       }
       ternary || (end :: elements.filter(stored)).distinctBy(_.id).exists {
-        case cfg: io.shiftleft.codepropertygraph.generated.nodes.CfgNode if cfg.method != null =>
+        case cfg: io.shiftleft.codepropertygraph.generated.nodes.CfgNode =>
           val method = cfg.method
           method.ast.isControlStructure.controlStructureType("IF").l.exists { cs =>
             cs.condition.l.headOption.exists { cond =>
@@ -899,7 +897,7 @@
   def fieldSourceNodes =
     if (fieldSrc != "true") Iterator.empty
     else {
-      val reads = cpg.call.nameExact(FIELD_ACCESS).filter(c => Option(c.method).exists(inScope)).l.filterNot(writeTarget)
+      val reads = cpg.call.nameExact(FIELD_ACCESS).filter(c => inScope(c.method)).l.filterNot(writeTarget)
       if (reads.isEmpty) Iterator.empty
       else reads.filter(r => taintedPrefixes(r.code.trim, fileS, r.method) && !overwrittenBefore(r)).iterator
     }
@@ -959,7 +957,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 => Option(r.method).exists(m => taintedPrefixes(r.code.trim, callback.filename, m)))
+      callback.ast.isCall.nameExact(FIELD_ACCESS).l.filter(r => taintedPrefixes(r.code.trim, callback.filename, r.method))
     framework ++ fields
   }
 
@@ -1014,7 +1012,7 @@
   def hookSourceNodes =
     if (hookTable.isEmpty || hookApply.isEmpty) Iterator.empty
     else {
-      val applies = cpg.call.filter(c => hookApply.contains(c.name)).filter(c => Option(c.method).exists(inScope)).l
+      val applies = cpg.call.filter(c => hookApply.contains(c.name)).filter(c => inScope(c.method)).l
       if (applies.isEmpty) Iterator.empty
       else {
         val keys = applies.flatMap { call =>
@@ -1025,7 +1023,7 @@
         else
           cpg.call
             .nameExact(INDEX_ACCESS)
-            .filter(c => Option(c.method).exists(inScope))
+            .filter(c => inScope(c.method))
             .l
             .filter(access => keys.contains(keyOf(access)))
             // ...and actually derived from the filtered value, not merely sharing a key name with it.
@@ -1109,7 +1107,7 @@
   def sinkCalls =
     sinkCandidatesMemo
       .getOrElseUpdate(sinksS, cpg.call.filter(c => sinkNames.exists(n => sinkMatches(c, n))).l)
-      .filter(c => Option(c.method).exists(inScope))
+      .filter(c => inScope(c.method))
 
   // ---- stage one of the two-stage join WITHOUT dataflow, for evidence mode (flow-aware-ranking, phase 2) --
   //
@@ -1171,7 +1169,7 @@
     )
 
   def fieldCarriedKind: String =
-    strongest(cpg.call.nameExact(FIELD_ACCESS).filter(c => Option(c.method).exists(inScope)).l.map(r => prefixKind(r.code.trim, fedFields(r.method.filename))))
+    strongest(cpg.call.nameExact(FIELD_ACCESS).filter(c => inScope(c.method)).l.map(r => prefixKind(r.code.trim, fedFields(r.method.filename))))
 
   def callbackFedKind(callback: String): String =
     callbackFedMemo.getOrElseUpdate(
@@ -1188,7 +1186,7 @@
   def hookCarriedKind: String =
     if (hookTable.isEmpty || hookApply.isEmpty) ""
     else
-      strongest(cpg.call.filter(c => hookApply.contains(c.name)).filter(c => Option(c.method).exists(inScope)).l.flatMap { call =>
+      strongest(cpg.call.filter(c => hookApply.contains(c.name)).filter(c => inScope(c.method)).l.flatMap { call =>
         val hook = call.argument.l.headOption.map(a => unquote(a.code)).getOrElse("")
         hookTable.getOrElse(hook, Nil).map(callbackFedKind)
       })
@@ -1234,7 +1232,7 @@
       else {
         val confirmed = cpg.call
           .nameExact(FIELD_ACCESS)
-          .filter(c => Option(c.method).exists(inScope))
+          .filter(c => inScope(c.method))
           .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"
@@ -1278,9 +1276,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 => Option(c.method).exists(inScope)).l.map(_.method.filename).toSet
+        cpg.call.nameExact(FIELD_ACCESS).filter(c => inScope(c.method)).l.map(_.method.filename).toSet
       val callbacks = if (hookApply.isEmpty) Nil else cpg.call.filter(c => hookApply.contains(c.name))
-        .filter(c => Option(c.method).exists(inScope)).l.flatMap { call =>
+        .filter(c => inScope(c.method)).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
@@ -1402,7 +1400,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 => Option(c.method).map(m => m.filename.split("/").last + ":" + m.name).getOrElse("?")
+          case c: io.shiftleft.codepropertygraph.generated.nodes.CfgNode => c.method.filename.split("/").last + ":" + c.method.name
           case _                                                          => "?"
         }
         s"$where:${node.lineNumber.getOrElse(-1)} [${node.label}] ${node.code.take(120)}"
@@ -1437,7 +1435,7 @@
           "sinkArg0Literal" -> arg0Literal,
           "bounded"         -> bounds.nonEmpty,
           "bound"            -> bounds.headOption.getOrElse("").take(120),
-          "inLabeledScope"  -> (function.isEmpty || Option(sink.method).exists(nestedInLabeled) || reachableMethods.contains(sink.method.fullName))
+          "inLabeledScope"  -> (function.isEmpty || nestedInLabeled(sink.method) || reachableMethods.contains(sink.method.fullName))
         )
         if (traceS == "true") row("trace") = traced(shortest)
         row
@@ -1491,12 +1489,6 @@
       // 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")
@@ -1540,7 +1532,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, traceS = trace): _*))
+      ujson.write(ujson.Arr(rowsFor(sources, sinks, sanitizers, hookCallbacks, dispatchApply, dispatchValue, function, parameterSources, parameterSources, fieldSources, evidenceOnly, file, callDepth, boundedSinks): _*))
     )
   }
   println("---OUSAST-CPG-END---")

src/openultrasast/cpg/queries/vocabulary.sc

@@ -51,13 +51,11 @@
   val sourceMemo = scala.collection.mutable.Map.empty[String, Boolean]
   def nearSource(c: io.shiftleft.codepropertygraph.generated.nodes.Call): Boolean =
     if (sourcePatterns.isEmpty) false
-    // CPS-IT/mailqueue Extension::KEY has a null .method; it cannot be near a method source.
-    else Option(c.method).exists { m =>
+    else
       sourceMemo.getOrElseUpdate(
-        m.fullName,
-        m.ast.isCall.exists(x => sourcePatterns.exists(p => boundedPattern(p).matcher(x.code).find()))
+        c.method.fullName,
+        c.method.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,12 +31,10 @@ 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
@@ -80,7 +78,6 @@ 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 = ""
@@ -202,34 +199,7 @@ class EngineSession:
         target = self.scratch / f"answer-{nonce}.txt"
         wrapped = _redirected(code, target, nonce)
         started = time.monotonic()
-        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
+        body = self._post(wrapped, timeout=remaining)
         seconds = time.monotonic() - started
         self.requests += 1
         if body is None:

src/openultrasast/learn/cli.py

@@ -15,14 +15,6 @@ 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)")
@@ -105,8 +97,6 @@ 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":
@@ -121,25 +111,6 @@ 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

@@ -1,612 +0,0 @@
-"""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,18 +82,6 @@ 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": "typescript",
+    "typescript": "javascript",
     "python": "python",
     "java": "java",
     "c": "c",
@@ -109,14 +109,19 @@ class PartitionGraph:
 
     def _dispatch(self, kind: str, requests: Mapping[str, Mapping[str, object]], method: str) -> dict[str, list[object]]:
         answers: dict[str, list[object]] = {}
-        languages = dict.fromkeys(self.path_languages.get(str(params.get("file", "")), "") for params in requests.values())
+
+        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())
         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 self.path_languages.get(str(params.get("file", ""))) == language}
+            selected = {rid: params for rid, params in requests.items() if graph_language(str(params.get("file", ""))) == language}
             if not selected:
                 continue
             batch = getattr(graph, method, None) or getattr(graph, "run_batch", None)
@@ -133,7 +138,12 @@ 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 = set(next(p.paths for p in self.partitions if p.language == language))
+                    expected = {
+                        path
+                        for p in self.partitions
+                        if p.language == language or (language == "javascript" and p.language == "typescript")
+                        for path in p.paths
+                    }
                     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)
@@ -234,10 +244,17 @@ def build_partitions(
                     graph.path_languages[relative] = languages[0]
                 elif languages:
                     boundaries.add("ambiguous_frontend_path")
-        if len(paths) > 1:
+        # 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:
             boundaries.add("cross_partition_semantics_unresolved")
         records: list[PartitionCoverage] = []
-        for language, sources in sorted(paths.items()):
+        for language, members in sorted(groups.items()):
+            sources = [source for member in members for source in paths[member]]
             relative_paths = tuple(sorted(p.relative_to(root).as_posix() for p in sources))
             frontend = FRONTENDS.get(language)
             if frontend is None:
@@ -247,7 +264,7 @@ def build_partitions(
                 continue
             scratch = Path(tempfile.mkdtemp(prefix="ousast-input-"))
             graph.roots[language] = scratch
-            size = 0
+            sizes: dict[str, int] = {}
             complete = True
             for source in sources:
                 if budget is not None and time.monotonic() >= budget.deadline_monotonic:
@@ -268,25 +285,31 @@ def build_partitions(
                             complete = False
                             break
                         output.write(chunk)
-                        size += len(chunk)
+                        relative = source.relative_to(root).as_posix()
+                        sizes[relative] = sizes.get(relative, 0) + 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", ()))))
-            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)
-            )
+            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
+                    )
+                )
         from dataclasses import replace
 
         graph.partitions = tuple(

src/openultrasast/model/regions.py

@@ -24,6 +24,7 @@ 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
 
@@ -41,7 +42,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 = {"typescript": "javascript", "c_cpp": "c"}
+_NORMALISE = {**TAINT_LANGUAGE_ALIAS, "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 SemanticFacts, load_facts
+from ..semantic.facts import TAINT_LANGUAGE_ALIAS, SemanticFacts, load_facts
 from ..semantic.obligations.facts import ObligationFacts, load_obligation_facts
 from .taxonomy import FamilyTaxonomy, load_families
 
@@ -245,7 +245,10 @@ 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.
     """
-    scoped = (facts if facts is not None else load_facts()).for_language(language)
+    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)
     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,7 +28,6 @@ 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.
 
@@ -41,7 +40,6 @@ 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 FactLoadError, load_facts
+from .semantic.facts import TAINT_LANGUAGE_ALIAS, 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", "typescript": "javascript"}.get(language, language)
+    return {"c_cpp": "c", "cpp": "c", **TAINT_LANGUAGE_ALIAS}.get(language, language)
 
 
 def _in_named_function(record: OverlayRecord, function: str, ranges: FunctionRanges | None) -> bool:

src/openultrasast/plane/memory.py

@@ -586,10 +586,6 @@ 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 server VM's kind cluster, ``k3s.toml`` for
+A profile is a TOML file under ``ops/k8s/profiles/`` (``kind.toml`` for the laptop'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,7 +59,6 @@ class PlaneProfile:
     name: str
     path: Path
     exec: str = "local"
-    kind_observability: bool = False
     kube_context: str = ""
     registry: str = ""
     images: Path = Path("images.json")
@@ -136,10 +135,6 @@ 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}")
@@ -177,10 +172,6 @@ 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)
@@ -250,8 +241,7 @@ 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}")
-    value = getattr(profile, head)
-    return str(value).lower() if isinstance(value, bool) else str(value)
+    return str(getattr(profile, head))
 
 
 def main(argv: list[str] | None = None) -> int:

src/openultrasast/plane/tasks/alerts.py

@@ -44,6 +44,7 @@ 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")
@@ -64,8 +65,9 @@ def quick_languages(directory: Path = DEFAULT_RULESET_DIR) -> frozenset[str]:
 
 
 def engine_languages(directory: Path = DEFAULT_RULESET_DIR) -> frozenset[str]:
-    """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"))
+    """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})
 
 
 def covers(covered: Iterable[str], language: str) -> bool:

src/openultrasast/preprocess.py

@@ -22,6 +22,8 @@ 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,14 +43,6 @@ _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(
     {
@@ -191,11 +183,6 @@ 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 ()
@@ -220,7 +207,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 in {"identifier", "name"}:
+        if child.type == "identifier":
             return _text(child)
         if child.type == "function_declarator":
             for grandchild in getattr(child, "named_children", ()) or ():
@@ -253,7 +240,7 @@ def _left_name(node: Any) -> str:
 def _identifiers(node: Any) -> tuple[str, ...]:
     names: list[str] = []
     for child in _walk(node):
-        if child.type in {"identifier", "variable_name"}:
+        if child.type == "identifier":
             names.append(_text(child))
     return tuple(names)
 

src/openultrasast/semantic/extra.py

@@ -17,7 +17,6 @@ _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,12 +15,14 @@ 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",
-    "typescript": "javascript",
+    **TAINT_LANGUAGE_ALIAS,
     "java": "java",
 }
 
@@ -145,8 +147,18 @@ 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 = _LANGUAGE_ALIASES.get(language, language)
+        key = self.taint_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,9 +21,8 @@ from openultrasast.pairs import load_pair_catalog
 
 ROOT = Path(__file__).resolve().parents[1]
 PAIRS = ROOT / "benchmarks" / "pairs"
-# 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")}
+# 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"}
 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]"
@@ -107,18 +106,10 @@ 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"}
 
 
-@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
+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
     cutoff, so memorisation can be measured on the post-cutoff pairs."""
-    for item in _raw(BATCHES[source_id]):
+    for item in _raw(BATCHES["advisory-fixes-2"]):
         assert item["license"] in PERMISSIVE, item["name"]
-        assert item["batch"] == batch, item["name"]
+        assert item["batch"] == "2026-10-03", 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

@@ -1,406 +0,0 @@
-"""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

@@ -1,130 +0,0 @@
-"""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,3 +1368,33 @@ 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,7 +20,6 @@ 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
 
@@ -29,18 +28,6 @@ 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,42 +343,3 @@ 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

@@ -1,291 +0,0 @@
-"""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

@@ -1,234 +0,0 @@
-"""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

@@ -1,190 +0,0 @@
-"""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 == "none"
-    assert builder.recorded_engine("typescript", [], result, False).state == "none"
+    assert builder.recorded_engine("typescript", [], None, True).state == "failed"
+    assert builder.recorded_engine("typescript", [], result, False).state == "ran"
     assert builder.recorded_engine("typescript", [], result, True).values["eng.findings"] == 0
 
 

tests/test_learn_mechanisms.py

@@ -1,289 +0,0 @@
-"""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,9 +150,7 @@ 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", group="github.com/owner/name"
-    )
+    example = replace(corpus(groups=1)[0], excerpt_sha=digest, source="pairs", language="python", family="injection")
     monkeypatch.setattr(audit_module, "load_examples", lambda store: [example])
 
     class ReadOnly:
@@ -163,8 +161,6 @@ 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,3 +344,37 @@ 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,3 +322,26 @@ 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,3 +311,15 @@ 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,27 +156,3 @@ 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"