feat(model): phase 2 of flow-aware-ranking -- spend the region budget by evidence, weights published

`Evidence.score` orders pairs within a tier by one table of weights (open sinks, cleansed sinks, local /
carried / near sources, entry, declared public), capped so volume stops counting, and zero for tier 0 so
the weights can rank but never manufacture. `ScanBudget.order_by_evidence` runs the evidence pass over
every region and cuts `max_regions` from each region's best (tier, score); `order_by_evidence()` is the
one ordering function, used by the scan and by `benchmarks/ranking/position.py --evidence`. Off by
default until the position benchmark says where the pinned CVEs land under it.

Co-Authored-By: Claude Fable 5.1 <[email protected]>
Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG

.kiro/specs/flow-aware-ranking/brief.md

@@ -125,6 +125,18 @@ A phase that cannot show that is not an improvement, it is a story.
   makes a scan finish, because the job is concentration: the CVE is already inside 500, the question is
   whether it is inside 50. **If this passes, the model is a refinement, not the plan** — and knowing that is
   worth more than assuming it.
+  **Implemented 2026-09-11, measurement in flight.** `Evidence.score` is the published weighted sum, and
+  every weight is in one table in `model/evidence.py`: per open sink +1.0 (capped at 5), per cleansed or
+  bound sink −0.5 (capped at 3), local source +2.0, carried +1.5, near source +1.0, entry +0.5, declared
+  public +1.0. A tier-0 pair scores 0 whatever its sources say, so the weights order but cannot
+  manufacture. `ScanBudget.order_by_evidence` runs the evidence pass over every region (or
+  `evidence_regions` of them), takes each region's best pair's (tier, score), and cuts `max_regions` from
+  that order; `order_by_evidence()` is one function used by the scan and by
+  `benchmarks/ranking/position.py --evidence`, so the benchmark measures the order the scan spends. Off
+  by default until the benchmark says where the pinned CVEs land. A region with no taint family sorts
+  after every region with evidence -- the scan does not know it is safe, it knows nothing -- which is a
+  known cost of this phase for families like `access_control` and is noted here rather than hidden.
+
 - **Phase 3 — the LLM ranker.** Same evidence, ordered by a model. Gate: beats phase 2 leave-one-repository-out.
   If it does not, phase 2 ships and phase 3 does not.
 - **Phase 4 — break the closed loop.** An exploration slice (~10% of budget sampled outside the top-K) and

benchmarks/ranking/position.py

@@ -38,8 +38,43 @@ def ordered_regions(root: Path) -> list[ScanRegion]:
     return sorted(regions, key=lambda r: (not r.shipped, -r.rank, r.path, r.function or ""))
 
 
-def measure(name: str, root: Path, known: list[dict[str, object]]) -> dict[str, object]:
+def evidence_ordered_regions(root: Path, regions: list[ScanRegion]) -> tuple[list[ScanRegion], dict[str, object]]:
+    """The regions in the order phase 2 spends the budget: by each region's best (tier, score), from a real
+    evidence pass over EVERY region. Needs Joern. The ordering function is the scan's own, imported."""
+    import time
+
+    from openultrasast.cpg.backend import resolve_cpg_backend
+    from openultrasast.model.scan import _collect, _evidence_pass, _hook_callbacks, order_by_evidence
+
+    backend = resolve_cpg_backend()
+    started = time.monotonic()
+    cpg = backend.build(root, language=regions[0].language if regions else "")
+    build_seconds = round(time.monotonic() - started, 1)
+    if cpg is None:
+        raise SystemExit(f"no graph for {root}: {getattr(backend, 'last_failure', '')}")
+    try:
+        started = time.monotonic()
+        evidence, failed = _evidence_pass(cpg, _collect(regions), _hook_callbacks(root, regions))
+        evidence_seconds = round(time.monotonic() - started, 1)
+    finally:
+        if callable(getattr(cpg, "cleanup", None)):
+            cpg.cleanup()
+    tiers = Counter(e.tier for e in evidence.values())
+    facts = {
+        "build_seconds": build_seconds,
+        "evidence_seconds": evidence_seconds,
+        "pairs": len(evidence),
+        "pairs_unanswered": failed,
+        "tier_counts": {str(k): v for k, v in sorted(tiers.items())},
+    }
+    return order_by_evidence(regions, evidence), facts
+
+
+def measure(name: str, root: Path, known: list[dict[str, object]], *, by_evidence: bool = False) -> dict[str, object]:
     regions = ordered_regions(root)
+    facts: dict[str, object] = {}
+    if by_evidence:
+        regions, facts = evidence_ordered_regions(root, regions)
     tiers = Counter(r.rank for r in regions)
     largest = max(tiers.values()) if tiers else 0
     rows: list[dict[str, object]] = []
@@ -76,6 +111,7 @@ def measure(name: str, root: Path, known: list[dict[str, object]]) -> dict[str,
         # The smallest examined-region budget that still holds every in-scope known vulnerability.
         "budget_at_recall": (max(positioned) + 1) if positioned else None,
         "known": rows,
+        **facts,
     }
 
 
@@ -83,6 +119,7 @@ def main() -> int:
     parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
     parser.add_argument("--json", action="store_true", help="emit one JSON object per repository")
     parser.add_argument("--repo", help="only this recipe")
+    parser.add_argument("--evidence", action="store_true", help="phase 2: order by a real evidence pass over every region (needs Joern)")
     args = parser.parse_args()
 
     results: list[dict[str, object]] = []
@@ -99,7 +136,7 @@ def main() -> int:
             {"id": k.id, "family": k.family, "file": k.file, "function": k.function, "line": k.line, "in_scope": k.in_scope}
             for k in recipe.known
         ]
-        results.append(measure(recipe.name, root, known))
+        results.append(measure(recipe.name, root, known, by_evidence=args.evidence))
 
     if args.json:
         print(json.dumps(results, indent=2))
@@ -114,6 +151,8 @@ def main() -> int:
             scope = "" if k["in_scope"] else "  [out of scope]"
             print(f"   {k['id']:16} {where:28} {k['site']}{scope}")
         print(f"   budget_at_recall = {r['budget_at_recall']}")
+        if "tier_counts" in r:
+            print(f"   build {r['build_seconds']}s, evidence {r['evidence_seconds']}s over {r['pairs']} pairs, tiers {r['tier_counts']}")
     return 0
 
 

src/openultrasast/model/evidence.py

@@ -21,6 +21,20 @@ TIER_CLEANSED = 2  # a source exists, but every sink call is bound or cleansed o
 TIER_OPEN = 3  # a source exists and some sink call is computed and uncleansed: ask first
 TIER_OPEN_PUBLIC = 4  # tier 3 and declared public, or a source on the same statement: ask very first
 
+# The within-tier weights, published here because they are the whole of what "heuristic ranking" means in
+# phase 2 and a reader should be able to see every one of them at once. They order, they never exclude.
+WEIGHTS: dict[str, float] = {
+    "open_sink": 1.0,  # per sink call that is computed, uncleansed and unbound, up to OPEN_SINK_CAP
+    "cleansed_sink": -0.5,  # per sink call that is cleansed on the call or bound, up to CLEANSED_SINK_CAP
+    "source_local": 2.0,  # a modelled source in the region's own body
+    "carried": 1.5,  # stage one of the two-stage join says a field or dispatch value arrives tainted
+    "source_near": 1.0,  # a modelled source within the arbiter's callDepth
+    "entry": 0.5,  # parameters untrusted by contract
+    "access_declared_public": 1.0,
+}
+OPEN_SINK_CAP = 5
+CLEANSED_SINK_CAP = 3
+
 
 @dataclass(frozen=True)
 class SinkEvidence:
@@ -77,6 +91,33 @@ class Evidence:
             return TIER_OPEN_PUBLIC
         return TIER_OPEN
 
+    @property
+    def score(self) -> float:
+        """Order WITHIN a tier, by the published weights below. Never across tiers: the tier is the predicate.
+
+        A weighted sum of the same exact facts the tier is made of, so it can rank but cannot manufacture --
+        a pair with no sink scores nothing and stays tier 0 whatever its weights. Counts are capped because
+        a file-scope region with 346 sinks is not 346 times more likely to hold the bug than one with 5;
+        the cap is where the evidence stops saying more. Phase 2 of flow-aware-ranking.
+        """
+        if self.tier == TIER_EXCLUDE:
+            return 0.0  # no sink in reach: nothing to weigh, whatever the sources say
+        w = WEIGHTS
+        return (
+            w["open_sink"] * min(len(self.open_sinks), OPEN_SINK_CAP)
+            + w["cleansed_sink"] * min(sum(1 for s in self.sinks if s.cleansed_on_call or self.is_bound(s)), CLEANSED_SINK_CAP)
+            + (w["source_local"] if self.source_local else 0.0)
+            + (w["carried"] if self.carried else 0.0)
+            + (w["source_near"] if self.source_near else 0.0)
+            + (w["entry"] if self.entry else 0.0)
+            + (w["access_declared_public"] if self.access_declared_public else 0.0)
+        )
+
+    @property
+    def order_key(self) -> tuple[int, float]:
+        """Higher sorts first: ``sorted(..., key=lambda e: e.order_key, reverse=True)``."""
+        return (self.tier, self.score)
+
 
 def evidence_from_rows(
     rows: Sequence[Mapping[str, object]] | None,
@@ -139,11 +180,14 @@ def _as_int(value: object) -> int:
 
 
 __all__ = [
+    "CLEANSED_SINK_CAP",
+    "OPEN_SINK_CAP",
     "TIER_CLEANSED",
     "TIER_EXCLUDE",
     "TIER_NO_SOURCE",
     "TIER_OPEN",
     "TIER_OPEN_PUBLIC",
+    "WEIGHTS",
     "Evidence",
     "SinkEvidence",
     "evidence_from_rows",

src/openultrasast/model/scan.py

@@ -26,10 +26,10 @@ from pathlib import Path
 from typing import Any
 
 from ..cpg.backend import CpgResult
-from .evidence import TIER_EXCLUDE, evidence_from_rows
 from .candidates import enumerate_candidates
 from .config_value import request_params as config_params
 from .dominance import request_params as dominance_params
+from .evidence import TIER_EXCLUDE, Evidence, evidence_from_rows
 from .ladder import Rung
 from .pipeline import ModelFinding, scan_region
 from .regions import MODULE_SCOPE, ScanRegion
@@ -75,6 +75,12 @@ class ScanBudget:
     # language, under any dataflow model. Skipping it cannot lose a finding. It is the only tier that
     # excludes, which is why it is the only one allowed to act before the ranker is measured.
     prune_tier0: bool = True
+    # Phase 2 of flow-aware-ranking: spend the region budget by EVIDENCE, not by static rank. The evidence
+    # pass runs over `evidence_regions` regions in static order (0 = every region; it costs milliseconds a
+    # pair), each region takes its best pair's (tier, score), and `max_regions` is then cut from that
+    # order. Off by default until the position benchmark says where the pinned CVEs land under it.
+    order_by_evidence: bool = False
+    evidence_regions: int = 0
 
 
 @dataclass(frozen=True)
@@ -185,50 +191,32 @@ def scan_repository(
             }
         )
 
-    # Phase 1: collect the whole scan's questions, for at most `max_regions` regions. Nothing is asked yet.
-    work: list[tuple[str, ScanRegion, ArbiterSpec]] = []
-    for region in ordered_regions[: limits.max_regions]:
-        for family in region.families:
-            spec = _spec_for(family, region.language)
-            if spec is None:
-                continue
-            work.append((f"{len(work)}", region, spec))
-
     # Phase 1b: the evidence tier of every taint pair, before any dataflow is asked. Same query, `evidenceOnly`,
-    # no `reachableByFlows`. Recorded on the result and NOT used to reorder or skip anything -- that is a
-    # later phase, and it must be measured against this record before it is allowed to change behaviour.
+    # no `reachableByFlows`. Phase 0 recorded it; phase 1 prunes tier 0 on it; phase 2 spends the region
+    # budget by it. Under phase 2 the pass runs over MORE regions than the budget, which is the point.
     tiers: list[tuple[str, str, str, int]] = []
-    tier_by_rid: dict[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)):
-        grouped_taint = _grouped(work, hook_callbacks=hooks).get("taint", {})
-        if grouped_taint:
-            evidence_requests = {rid: {**dict(params), "evidenceOnly": "true"} for rid, params in grouped_taint.items()}
-            tier_started = time.monotonic()
-            answered_evidence = cpg.run_batch("taint", evidence_requests)
-            per_kind["evidence"] = round(time.monotonic() - tier_started, 2)
-            if answered_evidence is None:
-                degradations.append({"stage": "model", "reason": "query_failed", "kind": "evidence", "requests": len(evidence_requests)})
-            else:
-                answered_evidence.pop("__census__", None)
-                by_rid = {rid: (region, spec) for rid, region, spec in work}
-                for rid, rows in answered_evidence.items():
-                    pair = by_rid.get(rid)
-                    if pair is None:
-                        continue
-                    region, spec = pair
-                    evidence = evidence_from_rows(
-                        rows,
-                        entry=_is_entry_point(region),
-                        # Rank 1.0 is the public tier. Provenance -- declared by a route contract versus
-                        # inferred from an absent decorator -- is not yet carried on ScanRegion, so this is
-                        # the proxy phase 0 has; the brief names it as the field to thread through next.
-                        access_declared_public=region.rank >= 1.0,
-                        bound_names=getattr(spec, "safe_shape_sinks", ()),
-                    )
-                    if evidence is not None:
-                        tiers.append((region.path, region.function or "", getattr(spec, "family", ""), evidence.tier))
-                        tier_by_rid[rid] = evidence.tier
+        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)
+        per_kind["evidence"] = round(time.monotonic() - tier_started, 2)
+        if failed:
+            degradations.append({"stage": "model", "reason": "query_failed", "kind": "evidence", "requests": failed})
+        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)
+
+    # Phase 1: collect the whole scan's questions, for at most `max_regions` regions. Nothing is asked yet.
+    work = _collect(ordered_regions[: limits.max_regions])
+    tier_by_rid = {
+        rid: evidence_by_pair[_pair_key(region, spec)].tier for rid, region, spec in work if _pair_key(region, spec) in evidence_by_pair
+    }
 
     # Phase 2: ONE invocation per query kind. JVM startup dominates a repository scan -- a call per region
     # per family put a ten-line file at four minutes and a thousand regions at roughly fifty hours -- while
@@ -362,6 +350,85 @@ def scan_repository(
     )
 
 
+def _pair_key(region: ScanRegion, spec: ArbiterSpec) -> tuple[str, str, str]:
+    return (region.path, region.function or "", getattr(spec, "family", ""))
+
+
+def _collect(regions: Sequence[ScanRegion]) -> list[tuple[str, ScanRegion, ArbiterSpec]]:
+    """Every (region, family) question with an arbiter, in the order given, with batch-local ids."""
+    work: list[tuple[str, ScanRegion, ArbiterSpec]] = []
+    for region in regions:
+        for family in region.families:
+            spec = _spec_for(family, region.language)
+            if spec is None:
+                continue
+            work.append((f"{len(work)}", region, spec))
+    return work
+
+
+def _evidence_pass(
+    cpg: Any, work: Sequence[tuple[str, ScanRegion, ArbiterSpec]], hooks: str
+) -> tuple[dict[tuple[str, str, str], Evidence], 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".
+    """
+    grouped_taint = _grouped(list(work), hook_callbacks=hooks).get("taint", {})
+    if not grouped_taint:
+        return {}, 0
+    requests = {rid: {**dict(params), "evidenceOnly": "true"} for rid, params in grouped_taint.items()}
+    answered = cpg.run_batch("taint", requests)
+    if answered is None:
+        return {}, len(requests)
+    answered.pop("__census__", None)
+    by_rid = {rid: (region, spec) for rid, region, spec in work}
+    out: dict[tuple[str, str, str], Evidence] = {}
+    for rid, rows in answered.items():
+        pair = by_rid.get(rid)
+        if pair is None:
+            continue
+        region, spec = pair
+        evidence = evidence_from_rows(
+            rows,
+            entry=_is_entry_point(region),
+            # Rank 1.0 is the public tier. Provenance -- declared by a route contract versus inferred from
+            # an absent decorator -- is not yet carried on ScanRegion, so this is the proxy phase 0 has;
+            # the brief names it as the field to thread through next.
+            access_declared_public=region.rank >= 1.0,
+            bound_names=getattr(spec, "safe_shape_sinks", ()),
+        )
+        if evidence is not None:
+            out[_pair_key(region, spec)] = evidence
+    return out, 0
+
+
+def order_by_evidence(regions: Sequence[ScanRegion], evidence: Mapping[tuple[str, str, str], Evidence]) -> list[ScanRegion]:
+    """Regions by their best pair's (tier, score), then by the static order they arrived in.
+
+    ONE definition, used by the scan to spend its budget and by `benchmarks/ranking/position.py` to say
+    where a pinned CVE lands under it; if they drifted, the benchmark would measure the wrong thing. A
+    region with no evidence at all -- no taint family, or a pass that failed -- sorts after every region
+    that has some, in static order: the scan does not know it is safe, it knows nothing about it.
+    """
+    best: dict[tuple[str, str], tuple[int, float]] = {}
+    for (path, function, _family), vector in evidence.items():
+        key = (path, function)
+        best[key] = max(best.get(key, (-1, 0.0)), vector.order_key)
+    indexed = list(enumerate(regions))
+    return [
+        region
+        for _, region in sorted(
+            indexed,
+            key=lambda item: (
+                -best.get((item[1].path, item[1].function or ""), (-1, 0.0))[0],
+                -best.get((item[1].path, item[1].function or ""), (-1, 0.0))[1],
+                item[0],
+            ),
+        )
+    ]
+
+
 def _dispose(cpg: object) -> None:
     """Release a built CPG's scratch tree. Never raises: cleanup must not turn a good scan into a failure."""
     cleanup = getattr(cpg, "cleanup", None)

tests/test_model_evidence.py

@@ -97,3 +97,29 @@ def test_a_stringified_boolean_is_read_as_its_value_not_its_truthiness() -> None
     rows[1]["cleansedOnCall"] = "False"
     e = evidence_from_rows(rows, entry=True)
     assert e is not None and e.tier == TIER_OPEN, "a 'False' string must not turn an open sink into a cleansed one"
+
+
+def test_the_score_orders_within_a_tier_and_never_across_one() -> None:
+    """Phase 2: the published weights rank pairs inside a tier. They cannot lift a pair out of tier 0 --
+    a pair with no sink scores nothing whatever the weights say -- and counts are capped so a 346-sink
+    file-scope region does not outrank everything by volume alone."""
+    from openultrasast.model.evidence import OPEN_SINK_CAP, WEIGHTS, Evidence, SinkEvidence
+
+    def sink(n: int, cleansed: bool = False) -> SinkEvidence:
+        return SinkEvidence(code=f"q({n})", line=n, method="m", arity=1, arg0_literal=False, cleansed_on_call=cleansed)
+
+    nothing = Evidence(sinks=(), source_local=True)
+    assert nothing.tier == 0 and nothing.score == 0.0, "no sink, no score: the weights cannot manufacture a pair"
+
+    one = Evidence(sinks=(sink(1),), source_near=True)
+    local = Evidence(sinks=(sink(1),), source_local=True)
+    assert local.tier > one.tier, "a local source is the higher tier, not merely the higher score"
+    assert local.score == WEIGHTS["open_sink"] + WEIGHTS["source_local"]
+
+    many = Evidence(sinks=tuple(sink(i) for i in range(50)), source_near=True)
+    assert many.tier == one.tier
+    assert many.score == WEIGHTS["open_sink"] * OPEN_SINK_CAP + WEIGHTS["source_near"], "capped: volume stops counting"
+
+    cleansed = Evidence(sinks=(sink(1), sink(2, cleansed=True)), source_near=True)
+    assert cleansed.score < one.score, "a cleansed sink beside an open one counts against the pair"
+    assert sorted([many, one, local], key=lambda e: e.order_key, reverse=True) == [local, many, one]

tests/test_model_scan.py

@@ -472,3 +472,67 @@ def test_tier_zero_pairs_are_not_asked_and_nothing_else_is_skipped() -> None:
     assert taint["count"] == 1, "only the pair with a sink in reach is asked"
     assert result.requests_pruned == 2, "and the two tier-0 pairs are counted, not silently dropped"
     assert not any(d.get("reason") == "query_failed" for d in result.degradations)
+
+
+def test_the_budget_is_spent_by_evidence_when_asked() -> None:
+    """Phase 2 of flow-aware-ranking. With `order_by_evidence`, the evidence pass runs over every region and
+    the region cap is cut from the (tier, score) order, so a low-ranked region with an open sink and a
+    local source is examined before a top-ranked region with no sink in reach. Off, the static order holds."""
+    from openultrasast.model.regions import ScanRegion
+    from openultrasast.model.scan import ScanBudget, scan_repository
+
+    class _Backend:
+        def __init__(self) -> None:
+            self.taint_files: list[list[str]] = []
+            self.evidence_sizes: list[int] = []
+
+        def available(self) -> bool:
+            return True
+
+        def build(self, root, *, language=""):  # type: ignore[no-untyped-def]
+            from openultrasast.cpg.backend import CpgResult
+
+            def batch(kind, requests):  # type: ignore[no-untyped-def]
+                if any(p.get("evidenceOnly") == "true" for p in requests.values()):
+                    self.evidence_sizes.append(len(requests))
+                    out = {}
+                    for rid, params in requests.items():
+                        hot = str(params.get("file", "")).endswith("low.py")
+                        out[rid] = [{"kind": "summary", "sinks": 1 if hot else 0, "sourceLocal": hot, "familyInRepo": True}] + (
+                            [
+                                {
+                                    "kind": "sink",
+                                    "sink": "os.system(x)",
+                                    "sinkLine": "4",
+                                    "sinkMethod": "run",
+                                    "sinkArity": 1,
+                                    "sinkArg0Literal": False,
+                                    "cleansedOnCall": False,
+                                }
+                            ]
+                            if hot
+                            else []
+                        )
+                    return out
+                if kind == "taint":
+                    self.taint_files.append(sorted({str(p.get("file", "")) for p in requests.values()}))
+                return {rid: [] for rid in requests}
+
+            return CpgResult(cpg_path=Path("cpg.bin"), run=lambda q, p: [], run_batch=batch)
+
+    top = ScanRegion(path="top.py", function="handle", language="python", families=("injection",), rank=1.0, source="entry_point")
+    low = ScanRegion(path="low.py", function="run", language="python", families=("injection",), rank=0.3, source="entry_point")
+
+    static = _Backend()
+    scan_repository(Path("."), [top, low], backend=static, budget=ScanBudget(max_model_calls=0, max_regions=1))
+    assert static.evidence_sizes == [1] and static.taint_files == [], (
+        "static order: the budget holds `top`, whose pair is tier 0 and pruned"
+    )
+
+    ordered = _Backend()
+    result = scan_repository(
+        Path("."), [top, low], backend=ordered, budget=ScanBudget(max_model_calls=0, max_regions=1, order_by_evidence=True)
+    )
+    assert ordered.evidence_sizes == [2], "the evidence pass covers every region, not only the budget"
+    assert ordered.taint_files == [["low.py"]], "and the budget is spent on the region the evidence points at"
+    assert result.tier_counts == {0: 1, 4: 1}