diff --git a/.kiro/specs/pre-push-safety-net/implementation-group-3.md b/.kiro/specs/pre-push-safety-net/implementation-group-3.md new file mode 100644 index 0000000..427133f --- /dev/null +++ b/.kiro/specs/pre-push-safety-net/implementation-group-3.md @@ -0,0 +1,63 @@ +# Group 3 implementation verification + +Date: 2026-09-13. Scope: tasks 3.1–3.4. The maintainer authorized continuing this +group after the group 2 completion and next-step summary. Tasks remain unchecked +until independent review and fresh verification establish their completion. + +## Boundary + +This group integrates the existing backend and ranker with generic execution, +change-context and scope contracts. It does not implement new detector semantics, +alert admission, persistent caching, or hook installation. Equivalent normalized +PHP and JavaScript evidence must retain equivalent core decisions. Vendor sources +must remain outside both graph inputs and targets; missing external semantics are +coverage limits. + +## Verification state + +Implementation is in progress. No group-level completion claim is made yet. + +## Baseline + +Groups 1 and 2 completed at commit `948ee02`; the working tree was clean before +this group began. The prior final suite recorded 1,069 passing and nine skipped +tests. Three pre-existing mypy diagnostics concerned reuse of `candidates` in +`model/scan.py`; fresh checks will report whether they remain. + +Historical Joern query failures cannot be declared resolved merely because a +bounded failure returns promptly. Runtime smoke establishes readable source, +nonempty graph census and query payloads for its declared small fixtures. It does +not establish full-workload detection, actionable precision or hook latency. + +## Task 3.1 evidence + +The backend accepts an optional generic `ExecutionBudget`. Each build binds an +independent session to its graph callbacks; probes, builds, overlays, census, +queries and retries consume the same remaining time. Process startup also counts. +A timed-out stage prevents further retries. The budgeted process path kills the +original process group even after its launcher exits, bounds reaping and owned +scratch cleanup, and exposes incomplete cleanup explicitly. The driver releases +graph ownership on exceptions as well as normal return. Legacy backends cannot +silently discard a supplied deadline; optional unbounded judge calls are withheld +with an explicit reason. Ordinary scans retain their existing budget behavior. + +`benchmarks/measurements/2026-09-13-engine-budget-smoke.json` records the locally +rebuilt non-root image and exact packaged backend/driver hashes. Under one +900-second lab budget, native PHP read 45 source bytes and JavaScript read 48; +both produced populated graphs, dataflow overlays and exact source-node witnesses. +Unreadable PHP source was explicitly refused. PHP took 73.620 seconds and JavaScript +34.756 seconds in this small smoke; these are runtime checks, not hook latency. + +An independent controlled process probe read its 241-byte input script, launched +a parent and child ignoring SIGTERM, and observed termination in 0.804 seconds +under a 0.8-second deadline plus 0.8-second cancellation allowance. Signal delivery +may briefly leave a child visible immediately after return; the probe polls only +within the original allowance. Zombies count as terminated, not running descendants. +Concrete reusable graph leases and persisted completion records remain later tasks. + +The three former mypy errors were resolved by renaming the evidence-region list +inside the driver; this does not change the ranker. Global mypy now checks all +108 source files successfully. Independent review APPROVED task 3.1. Its fresh final suite passed: 1,089 tests, +nine skipped, 96.85 seconds; global Ruff, formatting (219 files), mypy and diff +checks passed. Verification status: VERIFIED, claim type TASK. This completes +8/27 tasks; tasks 3.2–3.4 remain in progress for this group. diff --git a/.kiro/specs/pre-push-safety-net/spec.json b/.kiro/specs/pre-push-safety-net/spec.json index 183bd75..c12de36 100644 --- a/.kiro/specs/pre-push-safety-net/spec.json +++ b/.kiro/specs/pre-push-safety-net/spec.json @@ -1,7 +1,7 @@ { "feature_name": "pre-push-safety-net", "created_at": "2026-09-12T15:35:00Z", - "updated_at": "2026-09-12T19:09:25.547910Z", + "updated_at": "2026-09-13T08:02:59.282414+00:00", "language": "en", "phase": "implementation", "approvals": { @@ -19,7 +19,7 @@ } }, "ready_for_implementation": true, - "implementation_approval_note": "Groups 1 and 2 explicitly authorized by the maintainer and verified complete after independent review. Remaining task plan retains approval, with execution sequenced by the spec. Implementation of the complete hook is still pending.", + "implementation_approval_note": "Groups 1 and 2 verified complete. Maintainer authorized continuing group 3; remaining approved tasks execute in dependency order with independent review and fresh verification.", "depends_on": [ "contributor-scan" ], @@ -45,6 +45,11 @@ "at": "2026-09-12T18:32:19.033933Z", "source": "ok, next kiro phase, approved; correct. we can continue with the spec", "action": "Authorized group 2 (2.1\u20132.3), resumed after local Claude Mem repair. Independent review and fresh verification remain required." + }, + { + "at": "2026-09-13T07:48:02.155687Z", + "source": "continue with the task", + "action": "Authorized continuing group 3 (3.1\u20133.4) after the group 3 next-step summary. Independent task review and fresh verification remain required." } ], "task_review": { @@ -64,13 +69,15 @@ "1.4", "2.1", "2.2", - "2.3" + "2.3", + "3.1" ], - "remaining_tasks": 20, - "verification_record": "implementation-group-2.md", + "remaining_tasks": 19, + "verification_record": "implementation-group-3.md", "verification_records": [ "implementation-group-1.md", - "implementation-group-2.md" + "implementation-group-2.md", + "implementation-group-3.md" ], "next_group": 3 } diff --git a/.kiro/specs/pre-push-safety-net/tasks.md b/.kiro/specs/pre-push-safety-net/tasks.md index 24581a5..d3d7f02 100644 --- a/.kiro/specs/pre-push-safety-net/tasks.md +++ b/.kiro/specs/pre-push-safety-net/tasks.md @@ -1,7 +1,7 @@ # Implementation Plan: pre-push-safety-net Status: task plan approved by the maintainer's "approve." on 2026-09-12. Groups 1 and 2 (tasks 1.1–2.3) -are complete after independent review and fresh verification. Groups 3–8 remain pending; +are complete after independent review and fresh verification. Group 3 is in progress (3.1 verified); groups 4–8 remain pending; the hook and capability admission are not yet implemented. See `implementation-group-1.md` and `implementation-group-2.md`. ## Execution contract @@ -79,13 +79,14 @@ the hook and capability admission are not yet implemented. See `implementation-g - _Verified 2026-09-12:_ independent review APPROVED; fresh final suite 1,069 passed, 9 skipped; focused 148 passed. Global Ruff/format, touched mypy and regression gates passed. PHP/JavaScript fixtures retain removed spans and unchanged-operation anchors across declaration renames; uncertain moves, repeated lines, binary and unavailable inputs remain unresolved. Forty independent real-Git comparisons checked exact spans/anchors. Network-disabled image smoke preserved committed PHP/JavaScript bytes and dirty non-HEAD state, with cleanup verified; see `implementation-group-2.md`. Lexical anchors do not establish semantic impact or novelty. - [ ] 3. Integrate bounded execution and ranker scope with the existing engine -- [ ] 3.1 (P) Enforce one remaining-time budget in the backend +- [x] 3.1 (P) Enforce one remaining-time budget in the backend - Thread the absolute deadline through readability probes, frontend builds, overlays, census, query batches, fallbacks and retries; no phase gets a fresh full budget. - Bound process-group termination and lease/scratch cleanup by the cancellation allowance, replacing any longer fixed cleanup waits on the push path. - Done when controlled hanging subprocesses at build/overlay/query stages terminate without orphan children inside the declared allowance, while ordinary scans preserve their prior behavior. - _Boundary: Cache and Engine Boundary, generic driver integration_ - _Depends: 1.4_ - _Requirements: 4.1, 4.2, 7.3_ + - _Verified 2026-09-13:_ independent review APPROVED; fresh suite 1,089 passed, nine skipped; global Ruff/format and mypy (108 files) passed. Real hanging probe/frontend/overlay/census/query processes obey the shared deadline and cancellation allowance; retries do not reset it. Non-root packaged PHP/JavaScript graph/query smoke passed with matching source hashes under one 900-second lab budget. See `implementation-group-3.md` and `benchmarks/measurements/2026-09-13-engine-budget-smoke.json`. Runtime readiness only; no hook latency claim. Former three mypy errors resolved by a behavior-preserving variable rename. - [ ] 3.2 Return the ranker's exact selected and deferred work - Integrate the existing ranker through explicit mode selection and shared question identity; record selected/deferred IDs, evidence and reasons directly from the final order. - Preserve unanswered evidence as unknown rather than tier zero; report per-family coverage so taint ranking cannot masquerade as authorization/configuration coverage. diff --git a/benchmarks/measurements/2026-09-13-engine-budget-smoke.json b/benchmarks/measurements/2026-09-13-engine-budget-smoke.json new file mode 100644 index 0000000..6dd4507 --- /dev/null +++ b/benchmarks/measurements/2026-09-13-engine-budget-smoke.json @@ -0,0 +1,85 @@ +{ + "recorded_at": "2026-09-13T08:00:44.226437+00:00", + "task": "3.1", + "image_id": "sha256:f422c603db9f3a1f5e33f255b36157ac0d660b05a3845eb8b7bb87f1fb265ce7", + "build_exit_code": 0, + "smoke_exit_code": 0, + "packaged_source_hashes": { + "openultrasast.cpg.backend": "2e327d26958efd592932bbe5a84aa122d969c0e49a1c13469a4d956b621a9a34", + "openultrasast.model.scan": "3ceed02c315893c5e54b1e175faaf261eee3883fd20ca85a013ffc9d9a6650f2" + }, + "packaged_sources_match_workspace": true, + "smoke_script_sha256": "710a54e2a208bb71263771192d67a82e934526632e88f6aa626de9950fa80128", + "runtime": { + "joern_version": "4.0.625", + "php_version": "PHP 8.3.6 (cli) (built: Sep 2 2026 12:56:02) (NTS)", + "php_parser_bytes": 43, + "purpose": "runtime and query smoke; does not qualify vulnerability detection or hook latency", + "shared_budget_seconds": 900.0, + "languages": { + "php": { + "frontend": "php2cpg", + "source_bytes": 45, + "source_sha256": "f744d636071c3e5a8e72ef33fa746db22e705dfdc66dba98b5238875054d801a", + "graph_bytes": 49509, + "census": { + "files": "2", + "methods": "3", + "overlays": "base,controlflow,typerel,callgraph,dataflowOss", + "maxHeapMB": "1024" + }, + "source_witness": { + "files": [ + "probe.php", + "" + ], + "handlers": 1, + "evalCalls": 1 + }, + "elapsed_seconds": 73.62 + }, + "javascript": { + "frontend": "jssrc2cpg", + "source_bytes": 48, + "source_sha256": "10c0c21a6e2f988119f305809af1f98e1ff39fecb9f40d6cd3dfa05301e0015b", + "graph_bytes": 71165, + "census": { + "files": "1", + "methods": "5", + "overlays": "base,controlflow,typerel,callgraph,dataflowOss", + "maxHeapMB": "1024" + }, + "source_witness": { + "files": [ + "probe.js" + ], + "handlers": 1, + "evalCalls": 1 + }, + "elapsed_seconds": 34.756 + } + }, + "unreadable_source": { + "status": "explicit_failure", + "reason": "the php interpreter cannot read /tmp/ousast-engine-smoke-npgsdp9_/php/probe.php, so the frontend would parse nothing and write an empty graph. Check that php is a real interpreter with access to this tree -- a container or wrapper without the right mount fails exactly this way." + } + }, + "controlled_process_probe": { + "source_bytes": 241, + "source_sha256": "a7c6471185a8a05a8b812fa8dfb878dde321aefca03ad046d3bc5465d2cf4c14", + "deadline_seconds": 0.8, + "cancellation_allowance_seconds": 0.8, + "elapsed_seconds": 0.8036922400351614, + "process_states": [ + "absent", + "Z" + ], + "result": "explicit_failure", + "purpose": "controlled subprocess termination; not Joern detection or hook latency" + }, + "limits": [ + "Small runtime fixtures, not full-workload detection or hook latency.", + "Terminated zombie processes are not running descendants; no claim of reaping grandchildren.", + "900-second lab budget, not the 30-second hook acceptance target." + ] +} diff --git a/ops/README.md b/ops/README.md index be18021..4441972 100644 --- a/ops/README.md +++ b/ops/README.md @@ -85,6 +85,11 @@ PHP file. It prints JSON only after all checks pass; a failed build/query raises exit. Versions, source hashes/byte counts and timings describe this small runtime smoke, not detection quality or a pre-push latency guarantee. Keep the image's default non-root user for the permission test. +To exercise the same smoke through one shared backend deadline, add +`-e OUSAST_ENGINE_SMOKE_BUDGET_SECONDS=900` to the Docker command. Both language builds and +all census/witness queries consume that same budget. The generous smoke allowance verifies +runtime integration; it does not represent the hook's target latency. + 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 diff --git a/ops/smoke_engine.py b/ops/smoke_engine.py index d7d4492..21093f3 100644 --- a/ops/smoke_engine.py +++ b/ops/smoke_engine.py @@ -13,6 +13,7 @@ import time from pathlib import Path from openultrasast.cpg.backend import DATAFLOW_OVERLAY, JoernBackend, joern_version +from openultrasast.model.contracts import ExecutionBudget def require(condition: object, message: str) -> None: @@ -22,6 +23,8 @@ def require(condition: object, message: str) -> None: def main() -> None: require(os.getuid() != 0, "run as the image's non-root user so unreadable-file checks are meaningful") + budget_seconds = float(os.environ.get("OUSAST_ENGINE_SMOKE_BUDGET_SECONDS", "0")) + execution_budget = ExecutionBudget(time.monotonic() + budget_seconds, 2.0) if budget_seconds > 0 else None version = joern_version() require(version, "installed Joern version is unknown") php = Path(shutil.which("php") or "/missing-php").resolve() @@ -39,6 +42,7 @@ def main() -> None: "php_version": subprocess.run([str(php), "-v"], check=True, capture_output=True, text=True, timeout=15).stdout.splitlines()[0], "php_parser_bytes": php_bytes(parser), "purpose": "runtime and query smoke; does not qualify vulnerability detection or hook latency", + "shared_budget_seconds": budget_seconds or None, "languages": {}, } languages = report["languages"] @@ -72,7 +76,7 @@ def main() -> None: if language == "php": require(php_bytes(source) == len(data), "PHP and host source byte counts differ") print(f"{language}: opened {filename}, {len(data)} bytes; building graph", file=sys.stderr, flush=True) - backend = JoernBackend(build_timeout=180, query_timeout=180, heap_mb=1024) + backend = JoernBackend(build_timeout=180, query_timeout=180, heap_mb=1024, execution_budget=execution_budget) started = time.monotonic() graph = backend.build(source_root, language=language) require(graph, f"{language} graph build failed: {backend.last_failure}") @@ -84,7 +88,9 @@ def main() -> None: assert isinstance(census, dict) require(int(census["files"]) > 0 and int(census["methods"]) > 0, f"empty {language} census") require(DATAFLOW_OVERLAY in str(census["overlays"]).split(","), f"{language} missing dataflow overlay: {census}") - witness = JoernBackend(queries_dir=queries, query_timeout=180, heap_mb=1024).query(graph.cpg_path, "source_witness", {}) + witness = JoernBackend(queries_dir=queries, query_timeout=180, heap_mb=1024, execution_budget=execution_budget).query( + graph.cpg_path, "source_witness", {} + ) require(isinstance(witness, dict), f"{language} source witness unanswered") assert isinstance(witness, dict) require(any(str(path).endswith(filename) for path in witness["files"]), f"{language} source absent from graph") diff --git a/src/openultrasast/cpg/backend.py b/src/openultrasast/cpg/backend.py index c4c8de2..eb4637f 100644 --- a/src/openultrasast/cpg/backend.py +++ b/src/openultrasast/cpg/backend.py @@ -25,14 +25,17 @@ import os import re import shutil import signal +import stat import subprocess import tempfile import time from collections.abc import Callable, Mapping, Sequence -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace from pathlib import Path from typing import Protocol +from ..model.contracts import ExecutionBudget + logger = logging.getLogger(__name__) QUERIES_DIR = Path(__file__).resolve().parent / "queries" @@ -239,6 +242,7 @@ class CpgResult: # request files and joern's own workspace copy; nothing reclaimed it, and a day of scanning left 1,055 # directories and 355MB behind on a disk that was already at 95%. The driver disposes in a `finally`. cleanup: Callable[[], None] | None = None + execution_diagnostics: Callable[[], tuple[str, ...]] | None = None class CpgBackend(Protocol): @@ -272,13 +276,41 @@ class JoernBackend: queries_dir: Path = field(default_factory=lambda: QUERIES_DIR) # Why the last build refused, so the driver can name it instead of reporting a bare `cpg_build_failed`. last_failure: str = "" + execution_budget: ExecutionBudget | None = None + _diagnostics: list[str] = field(default_factory=list, init=False, repr=False) + _scratch: list[Path] = field(default_factory=list, init=False, repr=False) + _cancel_deadline: float | None = field(default=None, init=False, repr=False) def available(self) -> bool: from .capability import has_cpg return has_cpg() - def build(self, root: Path, *, language: str = "", exclude: Sequence[str] = ()) -> CpgResult | None: + def build( + self, root: Path, *, language: str = "", exclude: Sequence[str] = (), execution_budget: ExecutionBudget | None = None + ) -> CpgResult | None: + """Bind an independent graph session to the caller's absolute transaction deadline.""" + budget = execution_budget if execution_budget is not None else self.execution_budget + if budget is None: + return self._build_impl(root, language=language, exclude=exclude) + session = replace(self, execution_budget=budget, last_failure="") + result = None + try: + session._checkpoint() + result = session._build_impl(root, language=language, exclude=exclude) + session._checkpoint() + return result + except _DeadlineExpired: + result = None + return None + finally: + if result is None: + for scratch in session._scratch: + session._cleanup(scratch, final=True) + self.last_failure = session.last_failure + self._diagnostics = list(session._diagnostics) + + def _build_impl(self, root: Path, *, language: str = "", exclude: Sequence[str] = ()) -> CpgResult | None: """Build a CPG for ``root``. ``None`` on any failure — the caller degrades, never guesses. ``language`` lets a failed ``joern-parse`` retry through the frontend directly, which is what Joern @@ -287,10 +319,13 @@ class JoernBackend: parse = shutil.which("joern-parse") if parse is None and os.environ.get("OPENULTRASAST_JOERN_PROBE", "").strip().lower() not in {"1", "on", "true", "yes"}: return None - _sweep_stale_scratch() + if self.execution_budget is None: + _sweep_stale_scratch() scratch = Path(tempfile.mkdtemp(prefix=SCRATCH_PREFIX)) + if self.execution_budget is not None: + self._scratch.append(scratch) cpg_path = scratch / "cpg.bin" - dispose = lambda: shutil.rmtree(scratch, ignore_errors=True) # noqa: E731 -- one expression, named + dispose = lambda: self._cleanup(scratch, final=True) # noqa: E731 -- one expression, named # THE PRECONDITION. Before anything is built, make the toolchain open one file from the tree and say # how big it is. Every instrument failure this project has had takes the same shape -- the input was @@ -307,6 +342,7 @@ class JoernBackend: if unreadable: logger.error("cpg build refused for %s: %s", root, unreadable) self.last_failure = unreadable + dispose() return None # A frontend whose failures are visible and retryable is worth more than one whose are not. @@ -329,6 +365,7 @@ class JoernBackend: run_batch=lambda query, requests: self.query_batch_across(shards, query, requests), unparsed=unparsed, cleanup=dispose, + execution_diagnostics=lambda: tuple(self._diagnostics), ) command = [parse or "joern-parse", self._heap_flag(), str(root), "--output", str(cpg_path)] completed = self._run(command, timeout=self.build_timeout, cwd=scratch) @@ -350,6 +387,7 @@ class JoernBackend: run_batch=lambda query, requests: self.query_batch(cpg_path, query, requests), unparsed=unparsed, cleanup=dispose, + execution_diagnostics=lambda: tuple(self._diagnostics), ) def _files_too_large_to_flow(self, root: Path, language: str) -> tuple[str, ...]: @@ -358,7 +396,7 @@ class JoernBackend: if not extensions: return () found: list[str] = [] - for path in sorted(root.rglob("*")): + for path in self._paths(root, "*"): if path.suffix.lower() not in extensions or not path.is_file(): continue try: @@ -381,7 +419,7 @@ class JoernBackend: if version is not None and version >= CLOSURE_DEFECT_FIXED_IN: return () # fixed upstream; nothing to exclude and every file to keep found: list[str] = [] - for path in sorted(root.rglob("*.php")): + for path in self._paths(root, "*.php"): if not path.is_file(): continue try: @@ -394,6 +432,8 @@ class JoernBackend: continue depth, index = 0, opening while index < len(text): + if index % 4096 == 0: + self._checkpoint() if text[index] == "{": depth += 1 elif text[index] == "}": @@ -421,7 +461,7 @@ class JoernBackend: binary = shutil.which(interpreter) if binary is None: return "" # absent is a different failure, and the frontend reports it plainly - sample = next((path for path in sorted(root.rglob("*")) if path.suffix.lower() in extensions and path.is_file()), None) + sample = next((path for path in self._paths(root, "*") if path.suffix.lower() in extensions and path.is_file()), None) if sample is None: return "" # nothing of this language to read; not the interpreter's fault probe = "$f = $argv[1]; if (!is_readable($f) || filesize($f) < 1) { exit(3); } exit(0);" @@ -545,7 +585,7 @@ class JoernBackend: extensions = _LANGUAGE_EXTENSIONS.get(language.lower()) if not extensions: return None - expected = sum(1 for path in root.rglob("*") if path.suffix.lower() in extensions and path.is_file()) + expected = sum(1 for path in self._paths(root, "*") if path.suffix.lower() in extensions and path.is_file()) if not expected: return (0, 0) # nothing to count is not a failed count: an empty tree's graph is complete payload = self.query(cpg_path, "census", {}) @@ -598,6 +638,7 @@ class JoernBackend: island = scratch / "excluded-root" made = 0 for name in files: + self._checkpoint() source = Path(name) if not source.is_absolute(): source = root / name @@ -624,6 +665,9 @@ class JoernBackend: for shard in shards: rows = self.query(shard, query, params) if rows is None: + if self.execution_budget is not None: + self._note("shard_query_incomplete") + return None continue answered += 1 if isinstance(rows, list): @@ -645,6 +689,9 @@ class JoernBackend: for shard in shards: one = self.query_batch(shard, query, requests) if one is None: + if self.execution_budget is not None: + self._note("shard_query_incomplete") + return None continue answered += 1 census = one.pop("__census__", None) @@ -692,7 +739,7 @@ class JoernBackend: saved.replace(cpg_path) return True finally: - shutil.rmtree(workspace, ignore_errors=True) + self._cleanup(workspace) def _build_with_retries( self, root: Path, cpg_path: Path, scratch: Path, language: str, exclude: Sequence[str] = () @@ -716,6 +763,7 @@ class JoernBackend: """ best: tuple[str, ...] | None = None for attempt in range(FRONTEND_BUILD_ATTEMPTS): + self._checkpoint() unparsed = self._build_with_frontend(root, cpg_path, scratch, language, exclude) if unparsed is None: return best # the frontend is missing or failed outright; keep any earlier partial build @@ -773,6 +821,8 @@ class JoernBackend: point -- "no rows" and "could not ask" look identical to a caller that only sees a dict, and a 117k-line PHP CPG that threw while loading turned a failed scan into a clean bill of health. """ + if not self._time_available(): + return None script = self.queries_dir / f"{query}.sc" if not script.is_file(): logger.warning("no such cpg query: %s", script) @@ -786,6 +836,8 @@ class JoernBackend: # {} -- and {} reads as "no rows found". The scan reported 500 regions examined and nothing found, in # 0.05 seconds of taint query, which is a quiet failure of exactly the kind the rung ladder cannot # catch because no verdict was ever produced to label. + if not self._time_available(): + return None requests_file = cpg_path.parent / f"{query}-requests.json" try: requests_file.write_text(payload) @@ -810,6 +862,8 @@ class JoernBackend: logger.warning("cpg batch %s failed: %s", query, detail) return None parsed = extract_payload(completed.stdout or "") + if not self._time_available(): + return None if not isinstance(parsed, Mapping): # Unparseable output is a failure, not an empty answer. A 117k-line PHP CPG that throws while # loading printed a stack trace and no payload, and returning {} made the whole scan read as a @@ -821,6 +875,8 @@ class JoernBackend: def query(self, cpg_path: Path, query: str, params: Mapping[str, object]) -> object | None: """Run a shipped CPGQL script against a built CPG and return its fenced JSON payload.""" + if not self._time_available(): + return None script = self.queries_dir / f"{query}.sc" if not script.is_file(): logger.warning("no such cpg query: %s", script) @@ -835,7 +891,96 @@ class JoernBackend: detail = (completed.stderr or "")[-400:] if completed is not None else "timeout" logger.warning("cpg query %s failed: %s", query, detail) return None - return extract_payload(completed.stdout or "") + parsed = extract_payload(completed.stdout or "") + return parsed if self._time_available() else None + + def _note(self, reason: str) -> None: + if reason not in self._diagnostics: + self._diagnostics.append(reason) + self.last_failure = reason + + def _time_available(self) -> bool: + if self._cancel_deadline is not None: + return False + if self.execution_budget is not None and time.monotonic() >= self.execution_budget.deadline_monotonic: + self._note("deadline_exhausted") + return False + return True + + def _checkpoint(self) -> None: + if not self._time_available(): + raise _DeadlineExpired + + def _paths(self, root: Path, pattern: str): # type: ignore[no-untyped-def] + paths = root.rglob(pattern) + if self.execution_budget is None: + yield from sorted(paths) + else: + self._checkpoint() + for path in paths: + self._checkpoint() + yield path + + def _cleanup_limit(self) -> float: + assert self.execution_budget is not None + ceiling = self.execution_budget.deadline_monotonic + self.execution_budget.cancellation_allowance_seconds + return min(ceiling, self._cancel_deadline) if self._cancel_deadline is not None else ceiling + + def _cleanup(self, root: Path, *, final: bool = False) -> None: + if self.execution_budget is None: + shutil.rmtree(root, ignore_errors=True) + return + deadline = self._cleanup_limit() if final or self._cancel_deadline is not None else self.execution_budget.deadline_monotonic + try: + _remove_owned_tree(root, deadline) + except (OSError, _DeadlineExpired): + self._note("scratch_cleanup_incomplete") + + def _run_bounded(self, command: list[str], *, timeout: float, cwd: Path | None) -> subprocess.CompletedProcess[str] | None: + assert self.execution_budget is not None + if not self._time_available(): + return None + remaining = min(timeout, self.execution_budget.deadline_monotonic - time.monotonic()) + if remaining <= 0: + self._note("deadline_exhausted") + return None + process = None + try: + if self.runner is not None: + result = self.runner(command, capture_output=True, text=True, timeout=remaining, check=False, cwd=str(cwd) if cwd else None) + else: + process = subprocess.Popen( # noqa: S603 -- internal argument array + command, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + cwd=str(cwd) if cwd else None, + env=self._jvm_env(), + start_new_session=True, + ) + remaining = min(remaining, max(0.0, self.execution_budget.deadline_monotonic - time.monotonic())) + stdout, stderr = process.communicate(timeout=remaining) + result = subprocess.CompletedProcess(command, process.returncode, stdout, stderr) + if not self._time_available(): + return None + return result + except subprocess.TimeoutExpired: + self._cancel_deadline = min(self._cleanup_limit(), time.monotonic() + self.execution_budget.cancellation_allowance_seconds) + self._note("deadline_exhausted" if time.monotonic() >= self.execution_budget.deadline_monotonic else "process_timeout") + return None + except (OSError, subprocess.SubprocessError): + self._note("process_failed") + return None + finally: + if process is not None: + # No Popen context manager: its implicit wait has no timeout. Always signal + # the original group ID, even if the launcher already exited, so descendants + # retaining pipes (or closing them) cannot survive it. + if not _terminate_bounded_group(process, self._cleanup_limit()): + self._note("process_cleanup_incomplete") + for stream in (process.stdout, process.stderr): + if stream is not None: + stream.close() def _heap_mb(self) -> int: if self.heap_mb > 0: @@ -874,6 +1019,8 @@ class JoernBackend: row it would OOM rather than merely degrade -- and it would look like the SCAN needing more memory, which is the wrong lesson entirely. """ + if self.execution_budget is not None: + return self._run_bounded(command, timeout=timeout, cwd=cwd) run = self.runner if self.runner is not None else subprocess.run if self.runner is not None: # tests inject a fake runner and never spawn anything try: @@ -900,6 +1047,64 @@ class JoernBackend: return None +class _DeadlineExpired(Exception): + """Internal cooperative boundary; public builds/queries retain failure semantics.""" + + +def _terminate_bounded_group(process: subprocess.Popen[str], deadline: float) -> bool: + # SIGKILL immediately leaves the shared allowance for reaping and scratch cleanup. + # Unlike the legacy TERM/wait loop this also kills grandchildren after leader exit. + try: + os.killpg(process.pid, signal.SIGKILL) + except ProcessLookupError: + pass + except OSError: + return False + try: + process.wait(timeout=max(0.0, deadline - time.monotonic())) + except subprocess.TimeoutExpired: + return False + return True + + +def _remove_owned_tree(root: Path, deadline: float) -> None: + """Bounded Linux dirfd cleanup, never following substituted directory links.""" + + def check() -> None: + if time.monotonic() >= deadline: + raise _DeadlineExpired + + def contents(fd: int) -> None: + with os.scandir(fd) as entries: + for entry in entries: + check() + info = os.stat(entry.name, dir_fd=fd, follow_symlinks=False) + if stat.S_ISDIR(info.st_mode): + child = os.open(entry.name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=fd) + try: + contents(child) + finally: + os.close(child) + check() + os.rmdir(entry.name, dir_fd=fd) + else: + os.unlink(entry.name, dir_fd=fd) + + if not root.exists() and not root.is_symlink(): + return + check() + if root.is_symlink(): + root.unlink() + return + fd = os.open(root, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) + try: + contents(fd) + finally: + os.close(fd) + check() + root.rmdir() + + SCRATCH_PREFIX = "ousast-cpg-" # How long an abandoned scratch tree may survive. Long enough that a scan running in another process is never # disturbed, short enough that a machine scanning all day does not fill its disk. diff --git a/src/openultrasast/model/scan.py b/src/openultrasast/model/scan.py index 24c2305..9de45e6 100644 --- a/src/openultrasast/model/scan.py +++ b/src/openultrasast/model/scan.py @@ -18,6 +18,7 @@ the region's risk score said beforehand. from __future__ import annotations +import inspect import logging import time from collections.abc import Mapping, Sequence @@ -28,6 +29,7 @@ from typing import Any from ..cpg.backend import CpgResult from .candidates import enumerate_candidates from .config_value import request_params as config_params +from .contracts import ExecutionBudget from .dominance import request_params as dominance_params from .evidence import TIER_EXCLUDE, Evidence, evidence_from_rows from .ladder import Rung @@ -127,6 +129,49 @@ def scan_repository( client: Any | None = None, model: str = "", budget: ScanBudget | None = None, + execution_budget: ExecutionBudget | None = None, +) -> ModelScanResult: + """Run with an optional transaction deadline; graph ownership ends on every exit.""" + started = time.monotonic() + owned: list[Any] = [] + try: + result = _scan_repository_impl( + root, + regions, + backend=backend, + client=client if execution_budget is None else None, + model=model, + budget=budget, + execution_budget=execution_budget, + owned=owned, + ) + finally: + for cpg in owned: + _dispose(cpg) + diagnostics = list(result.degradations) + if execution_budget is not None: + if client is not None: + diagnostics.append({"stage": "model", "reason": "model_withheld_execution_budget"}) + reasons = ["deadline_exhausted"] if time.monotonic() >= execution_budget.deadline_monotonic else [] + for cpg in owned: + read_diagnostics = getattr(cpg, "execution_diagnostics", None) + if callable(read_diagnostics): + reasons.extend(read_diagnostics()) + for reason in dict.fromkeys(reasons): + diagnostics.append({"stage": "model", "reason": reason}) + return replace(result, degradations=tuple(diagnostics), seconds=round(time.monotonic() - started, 2)) + + +def _scan_repository_impl( + root: Path, + regions: Sequence[ScanRegion], + *, + backend: Any, + client: Any | None = None, + model: str = "", + budget: ScanBudget | None = None, + execution_budget: ExecutionBudget | None = None, + owned: list[Any], ) -> ModelScanResult: """Arbitrate a repository's regions under one CPG and one budget.""" limits = budget if budget is not None else ScanBudget() @@ -140,7 +185,7 @@ def scan_repository( for region in regions: counts[region.language] = counts.get(region.language, 0) + 1 dominant = max(counts, key=lambda name: (counts[name], name)) if counts else "" - cpg = _build(backend, root, dominant) + cpg = _build(backend, root, dominant, execution_budget=execution_budget) build_seconds = round(time.monotonic() - build_started, 2) if cpg is None: return ModelScanResult( @@ -151,6 +196,8 @@ def scan_repository( degradations=({"stage": "model", "reason": "cpg_build_failed", "detail": getattr(backend, "last_failure", "")},), ) + owned.append(cpg) + # A CPG the frontend built with files missing is not the repository the caller asked about, and no # verdict over it can say anything about the code that was dropped. There is no rung for that, because # the ladder labels verdicts and none was ever produced here -- so it is a degradation, named, with the @@ -198,14 +245,14 @@ def scan_repository( tiers: list[tuple[str, str, str, int]] = [] per_kind: dict[str, float] = {} evidence_by_pair: dict[tuple[str, str, str], Evidence] = {} - if limits.tiering and callable(getattr(cpg, "run_batch", None)): - candidates = ( + if _within_deadline(execution_budget) and limits.tiering and callable(getattr(cpg, "run_batch", None)): + evidence_candidates = ( ordered_regions[: (limits.evidence_regions or len(ordered_regions))] if limits.order_by_evidence else ordered_regions[: limits.max_regions] ) tier_started = time.monotonic() - evidence_by_pair, failed = _evidence_pass(cpg, _collect(candidates), hooks) + evidence_by_pair, failed = _evidence_pass(cpg, _collect(evidence_candidates), hooks) per_kind["evidence"] = round(time.monotonic() - tier_started, 2) if failed: degradations.append({"stage": "model", "reason": "query_failed", "kind": "evidence", "requests": failed}) @@ -227,6 +274,8 @@ def scan_repository( query_started = time.monotonic() pruned = 0 for kind, requests in _grouped(work, hook_callbacks=hooks).items(): + if not _within_deadline(execution_budget): + break if kind == "taint" and limits.prune_tier0 and tier_by_rid: # Tier 0 is exact: a family with no sink in reach cannot yield a flow. Measured on a 637-file # plugin, that is 2,161 of 2,500 requests -- 86% of a query that timed out at forty minutes. @@ -268,8 +317,15 @@ def scan_repository( rows_by_id.update(answered_batch) else: # a backend without batching still works, one call at a time for rid, params in requests.items(): + if not _within_deadline(execution_budget): + break answered = cpg.run(kind, params) - rows_by_id[rid] = list(answered) if isinstance(answered, list) else [] + if isinstance(answered, list): + rows_by_id[rid] = list(answered) + elif execution_budget is None: + rows_by_id[rid] = [] + else: + degradations.append({"stage": "model", "reason": "query_failed", "kind": kind, "requests": 1}) per_kind[kind] = round(time.monotonic() - kind_started, 2) query_seconds = round(time.monotonic() - query_started, 2) @@ -287,6 +343,12 @@ def scan_repository( judged: set[tuple[str, str | None]] = set() unasked: set[tuple[str, str | None]] = set() for rid, region, spec in work: + if not _within_deadline(execution_budget): + break + # Unanswered requests are deferred on the bounded path, never adjudicated + # from an invented empty response. Exact identities are added in task 3.2. + if execution_budget is not None and rid not in rows_by_id: + continue exhausted = counted is not None and counted.calls >= limits.max_model_calls family = getattr(spec, "family", "") prefetched = CpgResult(cpg_path=cpg.cpg_path, run=_prefetched(rows_by_id.get(rid, []))) @@ -331,7 +393,6 @@ def scan_repository( findings = _ordered(_deduplicated(collected)) # The graph has answered everything it is going to. Its scratch tree holds the CPG, the request files and # joern's own working copy -- tens of megabytes per scan -- and nothing else reclaims it. - _dispose(cpg) return ModelScanResult( findings=findings, by_rung=_tally(findings), @@ -441,11 +502,26 @@ def _dispose(cpg: object) -> None: logger.warning("could not remove the cpg scratch directory: %s", exc) -def _build(backend: Any, root: Path, language: str) -> Any: +def _within_deadline(budget: ExecutionBudget | None) -> bool: + return budget is None or time.monotonic() < budget.deadline_monotonic + + +def _build(backend: Any, root: Path, language: str, *, execution_budget: ExecutionBudget | None = None) -> Any: """Build through the backend, passing the language and the vendored trees when the backend can use them.""" from .layout import vendored_directories + if not _within_deadline(execution_budget): + return None excluded = vendored_directories(root) if root.is_dir() else () + if execution_budget is not None: + try: + inspect.signature(backend.build).bind(root, language=language, exclude=excluded, execution_budget=execution_budget) + except (TypeError, ValueError): + # An old backend cannot silently turn a bounded push into an ordinary scan. + return None + if not _within_deadline(execution_budget): + return None + return backend.build(root, language=language, exclude=excluded, execution_budget=execution_budget) try: return backend.build(root, language=language, exclude=excluded) except TypeError: # a backend from before the retry existed, including every test double diff --git a/tests/test_cpg_backend.py b/tests/test_cpg_backend.py index 78b9234..623e44b 100644 --- a/tests/test_cpg_backend.py +++ b/tests/test_cpg_backend.py @@ -75,6 +75,7 @@ def test_the_seam_imports_nothing_outside_the_standard_library() -> None: "re", "shutil", "signal", + "stat", "subprocess", "tempfile", "time", diff --git a/tests/test_cpg_budget.py b/tests/test_cpg_budget.py new file mode 100644 index 0000000..fe27253 --- /dev/null +++ b/tests/test_cpg_budget.py @@ -0,0 +1,372 @@ +"""Absolute transaction deadlines on the real process boundary (pre-push 4.1/4.2).""" + +from __future__ import annotations + +import contextlib +import os +import subprocess +import sys +import time +from pathlib import Path + +import pytest + +from openultrasast.cpg.backend import BEGIN, END, JoernBackend +from openultrasast.model.contracts import ExecutionBudget + + +def test_expired_budget_launches_nothing(tmp_path: Path) -> None: + calls = [] + + def run(command, **kwargs): + calls.append(command) + return subprocess.CompletedProcess(command, 0, "ok", "") + + backend = JoernBackend(runner=run, execution_budget=ExecutionBudget(time.monotonic() - 1, 0.2)) + assert backend._run(["unused"], timeout=300) is None + assert not calls + assert "deadline_exhausted" in backend.last_failure + + +def test_remaining_time_is_not_reset_between_commands() -> None: + remaining = [] + + def run(command, **kwargs): + remaining.append(kwargs["timeout"]) + time.sleep(0.04) + return subprocess.CompletedProcess(command, 0, "", "") + + backend = JoernBackend(runner=run, execution_budget=ExecutionBudget(time.monotonic() + 0.5, 0.2)) + assert backend._run(["first"], timeout=300) is not None + assert backend._run(["second"], timeout=300) is not None + assert 0 < remaining[1] < remaining[0] <= 0.5 + + +@pytest.mark.parametrize("leader_exits", [False, True]) +def test_hanging_process_group_killed_inside_shared_allowance(tmp_path: Path, leader_exits: bool) -> None: + source = tmp_path / "source.js" + source.write_bytes(b"const input = req.query.input;\n") + assert len(source.read_bytes()) == 31 + pids = tmp_path / "pids" + script = """import os,signal,time,sys +signal.signal(signal.SIGTERM,signal.SIG_IGN) +child=os.fork() +if child==0: + while True: time.sleep(1) +open(sys.argv[1], 'w').write(str(os.getpid())+' '+str(child)) +if sys.argv[2]=='yes': sys.exit(0) +while True: time.sleep(1) +""" + started = time.monotonic() + deadline = started + 0.25 + backend = JoernBackend(execution_budget=ExecutionBudget(deadline, 0.3)) + try: + assert backend._run([sys.executable, "-c", script, str(pids), "yes" if leader_exits else "no"], timeout=1) is None + assert time.monotonic() - started < 0.65 + assert pids.is_file() + for pid in map(int, pids.read_text().split()): + status = Path(f"/proc/{pid}/stat") + for _ in range(30): + if not status.exists() or status.read_text().split()[2] == "Z": + break + time.sleep(0.005) + else: + pytest.fail(f"process {pid} survived cancellation") + finally: + if pids.exists(): + for pid in map(int, pids.read_text().split()): + with contextlib.suppress(ProcessLookupError): + os.kill(pid, 9) + + +@pytest.mark.parametrize("hang_stage", ["probe", "frontend", "overlay", "census", "taint"]) +def test_real_hanging_engine_stages_share_deadline_and_cleanup(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, hang_stage: str) -> None: + from openultrasast.cpg import backend as module + + root = tmp_path / "source" + root.mkdir() + source = root / "api.php" + source.write_bytes(b" None: + from openultrasast.cpg import backend as module + + calls = [] + + def run(command, **kwargs): + calls.append(kwargs["timeout"]) + if "--output" in command: + Path(command[command.index("--output") + 1]).write_bytes(b"graph fixture") + return subprocess.CompletedProcess(command, 0, f'{BEGIN}\n{{"r":[]}}\n{END}', "") + + monkeypatch.setattr(module.shutil, "which", lambda name: name) + backend = JoernBackend(runner=run) + budget = ExecutionBudget(time.monotonic() + 1, 0.2) + graph = backend.build(tmp_path, language="javascript", execution_budget=budget) + assert graph is not None and graph.run_batch is not None and graph.cleanup is not None + assert backend.execution_budget is None + time.sleep(0.02) + assert graph.run_batch("taint", {"r": {}}) == {"r": []} + assert 0 < calls[1] < calls[0] <= 1 + graph.cleanup() + assert not graph.cpg_path.parent.exists() + + +def test_early_process_timeout_stops_fallback_and_retries(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + from openultrasast.cpg import backend as module + + calls = [] + + def run(command, **kwargs): + calls.append(command) + raise subprocess.TimeoutExpired(command, kwargs["timeout"]) + + monkeypatch.setattr(module.shutil, "which", lambda name: name) + backend = JoernBackend(runner=run, execution_budget=ExecutionBudget(time.monotonic() + 100, 0.2)) + assert backend.build(tmp_path, language="javascript") is None + assert len(calls) == 1 + assert "process_timeout" in backend._diagnostics + + +def test_retry_timeout_consumes_same_absolute_budget(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + from openultrasast.cpg import backend as module + + calls = [] + + def run(command, **kwargs): + calls.append(kwargs["timeout"]) + Path(command[command.index("-o") + 1]).write_bytes(b"partial graph") + time.sleep(0.04) + return subprocess.CompletedProcess(command, 0, "WARN Failed to process 'x.php'", "") + + monkeypatch.setattr(module.shutil, "which", lambda name: name) + backend = JoernBackend(runner=run, execution_budget=ExecutionBudget(time.monotonic() + 1, 0.2)) + assert backend._build_with_retries_raw(tmp_path, tmp_path / "cpg.bin", tmp_path, "php") == ("x.php",) + assert len(calls) == 4 + assert all(a > b > 0 for a, b in zip(calls, calls[1:], strict=False)) + assert calls[0] <= 1 + + +def test_shard_failure_cannot_become_complete_empty_answer(tmp_path: Path) -> None: + calls = [] + + def run(command, **kwargs): + calls.append(command) + if len(calls) == 2: + raise subprocess.TimeoutExpired(command, kwargs["timeout"]) + return subprocess.CompletedProcess(command, 0, f'{BEGIN}\n{{"r":[]}}\n{END}', "") + + backend = JoernBackend(runner=run, execution_budget=ExecutionBudget(time.monotonic() + 1, 0.2)) + assert backend.query_batch_across([tmp_path / "a.bin", tmp_path / "b.bin", tmp_path / "c.bin"], "taint", {"r": {}}) is None + assert len(calls) == 2 + assert "shard_query_incomplete" in backend._diagnostics + + +def test_cleanup_cannot_reset_expired_cancellation_allowance(tmp_path: Path) -> None: + graph = tmp_path / "owned" + graph.mkdir() + (graph / "data").write_bytes(b"data") + backend = JoernBackend(execution_budget=ExecutionBudget(time.monotonic() - 1, 0.1)) + started = time.monotonic() + backend._cleanup(graph, final=True) + assert time.monotonic() - started < 0.05 + assert graph.exists() + assert "scratch_cleanup_incomplete" in backend._diagnostics + + +def test_cleanup_does_not_follow_substituted_directory_links(tmp_path: Path) -> None: + outside = tmp_path / "outside" + outside.mkdir() + (outside / "keep").write_bytes(b"keep") + scratch = tmp_path / "scratch" + scratch.mkdir() + (scratch / "link").symlink_to(outside, target_is_directory=True) + backend = JoernBackend(execution_budget=ExecutionBudget(time.monotonic() + 1, 0.2)) + backend._cleanup(scratch, final=True) + assert not scratch.exists() + assert (outside / "keep").read_bytes() == b"keep" + + +def test_driver_propagates_budget_and_rejects_legacy_without_launch(tmp_path: Path) -> None: + from openultrasast.cpg.backend import CpgResult + from openultrasast.model.scan import scan_repository + + budget = ExecutionBudget(time.monotonic() + 1, 0.2) + + class Backend: + seen = None + cleaned = False + + def build(self, root, *, language, exclude, execution_budget): + self.seen = execution_budget + return CpgResult(tmp_path / "graph", lambda *args: [], cleanup=self.cleanup) + + def cleanup(self): + self.cleaned = True + + backend = Backend() + result = scan_repository(tmp_path, [], backend=backend, execution_budget=budget, client=object()) + assert backend.seen is budget and backend.cleaned + assert any(d["reason"] == "model_withheld_execution_budget" for d in result.degradations) + + class Legacy: + called = False + + def build(self, root): + self.called = True + raise AssertionError("must not launch without a deadline") + + legacy = Legacy() + result = scan_repository(tmp_path, [], backend=legacy, execution_budget=budget) + assert not legacy.called + assert any(d["reason"] == "cpg_build_failed" for d in result.degradations) + + +def test_budgeted_driver_does_not_retry_internal_typeerror(tmp_path: Path) -> None: + from openultrasast.model.scan import scan_repository + + class Backend: + calls = 0 + + def build(self, root, **kwargs): + self.calls += 1 + raise TypeError("internal backend bug") + + backend = Backend() + with pytest.raises(TypeError, match="internal backend bug"): + scan_repository(tmp_path, [], backend=backend, execution_budget=ExecutionBudget(time.monotonic() + 1, 0.2)) + assert backend.calls == 1 + + +def test_cancelled_graph_session_does_not_poison_next_build(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + from openultrasast.cpg import backend as module + + calls = [] + + def run(command, **kwargs): + calls.append(command) + if len(calls) == 1: + raise subprocess.TimeoutExpired(command, kwargs["timeout"]) + if "--output" in command: + Path(command[command.index("--output") + 1]).write_bytes(b"graph") + return subprocess.CompletedProcess(command, 0, f'{BEGIN}\n{{"r":[]}}\n{END}', "") + + monkeypatch.setattr(module.shutil, "which", lambda name: name) + backend = JoernBackend(runner=run) + budget = ExecutionBudget(time.monotonic() + 2, 0.2) + assert backend.build(tmp_path, execution_budget=budget) is None + assert backend._cancel_deadline is None + graph = backend.build(tmp_path, execution_budget=budget) + assert graph is not None and graph.cleanup is not None + assert graph.execution_diagnostics is not None + assert graph.execution_diagnostics() == () + graph.cleanup() + + +def test_driver_cleans_graph_on_unexpected_exception(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + from openultrasast.cpg.backend import CpgResult + from openultrasast.model import scan + + cleaned = [] + + class Backend: + def build(self, root, **kwargs): + return CpgResult(tmp_path / "graph", lambda *args: [], cleanup=lambda: cleaned.append(True)) + + def broken(*args, **kwargs): + raise RuntimeError("broken context") + + monkeypatch.setattr(scan, "_hook_callbacks", broken) + with pytest.raises(RuntimeError, match="broken context"): + scan.scan_repository(tmp_path, [], backend=Backend(), execution_budget=ExecutionBudget(time.monotonic() + 1, 0.2)) + assert cleaned == [True] + + +def test_failed_nonbatch_query_is_unjudged_on_budgeted_path(tmp_path: Path) -> None: + from openultrasast.cpg.backend import CpgResult + from openultrasast.model.regions import ScanRegion + from openultrasast.model.scan import scan_repository + + source = tmp_path / "api.py" + source.write_bytes(b"def run(x): return x\n") + assert len(source.read_bytes()) > 0 + + class Backend: + def build(self, root, **kwargs): + return CpgResult(tmp_path / "graph", lambda *args: None) + + region = ScanRegion(path="api.py", function="run", language="python", families=("injection",), rank=1.0, source="entry_point") + result = scan_repository(tmp_path, [region], backend=Backend(), execution_budget=ExecutionBudget(time.monotonic() + 1, 0.2)) + assert result.regions_scanned == 0 and result.regions_unjudged == 1 + assert any(d["reason"] == "query_failed" for d in result.degradations)