wip(g2): park change-scoped attempt + full diagnosis (NOT the fix)

Scope+bound (change-first ordering, adaptive budget-fitted portioning,
_context_pass scoped to changed files, evidence_pass_bounded degradation) is
preserved here but does NOT unblock analysis: query_seconds stayed 0 across every
live re-verify. Uncontended measurement proved why -- the evidence pass is ~150s
fixed (first-query graph-load + census) + ~93ms/pair, so ~200s fixed overhead
(with the 48s build) precedes any analysis and a 300s budget leaves ~100s.
Pair-count scoping only trims the ~86s per-pair tail. The warm-session env
(OUSAST_ENGINE_SESSION=1) times out at startup under contention and, even working,
spares only subsequent batches -- not the first evidence query's load.

Parked per decision to finish the G0 baseline first; resume on a clean host.
Branch unmerged. 95 model tests pass; no push-suite regression.

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

.kiro/specs/engine-change-scoped-analysis/brief.md

@@ -83,6 +83,45 @@ detection is being lost, not in Joern's capability.
       pass already returns, tier those in a second small pass — exact reach-into-unchanged, more code.
     - (iii) **Per-pair speedup:** attack the ~235 ms/pair (warm server / batching) rather than the pair count —
       helps every scan but is a deeper engine change and does not by itself bound a huge repo.
+
+  **CORRECTION (2026-10-11, after implementing (i) + four live re-verifies — the cost model was wrong).**
+  Scope + bound was built and tested (change-first ordering, adaptive budget-fitted portioning, `_context_pass`
+  scoped to changed files, honest `evidence_pass_bounded` degradation; branch `engine-change-scoped-evidence-pass`).
+  It correctly cut the tiered pair count — but `query_seconds` stayed **0** every time. The live numbers show why:
+
+  | pairs tiered | run_batch calls | evidence s | build s |
+  |---|---|---|---|
+  | 1015 | 1 | 238 | 52 |
+  | 512 | 1 | 219 | 68 |
+  | 64 | 2 | 220 | 68 |
+  | 4 (probe) | 1 | 160 | 127 |
+
+  Evidence time is **flat regardless of pair count** → the cost is **per-*call* fixed** (~150–220 s: loading the CPG
+  into the query engine + the census the evidenceOnly batch computes), ~85 ms/pair on top. Pair-count scoping
+  therefore cannot free budget, and portioning re-pays the fixed cost per call (so (i) can be counterproductive).
+  Build is also large and swings 52→127 s — two engines thrashing the 7 GB host (the record-07 baseline shares it).
+  The real levers are now (iii)-shaped, not (i): **warm/persistent query engine or a cached census** to kill the
+  ~150 s per-call fixed cost, and/or a **bigger engine-arm budget** than 300 s; plus a **clean (uncontended)
+  measurement**, because contention alone may be inflating build + fixed cost 2–3×. Scope+bound is NOT merged.
+
+  **UNCONTENDED measurement (2026-10-11, baseline paused via SIGSTOP + container kill, then resumed).** Shipped
+  engine, one evidence call over 1,015 pairs: `build=48 s`, `evidence=243.79 s`, `query=0`. The evidence pass is
+  **the same ~244 s uncontended as ~238 s contended** → the cost is **INTRINSIC, not contention**; contention only
+  inflated `build` (48 → 127 s) and added run-to-run noise. The model is `evidence ≈ 150 s fixed (CPG load into the
+  query engine + the census the evidenceOnly batch computes) + ~93 ms/pair` (244 = 150 + 1015×0.093). Consequences:
+  - **~200 s of fixed overhead** (`build` + evidence-load) precedes any analysis, so a 300 s budget leaves only
+    ~100 s for everything else. This is why `query=0` even on a clean host.
+  - **Change-scoping still helps, but only by cutting the per-pair tail** (~1,015×93 ms ≈ 95 s): one evidence call
+    over ~30 changed pairs is ~150 s + 3 s instead of 244 s, which would free ~90 s — possibly enough for analysis,
+    but marginal.
+  - **It must be ONE call, not portioned.** The ~150 s is a first-call load, so the adaptive probe attributed it to
+    its 4 probe pairs (per-pair≈37 s), concluded the budget fit ~0 more, and tiered only 4 — the implemented
+    scope+bound is doubly wrong for this cost model. The correct shape: change-first candidates, count-capped, ONE
+    evidence call, `deadline=None`.
+  - **Pre-push viability:** the ~200 s fixed floor means the engine arm cannot be a fast (≤30 s) pre-push check on a
+    repo this size without a warm/persistent session or a cached census. A larger budget (e.g. 600 s) makes the
+    benchmark arm run but is not a realistic pre-push deadline. The lever that matters for the product is killing
+    the ~150 s fixed cost, not scoping pairs.
 - [ ] **2. Seed analysis from the change.** Prioritize (or restrict first-pass to) the changed declarations plus
   their call/data neighborhood over repo-wide static rank, within budget.
 - [ ] **3. Soundness.** Keep cross-file dataflow INTO the changed function (don't scope to literal changed lines);

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,68 @@ 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
+# The deadline is only checked BETWEEN portions and a run_batch cannot be interrupted, so a portion's duration is
+# the overshoot. A cold large CPG answered a changed-region evidenceOnly pair in SECONDS (measured ~3-5 s under
+# load, 50-100x the warm few ms), so even a fixed 32-pair portion overshot a ~115 s sub-budget by a whole ~150 s
+# batch and left nothing for the flow queries. So the bounded pass sizes ADAPTIVELY: a small probe first, then each
+# next chunk sized from the measured per-pair rate to fit the budget left -- overshoot is one probe, at any rate.
+EVIDENCE_PROBE = 4  # the opening chunk, before the per-pair rate is known
+# The cap on any one chunk under bounding: the adaptive rate estimate assumes a uniform per-pair cost, so a cap
+# keeps a chunk from overshooting badly if the regions after the probe turn out slower than it did.
+BOUNDED_EVIDENCE_PORTION = 16
 
-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. 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).
+
+    Without a ``deadline`` the pass runs the whole population in fixed ``portion`` chunks, as before. WITH one (a
+    monotonic time), it is bounded ADAPTIVELY: a small probe chunk first, then each next chunk sized from the
+    measured per-pair rate to fit the budget that is left, capped at ``portion``. A run_batch cannot be
+    interrupted, so sizing to the remaining budget -- rather than a fixed count -- keeps the overshoot to one probe
+    whether a pair costs milliseconds or seconds. 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
+    start = 0
+    per_pair = 0.0
+    while start < len(ids):
+        now = time.monotonic()
+        if deadline is not None and start > 0 and now >= deadline:
+            break
+        if deadline is None:
+            step = portion
+        elif per_pair <= 0.0:
+            step = min(EVIDENCE_PROBE, portion)  # probe first, before the rate is known
+        else:
+            step = max(1, min(portion, int((deadline - now) / per_pair)))  # fit the budget that is left
+        chunk = ids[start : start + step]
+        batch_started = time.monotonic()
+        attempted += len(chunk)
+        part = cpg.run_batch("taint", {rid: requests[rid] for rid in chunk})
+        if deadline is not None and chunk:
+            per_pair = (time.monotonic() - batch_started) / len(chunk)
+        start += len(chunk)
         if part is None:
             continue
         answered_any = True
@@ -1269,7 +1336,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 +1348,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 +1372,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 {}
@@ -1357,8 +1427,8 @@ def _evidence_pass(
         if len(languages) > 1:
             out.pop(key, None)
     if context_rows is not None:
-        _context_pass(cpg, work, context_rows, unit=unit)
-    return out, len(requests) - len(out)
+        _context_pass(cpg, work, context_rows, unit=unit, context_files=context_files)
+    return out, max(0, attempted - len(out)), len(requests) - attempted
 
 
 def _context_pass(
@@ -1367,6 +1437,7 @@ def _context_pass(
     context_rows: dict[QuestionIdentity, list[Mapping[str, object]]],
     *,
     unit: str,
+    context_files: tuple[str, ...] | None = None,
 ) -> None:
     """Collect change context for the families the evidence pass does not ask.
 
@@ -1377,7 +1448,16 @@ def _context_pass(
 
     Each family's own query supplies its own scope. A failure here leaves the family's context
     absent, which the caller already treats as unresolved, so it degrades exactly as before.
+
+    ``context_files`` scopes the pass to the regions a change TOUCHED. This pass is change context, so on a push
+    only the changed regions need it -- and over a whole repository it is otherwise as costly as the tier pass
+    (measured ~200 s on one push, every bit of it unchanged-region context nothing would consult). A changed
+    region's own dominance/config context is what `affected_context` reads to establish what the change reaches;
+    the unchanged tail is left to the same `context_projection_unavailable` it would hit under a spent budget.
     """
+    if context_files is not None:
+        allowed = set(context_files)
+        work = [item for item in work if item[1].path in allowed]
     grouped = _grouped(list(work))
     by_rid = {rid: (region, spec) for rid, region, spec in work}
     for kind in ("dominance", "config"):

tests/test_model_change_context.py

@@ -73,6 +73,98 @@ 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(50 * scanmod.BOUNDED_EVIDENCE_PORTION)}
+    answered, attempted = scanmod._portioned_evidence(Cpg(), reqs, deadline=_t.monotonic() - 1, portion=scanmod.BOUNDED_EVIDENCE_PORTION)
+    assert seen == [scanmod.EVIDENCE_PROBE]  # a past deadline stops after only the opening probe chunk
+    assert attempted == scanmod.EVIDENCE_PROBE
+    assert answered is not None and len(answered) == scanmod.EVIDENCE_PROBE
+    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
+    assert all(s <= scanmod.BOUNDED_EVIDENCE_PORTION for s in seen)  # in capped chunks
+
+
+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",))