feat(scan): re-ask a killed portion remainder instead of dropping it

The portioner iterated a fixed pre-split list, so a killed portion left its unanswered ids behind:
on the plugin the list ran out at 1,263 s of a 3,600 s budget with those remainders lost, and
completion sat far below what the budget could buy. It is now a work queue. A portion that comes back
short pushes its unanswered ids to the front, to be re-asked at the budget the kill just halved;
`fit` splits an over-budget head, so heads shrink toward single requests, and a single request a
whole ceiling could not answer is dropped rather than looped on forever. A control asserts both the
re-ask and that termination.

This is the last piece of portioning-over-cap: streaming keeps what a kill finished, single-attempt
timing keeps the sizer honest, coverage tells a too-big portion from a slow one, and the queue keeps
working while budget and work remain.

1,425 tests pass, nineteen skipped.

Co-Authored-By: Claude Opus 4.8 <[email protected]>
Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo

src/openultrasast/model/scan.py

@@ -904,27 +904,39 @@ def _batched(
     )
     merged: dict[str, list[object]] = {}
     answered_any = False
-    for index, chunk in enumerate(chunks):
-        # Checked per CHUNK, not per kind. A first attempt checked between kinds, and one kind is a single
-        # iteration that runs every chunk inside it -- so taint spent the whole budget and the reserve only
-        # stopped the cheap kinds behind it. The granularity of a budget check has to match the granularity
-        # of the work it guards.
+    # A QUEUE, not a fixed list, because a killed portion leaves a remainder that must be re-asked at the now
+    # smaller budget rather than dropped. On the plugin the fixed list ran out at 1,263 s of a 3,600 s budget
+    # with the killed portions' remainders lost -- completion far below what the budget could buy. The queue
+    # keeps working while there is time and something left to ask; `fit` sizes the head against the current
+    # budget, and a portion that comes back short pushes its unanswered ids back for a smaller retry.
+    queue: list[dict[str, Mapping[str, object]]] = list(chunks)
+    while queue:
+        # Checked per portion, not per kind: one kind runs every portion inside a single loop iteration, so a
+        # per-kind check would let taint spend the whole budget before the reserve looked.
         if not within():
-            logger.info("cpg %s: %d batch(es) left unasked to keep time for arbitration", kind, len(chunks) - index)
+            left = sum(len(c) for c in queue)
+            logger.info("cpg %s: %d request(s) left unasked to keep time for arbitration", kind, left)
             break
-        chunk = sizer.fit(chunk, pending=chunks[index + 1 :])
+        chunk = sizer.fit(queue.pop(0), pending=queue)
         started = time.monotonic()
         answer = batch(kind, chunk)
         # WHOLE, not merely non-empty. A killed portion streams some answers and returns them, which is the
         # point -- but its time is the ceiling, not the cost of the work, so fitting a per-visit rate from it
-        # is meaningless. Only a portion that answered every id it was asked gives a clean sample; a short
-        # one halves the budget, because it was too big for the ceiling.
-        whole = answer is not None and all(rid in answer for rid in chunk)
+        # is meaningless. Only a portion that answered every id it was asked gives a clean sample.
+        answered_ids = set(answer or ())
+        whole = answer is not None and all(rid in answered_ids for rid in chunk)
         sizer.observe(chunk, time.monotonic() - started, answered=whole)
-        if answer is None:
-            continue
-        answered_any = True
-        merged.update(answer)
+        if answer is not None:
+            answered_any = True
+            merged.update(answer)
+        # The unanswered remainder goes back to the front, to be re-asked at the now-smaller budget: either
+        # progress was made and the rest is a genuine remainder, or nothing answered and `fit` will split the
+        # portion smaller next time. Termination is guaranteed because a portion that answers nothing halves
+        # the budget, `fit` caps the head at the budget, so heads shrink to single requests; and a SINGLE
+        # request a whole ceiling could not answer is dropped rather than looped on forever.
+        remainder = {rid: req for rid, req in chunk.items() if rid not in answered_ids}
+        if remainder and len(chunk) > 1:
+            queue.insert(0, remainder)
     return merged if answered_any else None
 
 

tests/test_cpg_backend.py

@@ -1311,3 +1311,25 @@ def test_the_portioner_asks_single_attempts_and_halves_on_a_short_answer(tmp_pat
     assert once_calls, "the single-attempt path was not used"
     # The portion that came back short of its ids was reported as not-whole (False) to the sizer.
     assert False in seen, seen
+
+
+def test_the_queue_re_asks_a_killed_remainder_and_terminates() -> None:
+    """A killed portion's unanswered ids are re-asked at a smaller budget; a lone hard request is dropped."""
+    from openultrasast.model import scan
+
+    attempts: list[list[str]] = []
+
+    def batch(kind: str, requests: dict[str, dict[str, object]]) -> dict[str, list[object]] | None:
+        attempts.append(sorted(requests))
+        # "hard" never answers, whatever size it is asked in; everything else answers.
+        answered = {rid: [] for rid in requests if rid != "hard"}
+        return answered or None
+
+    weights = {"a": 500, "b": 500, "hard": 500, "c": 500}
+    requests = {rid: {} for rid in weights}
+    merged = scan._batched(batch, "taint", requests, weights=weights)
+    assert merged is not None
+    assert {"a", "b", "c"} <= set(merged) and "hard" not in merged, merged
+    # It must stop: "hard" is asked, dropped once it is alone, and never re-queued forever.
+    assert len(attempts) < 40, len(attempts)
+    assert attempts.count(["hard"]) <= 1, attempts