feat(pre-push-safety-net): connect change context to ranked questions

ops/README.md

@@ -97,6 +97,18 @@ that the lower static-ranked risky function is selected, that the harmless funct
 recorded as deferred, and that selected IDs match completed query outcomes and per-family
 coverage. It uses one 900-second lab budget; it is not a mapper or hook-quality evaluation.
 
+`ops/smoke_context.py` uses the same invocation with its own filename. It creates local
+PHP/JavaScript Git histories with a removed caller guard and an unchanged helper-file sink, then feeds
+immutable snapshot change evidence into the real driver. It checks affected-question
+relationships, a controlled witness and preservation of the dirty checkout. Its lexical
+change attribution does not prove that the removed guard enforced security or that the
+finding is new; comparison and admission are separate stages.
+
+`ops/smoke_context_boundaries.py` is the production-query negative regression using
+the same invocation. Unknown consumers of a framework source, nested sink, or sanitizer
+name must retain unresolved context rather than become tier-zero exclusions. Direct
+modeled sink and sanitizer calls remain positive controls for the boundary projection.
+
 Why this exists: Joern was first installed here by hand with `curl` and `unzip`, which is fine once and
 unreproducible thereafter. The engine decides whether a finding is a `suspicion` or a `model_entailed`, and
 every ceiling number in `benchmarks/measurements` is only comparable against a fixed version — so the version

ops/smoke_context.py

@@ -0,0 +1,120 @@
+"""Real Git-to-engine change-context smoke; lexical attribution is not novelty proof."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import subprocess
+import tempfile
+import time
+from dataclasses import asdict
+from pathlib import Path
+
+from openultrasast.cpg.backend import JoernBackend
+from openultrasast.model.contracts import ExecutionBudget
+from openultrasast.model.regions import ScanRegion
+from openultrasast.model.scan import ScanBudget, scan_repository
+from openultrasast.push.snapshot import SnapshotAdapter
+
+
+def main() -> None:
+    budget = ExecutionBudget(time.monotonic() + 900, 2.0)
+    report: dict[str, object] = {
+        "purpose": "Committed removed-guard context reaches real ranked questions; not semantic novelty or admission proof",
+        "shared_budget_seconds": 900,
+        "languages": {},
+    }
+    languages = report["languages"]
+    assert isinstance(languages, dict)
+    with tempfile.TemporaryDirectory(prefix="ousast-context-smoke-") as temporary:
+        for language, filename, before, removed, helper_name, helper in (
+            (
+                "php",
+                "api.php",
+                b"<?php\nrequire_once 'helper.php';\nfunction handler($req) {\n  if (!authorized($req)) return;\n  helper($req);\n}\n",
+                b"  if (!authorized($req)) return;\n",
+                "helper.php",
+                b"<?php\nfunction helper($value) { eval($value); }\n",
+            ),
+            (
+                "javascript",
+                "api.js",
+                b"import { helper } from './helper.js';\nexport function handler(req) {\n"
+                b"  if (!authorized(req)) return;\n  helper(req);\n}\n",
+                b"  if (!authorized(req)) return;\n",
+                "helper.js",
+                b"export function helper(value) { eval(value); }\n",
+            ),
+        ):
+            repo = Path(temporary) / language
+            repo.mkdir()
+
+            def git(*args: str, root: Path = repo) -> bytes:
+                return subprocess.run(["git", "-C", str(root), *args], check=True, capture_output=True, timeout=10).stdout
+
+            def commit(message: str, path: str = filename) -> str:
+                git("add", path)
+                git("commit", "-m", message)
+                return git("rev-parse", "HEAD").decode().strip()
+
+            git("init")
+            git("config", "user.name", "Context smoke")
+            git("config", "user.email", "[email protected]")
+            git("config", "core.hooksPath", os.devnull)
+            source = repo / filename
+            source.write_bytes(before)
+            (repo / helper_name).write_bytes(helper)
+            git("add", helper_name)
+            base = commit("base")
+            after = before.replace(removed, b"")
+            source.write_bytes(after)
+            head = commit("removed guard")
+            source.write_bytes(b"uncommitted content\n")
+            index = (repo / ".git/index").read_bytes()
+            status = git("status", "--porcelain=v1", "-z")
+            adapter = SnapshotAdapter(repo)
+            (comparison,) = adapter.resolve_updates(f"refs/heads/topic {head} refs/heads/topic {base}\n").comparisons
+            context = adapter.compare(comparison, declaration_paths=(), budget=budget)
+            assert any(span.side == "base" for span in context.spans), context
+            with adapter.materialize(head, budget=budget) as snapshot:
+                data = (snapshot.root / filename).read_bytes()
+                assert data == after and data
+                assert (snapshot.root / helper_name).read_bytes() == helper
+                result = scan_repository(
+                    snapshot.root,
+                    (ScanRegion(filename, "handler", language, ("injection",), 1.0, "entry_point"),),
+                    backend=JoernBackend(heap_mb=1024),
+                    budget=ScanBudget(max_regions=1),
+                    execution_budget=budget,
+                    ranking_mode="evidence",
+                    unit="context-smoke",
+                    population_complete=True,
+                    change_context=context,
+                )
+                enriched = result.change_context
+                assert enriched is not None and enriched.relationships, "changed context did not reach any question"
+                assert result.scope is not None and result.scope.selected
+                selected = {question.identity for question in result.scope.selected}
+                assert any(relation.target in selected for relation in enriched.relationships)
+                assert result.findings, "controlled unchanged sink has no witness"
+                assert any(Path(finding.site.split(":", 1)[0]).name == helper_name for finding in result.findings), result.findings
+                languages[language] = {
+                    "base_revision": base,
+                    "head_revision": head,
+                    "source_bytes": len(data) + len(helper),
+                    "source_sha256": hashlib.sha256(data).hexdigest(),
+                    "helper_bytes": len(helper),
+                    "helper_sha256": hashlib.sha256(helper).hexdigest(),
+                    "input_context": context.to_payload(),
+                    "result": asdict(result),
+                }
+            assert not snapshot.root.exists()
+            assert source.read_bytes() == b"uncommitted content\n"
+            assert (repo / ".git/index").read_bytes() == index
+            assert git("status", "--porcelain=v1", "-z") == status
+    print(json.dumps(report, indent=2))
+
+
+if __name__ == "__main__":
+    main()

ops/smoke_context_boundaries.py

@@ -0,0 +1,64 @@
+"""Production-query regression: a known argument does not summarize an unknown callee."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import tempfile
+import time
+from dataclasses import asdict
+from pathlib import Path
+
+from openultrasast.cpg.backend import JoernBackend
+from openultrasast.model.contracts import ChangeContext, ChangedSpan, ExecutionBudget
+from openultrasast.model.regions import ScanRegion
+from openultrasast.model.scan import ScanBudget, scan_repository
+
+
+def main() -> None:
+    content = (
+        b"function unknownSource(req) { unknown(req.body); }\n"
+        b"function unknownSink(req) { unknown(eval(req.body)); }\n"
+        b'function unknownSanitizer(req) { unknown("escape"); }\n'
+        b"function knownSink(req) { eval(req.body); }\n"
+        b"function knownSanitizer(req) { escape(req.body); }\n"
+    )
+    with tempfile.TemporaryDirectory(prefix="ousast-context-boundary-") as temporary:
+        root = Path(temporary)
+        source = root / "api.js"
+        source.write_bytes(content)
+        assert source.read_bytes() == content
+        names = ("unknownSource", "unknownSink", "unknownSanitizer", "knownSink", "knownSanitizer")
+        result = scan_repository(
+            root,
+            tuple(ScanRegion("api.js", name, "javascript", ("injection",), 1.0, "entry_point") for name in names),
+            backend=JoernBackend(heap_mb=1024),
+            budget=ScanBudget(max_regions=5),
+            execution_budget=ExecutionBudget(time.monotonic() + 600, 2.0),
+            ranking_mode="evidence",
+            unit="context-boundaries",
+            population_complete=True,
+            change_context=ChangeContext("base", "head", ("api.js",), (), (ChangedSpan("api.js", 1, 5, "head"),), (), (), (), ()),
+        )
+        assert result.scope is not None
+        outcomes = {item.identity.function: item for item in result.question_outcomes}
+        for name in names[:3]:
+            assert name in outcomes, (name, "unknown call was excluded", result.scope)
+            assert outcomes[name].status == "unresolved", (name, outcomes[name])
+            question_id = outcomes[name].identity.question_id
+            assert any(
+                "dynamic_external_or_depth_context_unresolved" in gap and question_id in gap for gap in result.scope.unresolved_boundaries
+            )
+        assert outcomes["knownSink"].status == "completed", outcomes["knownSink"]
+        known_sanitizer = next(item for item in result.scope.deferred if item.identity.function == "knownSanitizer")
+        assert known_sanitizer.reason == "tier_zero", known_sanitizer
+        assert not any(known_sanitizer.identity.question_id in gap for gap in result.scope.unresolved_boundaries)
+        print(
+            json.dumps(
+                {"source_bytes": len(content), "source_sha256": hashlib.sha256(content).hexdigest(), "result": asdict(result)}, indent=2
+            )
+        )
+
+
+if __name__ == "__main__":
+    main()