From 75589514930ca1755b23322bf6967b8e5d4bdc20 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sun, 11 Oct 2026 12:28:35 +0200 Subject: [PATCH] feat(g2): change-scoped, budget-bounded evidence pass (unblock push analysis) Measured: on an unseen push the evidence-tier pass is the whole cost -- 238s of a 291s bedrock run, 99.9% of the non-build/non-query time -- because it tiers every region of the repo to rank them, and a cold large CPG answers each pair ~50-100x slower than the design's assumed few ms. The budget is gone before one flow query runs: selected=0, deferred=1365 all deadline_exhausted, query_seconds=0. When the push carries a change context AND an execution budget, the evidence pass is now spent CHANGE-FIRST and BOUNDED: - evidence candidates are ordered so the changed regions tier before the rest; - the pass stops once it has taken EVIDENCE_BUDGET_FRACTION (0.5) of the remaining budget, in BOUNDED_EVIDENCE_PORTION (256) chunks so the stop is granular; - the CPG stays whole-repo, so a flow INTO a changed region from unchanged code is still traced -- only the question population's long unchanged tail may go untiered. Honesty: _portioned_evidence returns requests ATTEMPTED, so _evidence_pass splits a genuine query failure (engine asked, no answer) from a budget skip; the skipped tail is recorded as evidence_pass_bounded, never read as a clean repository. Preserves the tested invariants: with no execution budget (the unit tests) the path is inert, so change context still does not alter the ranker's selection and a changed dependency still reaches its unchanged ranked question. Tests: _portioned_evidence bounds at the deadline keeping the first portion; the scan sets the sub-deadline + bounded portion + change-first order; the bounded tail is recorded and is not a query failure. 128 model/change-context tests pass; no push-suite regression (Joern-gated engine_trace/cpg_budget failures pre-exist on HEAD). Co-Authored-By: Claude Opus 4.8 --- src/openultrasast/model/scan.py | 79 ++++++++++++++++++++------ tests/test_model_change_context.py | 91 ++++++++++++++++++++++++++++++ 2 files changed, 153 insertions(+), 17 deletions(-) diff --git a/src/openultrasast/model/scan.py b/src/openultrasast/model/scan.py index 668f0e9..370ccac 100644 --- a/src/openultrasast/model/scan.py +++ b/src/openultrasast/model/scan.py @@ -341,19 +341,38 @@ def _scan_repository_impl( if limits.order_by_evidence else ordered_regions[: limits.max_regions] ) + changed_scope = _changed_files(change_context) if change_context is not None else None + evidence_deadline: float | None = None + evidence_portion = EVIDENCE_PORTION + if changed_scope is not None and execution_budget is not None: + # Change-scoped evidence pass (see EVIDENCE_BUDGET_FRACTION): tier the changed regions first and stop + # once the pass has taken its share of the remaining budget, so the flow queries get the rest. A flow + # into a changed region from unchanged code is still traced -- the CPG is whole-repo and every changed + # region is tiered; only the unchanged tail, tiered while the sub-budget lasts, may be cut (recorded). + changed_set = set(changed_scope) + evidence_candidates = sorted(evidence_candidates, key=lambda region: region.path not in changed_set) + now = time.monotonic() + evidence_deadline = now + EVIDENCE_BUDGET_FRACTION * max(0.0, execution_budget.deadline_monotonic - now) + evidence_portion = BOUNDED_EVIDENCE_PORTION tier_started = time.monotonic() - evidence_by_pair, failed = _evidence_pass( + evidence_by_pair, failed, bounded = _evidence_pass( cpg, _collect(evidence_candidates, execution_budget=execution_budget), hooks, degradations=degradations, context_rows=context_rows if change_context is not None else None, - context_files=_changed_files(change_context) if change_context is not None else None, + context_files=changed_scope, unit=unit, + deadline=evidence_deadline, + portion=evidence_portion, ) per_kind["evidence"] = round(time.monotonic() - tier_started, 2) if failed: degradations.append({"stage": "model", "reason": "query_failed", "kind": "evidence", "requests": failed}) + if bounded: + # The budget, not a failure, left these pairs untiered. Named so a change-scoped scan's silence over + # the unchanged tail is legible rather than read as a clean bill of health for the whole repository. + degradations.append({"stage": "model", "reason": "evidence_pass_bounded", "tiered_pairs": len(evidence_by_pair), "skipped_requests": bounded}) tiers = [(path, function, family, evidence.tier) for (path, function, family), evidence in evidence_by_pair.items()] if limits.order_by_evidence and evidence_by_pair: ordered_regions = order_by_evidence(ordered_regions, evidence_by_pair) @@ -1248,20 +1267,43 @@ def _collect( EVIDENCE_PORTION = 4000 # requests per evidence invocation; roughly ten seconds of work behind one JVM start +# A push scan's whole cost is the evidence-tier pass -- measured 238 s of a 291 s run, 99.9 % of everything that +# is not the CPG build or the flow queries -- because it tiers every region of the repository to rank them, and +# a cold large CPG answers each pair far slower than the design's few milliseconds. When the push carries a change +# context the pass is therefore spent CHANGE-FIRST (the regions the push touched are tiered before the rest) and +# BOUNDED: it stops once it has taken this fraction of the remaining budget, leaving the rest for the flow queries +# that actually find the defect. The CPG is still whole-repo, so a flow INTO a changed region from unchanged code is +# traced as before; only the long tail of unchanged regions goes untiered, and that is recorded. A smaller portion +# makes the stop granular -- 4000 is one atomic batch that could never stop before the deadline. +EVIDENCE_BUDGET_FRACTION = 0.5 +BOUNDED_EVIDENCE_PORTION = 256 -def _portioned_evidence(cpg: Any, requests: Mapping[str, Mapping[str, object]]) -> dict[str, Any] | None: - """The evidence pass in portions, merged. ``None`` only when no portion answered. - Typed as the batch returns it -- rows are the engine's mappings, and the caller reads them as such. +def _portioned_evidence( + cpg: Any, requests: Mapping[str, Mapping[str, object]], *, deadline: float | None = None, portion: int = EVIDENCE_PORTION +) -> tuple[dict[str, Any] | None, int]: + """The evidence pass in portions, merged, with the number of requests ATTEMPTED. + + ``None`` only when no attempted portion answered. ``deadline`` (a monotonic time) bounds the pass: the loop + checks it before each portion after the first and stops, so the requests in the portions it reached are + tiered and the tail is left untiered. The attempted count lets the caller tell a bounded skip (budget ran + out) from a query failure (the engine was asked and did not answer). Typed as the batch returns it. """ ids = list(requests) - if len(ids) <= EVIDENCE_PORTION: + if deadline is None and len(ids) <= portion: answered: dict[str, Any] | None = cpg.run_batch("taint", requests) - return answered + # Attempted counts what was SENT, not what came back: a None here is the engine failing on requests it + # was asked, which is a failure to record -- never a budget skip (that is only the deadline-bounded tail). + return answered, len(ids) merged: dict[str, Any] = {} answered_any = False - for start in range(0, len(ids), EVIDENCE_PORTION): - part = cpg.run_batch("taint", {rid: requests[rid] for rid in ids[start : start + EVIDENCE_PORTION]}) + attempted = 0 + for start in range(0, len(ids), portion): + if deadline is not None and start > 0 and time.monotonic() >= deadline: + break + chunk = ids[start : start + portion] + attempted += len(chunk) + part = cpg.run_batch("taint", {rid: requests[rid] for rid in chunk}) if part is None: continue answered_any = True @@ -1269,7 +1311,7 @@ def _portioned_evidence(cpg: Any, requests: Mapping[str, Mapping[str, object]]) if census is not None and "__census__" not in merged: merged["__census__"] = census merged.update(part) - return merged if answered_any else None + return (merged if answered_any else None), attempted def _evidence_pass( @@ -1281,15 +1323,18 @@ def _evidence_pass( context_rows: dict[QuestionIdentity, list[Mapping[str, object]]] | None = None, context_files: tuple[str, ...] | None = None, unit: str = "repository", -) -> tuple[dict[tuple[str, str, str], Evidence], int]: + deadline: float | None = None, + portion: int = EVIDENCE_PORTION, +) -> tuple[dict[tuple[str, str, str], Evidence], int, int]: """The evidence vector of every taint pair in ``work``, keyed by (path, function, family). - Returns the vectors and the number of requests the engine did not answer (0 when it answered), so a - failed pass is recorded rather than read as "no evidence anywhere". + Returns the vectors, the number of ATTEMPTED requests the engine did not answer (a genuine failure), and + the number left untiered because ``deadline`` bounded the pass (a budget skip, not a failure) -- so each is + recorded as what it is rather than both read as "no evidence anywhere". """ grouped_taint = _grouped(list(work), hook_callbacks=hooks).get("taint", {}) if not grouped_taint: - return {}, 0 + return {}, 0, 0 # With change context, the engine itemises locations only in the files the change touched and summarises # the rest, because those are the only locations `affected_context` can attach. `None` means the change's # files cannot be named safely, and then every location is itemised as before. @@ -1302,9 +1347,9 @@ def _evidence_pass( # batch: a 22,125-request evidence pass on a plugin's push transaction exceeded its ceiling under load and # cost the whole scan. Streaming keeps what finished when a portion is killed; portioning keeps a single # kill from being the whole pass. - answered = _portioned_evidence(cpg, requests) + answered, attempted = _portioned_evidence(cpg, requests, deadline=deadline, portion=portion) if answered is None: - return {}, len(requests) + return {}, attempted, len(requests) - attempted census = answered.pop("__census__", None) if census and degradations is not None: entry = census[0] if isinstance(census[0], Mapping) else {} @@ -1358,7 +1403,7 @@ def _evidence_pass( out.pop(key, None) if context_rows is not None: _context_pass(cpg, work, context_rows, unit=unit) - return out, len(requests) - len(out) + return out, max(0, attempted - len(out)), len(requests) - attempted def _context_pass( diff --git a/tests/test_model_change_context.py b/tests/test_model_change_context.py index 02ce365..3a12388 100644 --- a/tests/test_model_change_context.py +++ b/tests/test_model_change_context.py @@ -73,6 +73,97 @@ def test_change_does_not_replace_ranker_selection(tmp_path): assert [q.score for q in baseline.scope.selected] == [q.score for q in changed.scope.selected] +# -- change-scoped evidence pass (the G0/G2 bottleneck fix) ------------------------------------------------- +# The evidence-tier pass is the whole cost of a push scan (measured 238s of 291s), so when the push carries a +# change context it is spent change-first and bounded to a fraction of the budget, leaving time for the flow +# queries. These pin the mechanism (bounds at the deadline, keeps the first portion), the wiring (scan sets a +# sub-deadline, the bounded portion, and orders the changed regions first), and the honesty (a bounded tail is +# recorded, never read as a clean repository). + + +def test_portioned_evidence_stops_at_deadline_keeping_the_first_portion(): + import time as _t + + from openultrasast.model import scan as scanmod + + seen: list[int] = [] + + class Cpg: + def run_batch(self, kind, reqs): + seen.append(len(reqs)) + return {rid: [] for rid in reqs} + + reqs = {str(i): {"file": "f"} for i in range(2 * scanmod.BOUNDED_EVIDENCE_PORTION + 5)} + answered, attempted = scanmod._portioned_evidence(Cpg(), reqs, deadline=_t.monotonic() - 1, portion=scanmod.BOUNDED_EVIDENCE_PORTION) + assert seen == [scanmod.BOUNDED_EVIDENCE_PORTION] # only the first portion ran before the (past) deadline + assert attempted == scanmod.BOUNDED_EVIDENCE_PORTION + assert answered is not None and len(answered) == scanmod.BOUNDED_EVIDENCE_PORTION + seen.clear() + _, attempted_full = scanmod._portioned_evidence(Cpg(), reqs, deadline=None, portion=scanmod.BOUNDED_EVIDENCE_PORTION) + assert attempted_full == len(reqs) and sum(seen) == len(reqs) # no deadline -> the whole population tiers + + +def test_change_scoped_scan_bounds_and_orders_the_evidence_pass_change_first(tmp_path, monkeypatch): + import time as _t + + from openultrasast.model import scan as scanmod + from openultrasast.model.contracts import ExecutionBudget + + changed, other = "z_changed.py", "a_other.py" # path sort puts a_other first; change-first must flip it + (tmp_path / changed).write_text("sink(input)\n") + (tmp_path / other).write_text("sink(input)\n") + captured: dict = {} + + def fake(cpg, requests, *, deadline=None, portion=scanmod.EVIDENCE_PORTION): + captured["deadline"] = deadline + captured["portion"] = portion + captured["files"] = [p.get("file") for p in requests.values()] + return {rid: vector() for rid in requests}, len(requests) + + monkeypatch.setattr(scanmod, "_portioned_evidence", fake) + regions = [region(other, "python"), region(changed, "python")] + backend = Backend({changed: vector(), other: vector()}) + budget = ScanBudget(order_by_evidence=True) + execution_budget = ExecutionBudget(_t.monotonic() + 100, 0.5) + scan_repository(tmp_path, regions, backend=backend, budget=budget, change_context=context(changed), execution_budget=execution_budget) + assert captured["deadline"] is not None # a sub-deadline bounds the pass + assert captured["portion"] == scanmod.BOUNDED_EVIDENCE_PORTION + assert captured["files"][0] == changed # the changed region is tiered FIRST despite sorting last by path + + # No execution budget -> no change-scoping: the whole population tiers in one unbounded portion, as before. + captured.clear() + scan_repository(tmp_path, regions, backend=backend, budget=budget, change_context=context(changed)) + assert captured["deadline"] is None and captured["portion"] == scanmod.EVIDENCE_PORTION + + +def test_change_scoped_bounded_tail_is_recorded_not_read_as_clean(tmp_path, monkeypatch): + import time as _t + + from openultrasast.model import scan as scanmod + from openultrasast.model.contracts import ExecutionBudget + + (tmp_path / "changed.py").write_text("sink(input)\n") + (tmp_path / "other.py").write_text("sink(input)\n") + + def fake(cpg, requests, *, deadline=None, portion=scanmod.EVIDENCE_PORTION): + first = list(requests)[0] # simulate a bound: one pair tiered, the rest cut by the deadline + return {first: vector()}, 1 + + monkeypatch.setattr(scanmod, "_portioned_evidence", fake) + regions = [region("changed.py", "python"), region("other.py", "python")] + result = scan_repository( + tmp_path, + regions, + backend=Backend({"changed.py": vector(), "other.py": vector()}), + budget=ScanBudget(order_by_evidence=True), + change_context=context("changed.py"), + execution_budget=ExecutionBudget(_t.monotonic() + 100, 0.5), + ) + bounded = [d for d in result.degradations if d.get("reason") == "evidence_pass_bounded"] + assert bounded and bounded[0]["skipped_requests"] >= 1 # the untiered tail is named + assert not any(d.get("reason") == "query_failed" and d.get("kind") == "evidence" for d in result.degradations) # a bound is not a failure + + def test_global_boundary_survives_answer(tmp_path): (tmp_path / "sink.php").write_text("readable\n") ctx = replace(context(), unresolved_boundaries=("external summary missing",))