diff --git a/.kiro/specs/contributor-scan/tasks.md b/.kiro/specs/contributor-scan/tasks.md index 4a86f23..132505d 100644 --- a/.kiro/specs/contributor-scan/tasks.md +++ b/.kiro/specs/contributor-scan/tasks.md @@ -1275,3 +1275,13 @@ the ontology describes libraries, frameworks and language constructs, never repo - _Boundary: contributor-scan scan budget, cpg backend_ - _Depends: 7.3_ - _Requirements: 4.1, 6.3, 8.1_ + +- [ ] 7.8 Portion the work instead of capping it: a kill keeps what finished + - **Asked for by the maintainer on 2026-09-22, after 7.7.** A hard per-invocation ceiling is all-or-nothing: a batch killed at 301 s loses every answer it had already computed, and the retry then pays a fresh JVM start per part to ask them all again. Measured on the plugin: 116 requests killed at 300 s, eight restarts, and the same everywhere-sinks in every part. + - Three coordinated changes replace the cap as the thing that governs cost. The taint query STREAMS: the census first, then one line per request as it is answered, flushed as it goes, inside the unchanged fence. The driver harvests a killed process's stdout and assembles every complete line, marking the result partial; the retry then asks only the remainder. And portion sizing ADAPTS: after each portion the observed seconds per sink visit, less the fixed start, resizes the rest toward a target duration, and a portion that did not answer halves the budget rather than being fitted. The ceiling stays as a safety net that the sizing is meant to make rare. + - Legacy single-document payloads must keep parsing unchanged; the other queries do not stream and need not. A census line is one whose only key is the census — a first cut misread a legacy document that carried the census beside its answers as a census line and dropped every answer it sat beside, which a control caught. + - Done when a killed taint batch hands back the answers it finished and the retry asks only the rest; portion sizes follow observed cost; the streamed and legacy forms both parse; and the plugin population completes more questions per unit of budget than under 7.7 with no finding lost. + - _Status:_ implemented with host controls passing for the assembler, the remainder retry and the sizer; the joern-gated streaming check and the plugin measurement wait for the transfer replay to free the machine. + - _Boundary: cpg backend, contributor-scan scan budget, taint query_ + - _Depends: 7.7_ + - _Requirements: 4.1, 6.3, 8.1_ diff --git a/src/openultrasast/cpg/backend.py b/src/openultrasast/cpg/backend.py index c8082e3..29a9b2b 100644 --- a/src/openultrasast/cpg/backend.py +++ b/src/openultrasast/cpg/backend.py @@ -186,13 +186,36 @@ _SHARED_REQUEST_FIELDS = ("hookCallbacks", "dispatchApply") Runner = Callable[..., subprocess.CompletedProcess[str]] +# A streamed answer: one JSON document per line inside the fence, each carrying the request id it answers, +# printed and flushed as the engine finishes it. The point is what a KILL keeps. A batch that used to build +# its whole answer and print it once lost every answer when it was killed at the ceiling -- 116 requests at +# 300 s, then eight fresh JVM starts to ask them again. Streamed, the same kill keeps everything that had +# finished, and only the unfinished requests are asked again. +STREAM_ID = "id" +STREAM_ROWS = "rows" +PARTIAL = "__partial__" + + def extract_payload(stdout: str) -> object | None: - """The JSON between the fence markers, or ``None`` when it is absent or malformed.""" + """The JSON between the fence markers, or ``None`` when it is absent or malformed. + + Two shapes are accepted. The committed one is a single document. The streamed one is a document per + line, each `{"id": ..., "rows": [...]}` or `{"__census__": [...]}`, which this assembles into the same + mapping the callers already read. A streamed body whose closing fence never arrived -- the process was + killed -- still yields every complete line, marked ``__partial__`` so a caller can tell a harvest from a + whole answer. A single-document body without its closing fence is still nothing, as before: there is + no partial truth in half a document. + """ start = stdout.find(BEGIN) - end = stdout.find(END, start + 1) if start >= 0 else -1 - if start < 0 or end < 0: + if start < 0: + return None + end = stdout.find(END, start + 1) + body = stdout[start + len(BEGIN) : end if end >= 0 else len(stdout)].strip() + streamed = _streamed(body, complete=end >= 0) + if streamed is not None: + return streamed + if end < 0: return None - body = stdout[start + len(BEGIN) : end].strip() # Joern logs to STDOUT, and it does so while the script is running -- so a log line can land INSIDE the # fence, ahead of the payload. `[INFO ] Attempting to determine flows from empty list of sources.` did # exactly that and cost a 205-request batch every one of its answers, reported as `query_failed`. The @@ -204,6 +227,40 @@ def extract_payload(stdout: str) -> object | None: return payload +def _streamed(body: str, *, complete: bool) -> dict[str, object] | None: + """Assemble a streamed body, or ``None`` when the body is not streamed at all. + + A line is streamed when it is a JSON object carrying an id or the census key. Anything else on a line + is what Joern logged and is skipped; a trailing line that does not parse is the one the kill cut + through, and it is dropped rather than guessed at. + """ + answers: dict[str, object] = {} + saw_stream = False + for line in body.splitlines(): + text = line.strip() + if not text.startswith("{"): + continue + try: + document = json.loads(text) + except ValueError: + continue + if not isinstance(document, dict): + continue + # A census LINE carries nothing but the census. A legacy single document carries the census beside + # every answer, and reading that as a census line dropped every answer it sat beside. + if set(document) == {"__census__"}: + answers["__census__"] = document["__census__"] + saw_stream = True + elif STREAM_ID in document and STREAM_ROWS in document: + answers[str(document[STREAM_ID])] = document[STREAM_ROWS] + saw_stream = True + if not saw_stream: + return None + if not complete: + answers[PARTIAL] = True + return answers + + def _decoded_json(body: str) -> object | None: """The first complete JSON document in ``body``, ignoring whatever Joern logged around it. @@ -1067,18 +1124,30 @@ class JoernBackend: time.monotonic() - started, "answered" if answered is not None else "unanswered", ) - if answered is not None or len(requests) <= 1: + # Only what is still unanswered is asked again. A streamed batch that was killed hands back every + # answer it finished, so the retry is the REMAINDER, not the batch; and a batch that answered whole + # has no remainder and returns here as it always did. + merged: dict[str, list[object]] = {rid: rows for rid, rows in (answered or {}).items()} + remaining = {rid: req for rid, req in requests.items() if rid not in merged} + if not remaining or len(requests) <= 1: return answered - size = max(1, len(requests) // _BATCH_SPLITS) - ids = list(requests) - merged: dict[str, list[object]] = {} - recovered = False + if merged: + logger.info( + "cpg batch %s: %d of %d answered before the kill; asking the other %d in parts", + query, + len([k for k in merged if k != "__census__"]), + len(requests), + len(remaining), + ) + size = max(1, len(remaining) // _BATCH_SPLITS) + ids = list(remaining) + recovered = bool(answered) for start in range(0, len(ids), size): if not self._time_available(): logger.info("cpg batch %s: split abandoned at part %d, no time left", query, start // size + 1) break part_started = time.monotonic() - part = self._batch_once(cpg_path, query, {rid: requests[rid] for rid in ids[start : start + size]}) + part = self._batch_once(cpg_path, query, {rid: remaining[rid] for rid in ids[start : start + size]}) logger.info( "cpg batch %s part %d: %d request(s) in %.1fs, %s", query, @@ -1151,7 +1220,8 @@ class JoernBackend: for name, value in shared.items(): command += ["--param", f"{name}={value}"] completed = self._run(command, timeout=self.query_timeout, cwd=cpg_path.parent) - if completed is None or completed.returncode != 0: + harvested = completed is not None and completed.returncode != 0 and BEGIN in (completed.stdout or "") + if completed is None or (completed.returncode != 0 and not harvested): detail = _engine_failure(completed) logger.warning("cpg batch %s failed: %s", query, detail) return None @@ -1165,8 +1235,11 @@ class JoernBackend: # clean repository -- 500 regions examined, nothing found, no degradation recorded. logger.warning("cpg batch %s returned no parseable payload", query) return None + if parsed.get(PARTIAL): + answered = sum(1 for rid, rows in parsed.items() if rid != PARTIAL and isinstance(rows, list) and rid != "__census__") + logger.info("cpg batch %s was killed with %d of %d answer(s) already streamed", query, answered, len(requests)) # A request the engine did not answer is absent, not empty: the caller must be able to tell them apart. - return {str(rid): list(rows) for rid, rows in parsed.items() if isinstance(rows, list)} + return {str(rid): list(rows) for rid, rows in parsed.items() if rid != PARTIAL and isinstance(rows, list)} 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.""" @@ -1298,6 +1371,7 @@ class JoernBackend: self._note("deadline_exhausted") return None process = None + terminated = False 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) @@ -1317,7 +1391,10 @@ class JoernBackend: if not self._time_available(): return None return result - except subprocess.TimeoutExpired: + except subprocess.TimeoutExpired as expired: + # What the process had already written is kept, because a streamed answer is worth exactly the + # lines that finished. `communicate` raises without returning them, so they are drained after + # the kill -- see `_partial_stdout`, which reads what the pipe still holds. # Two different timeouts end here and they must not have the same consequence. If the SCAN's # deadline is gone, cancellation is right: stop everything promptly, which is what the push's # two-second gate measures. If a single query merely exceeded its own ceiling, the scan may have @@ -1330,12 +1407,25 @@ class JoernBackend: if exhausted: self._cancel_deadline = min(self._cleanup_limit(), time.monotonic() + self.execution_budget.cancellation_allowance_seconds) self._note("deadline_exhausted" if exhausted else "process_timeout") + # The group is killed BEFORE its output is read. A first cut read the pipe first, and a pipe whose + # writer is still alive never reaches end-of-file: the suite's own hanging-engine control sat on + # that read for twenty-eight minutes. Once the group is dead, `communicate` returns promptly with + # everything the pipe still held. + if process is not None: + if not _terminate_bounded_group(process, self._cleanup_limit()): + self._note("process_cleanup_incomplete") + terminated = True + partial = _partial_stdout(expired, process) + if partial: + # A killed process whose output holds finished answers is not nothing. The negative return + # code says it was killed; the stdout says what it finished; `_batch_once` reads both. + return subprocess.CompletedProcess(command, -1, partial, "") return None except (OSError, subprocess.SubprocessError): self._note("process_failed") return None finally: - if process is not None: + if process is not None and not terminated: # 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. @@ -1416,6 +1506,26 @@ class _DeadlineExpired(Exception): """Internal cooperative boundary; public builds/queries retain failure semantics.""" +def _partial_stdout(expired: subprocess.TimeoutExpired, process: subprocess.Popen[str] | None) -> str: + """Whatever the killed process had written, from the exception and then from the pipe itself. + + `communicate(timeout=)` raises with the output it had buffered so far, but a pipe can hold more than the + buffer had drained by the time the timer fired. After the group is killed the pipe has nothing left to + wait on, so a final read returns the rest without blocking. + """ + output = expired.output if isinstance(expired.output, str) else (expired.output.decode(errors="replace") if expired.output else "") + if process is None: + return output + # Only valid once the group is dead; the caller guarantees that. `communicate` on a finished process + # returns whatever the pipes still held and does not wait on anything. + try: + rest, _ = process.communicate(timeout=1.0) + except (subprocess.TimeoutExpired, OSError, ValueError): + return output + rest_text = rest if isinstance(rest, str) else (rest.decode(errors="replace") if rest else "") + return rest_text if rest_text else output + + 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. diff --git a/src/openultrasast/cpg/queries/taint.sc b/src/openultrasast/cpg/queries/taint.sc index 2fc43dc..382a88a 100644 --- a/src/openultrasast/cpg/queries/taint.sc +++ b/src/openultrasast/cpg/queries/taint.sc @@ -874,15 +874,23 @@ ) } - println("---OUSAST-CPG-BEGIN---") if (requestsJson.nonEmpty) { val parsed = ujson.read(requestsJson).obj - val answers = parsed.map { case (id, req) => + // STREAMED: the census first, then one line per request as it is answered, flushed as it goes. What + // this buys is what a kill keeps. Built as one object and printed at the end, a batch killed at its + // ceiling lost every answer it had finished -- 116 requests at 300 s, then eight JVM starts to ask them + // all again. Printed as it goes, the same kill keeps everything that finished, the driver harvests it, + // and only the unfinished requests are asked again. The fence is unchanged; the driver reads both forms. + val census = ujson.Arr(ujson.Obj("methods" -> cpg.method.size.toString, "files" -> cpg.file.size.toString, "file_names" -> ujson.Arr.from(cpg.file.name.l))) + println("---OUSAST-CPG-BEGIN---") + println(ujson.write(ujson.Obj("__census__" -> census))) + System.out.flush() + parsed.foreach { case (id, req) => def field(name: String): String = req.obj.get(name).map(_.str).getOrElse("") val paramSrc = req.obj.get("parameterSources").map(_.str).getOrElse("false") val fieldSrc = req.obj.get("fieldSources").map(_.str).getOrElse("true") val evidence = req.obj.get("evidenceOnly").map(_.str).getOrElse("false") - id -> ujson.Arr( + val rows = ujson.Arr( rowsFor( field("sources"), field("sinks"), @@ -902,15 +910,15 @@ field("contextEvidence") ): _* ) + println(ujson.write(ujson.Obj("id" -> id, "rows" -> rows))) + System.out.flush() } - // A census of the graph, under a key no request id can collide with (ids are numbers). A frontend can - // fail every file and STILL exit 0 with a valid, empty CPG -- `joern-parse` does not even propagate the - // per-file warnings -- so "no rows" and "no graph" are indistinguishable to the driver without this. - // It rides the batch rather than costing its own invocation, because JVM startup is what this whole - // batching design exists to avoid. - val census = ujson.Arr(ujson.Obj("methods" -> cpg.method.size.toString, "files" -> cpg.file.size.toString, "file_names" -> ujson.Arr.from(cpg.file.name.l))) - println(ujson.write(ujson.Obj.from(answers.toSeq :+ ("__census__" -> census)))) + // The census rides the stream as its FIRST line rather than costing its own invocation, because JVM + // startup is what this whole batching design exists to avoid -- and first because a frontend that fails + // every file still exits 0 with an empty graph, and the driver must see that even when the process is + // killed before any request answers. } else { + println("---OUSAST-CPG-BEGIN---") println( ujson.write(ujson.Arr(rowsFor(sources, sinks, sanitizers, hookCallbacks, dispatchApply, function, parameterSources, fieldSources, evidenceOnly, file, callDepth, boundedSinks): _*)) ) diff --git a/src/openultrasast/model/scan.py b/src/openultrasast/model/scan.py index 13e98a7..0a94ba6 100644 --- a/src/openultrasast/model/scan.py +++ b/src/openultrasast/model/scan.py @@ -885,6 +885,11 @@ def _batched( if len(chunks) <= 1: answered: dict[str, list[object]] | None = batch(kind, requests) return answered + # The budget adapts to what the engine actually does. The constant it starts from was read off one + # subject, and a graph where sinks are cheaper or dearer would either waste portions or blow them. After + # each portion the observed cost per sink visit, less the fixed start, resizes the rest toward a target + # duration -- so a kill becomes something the sizing avoids rather than something the ceiling handles. + sizer = _PortionSizer(weights) logger.info( "cpg %s split into %d batches by sink weight (%d request(s), %d sink visit(s))", kind, @@ -902,7 +907,10 @@ def _batched( if not within(): logger.info("cpg %s: %d batch(es) left unasked to keep time for arbitration", kind, len(chunks) - index) break + chunk = sizer.fit(chunk, pending=chunks[index + 1 :]) + started = time.monotonic() answer = batch(kind, chunk) + sizer.observe(chunk, time.monotonic() - started, answered=bool(answer)) if answer is None: continue answered_any = True @@ -910,6 +918,59 @@ def _batched( return merged if answered_any else None +# What one portion should take: long enough that the fixed JVM start is a minority of it, short enough that +# a mis-estimate is bounded and a kill costs little. +PORTION_TARGET_SECONDS = 120.0 +PORTION_FIXED_SECONDS = 70.0 # measured start-and-load on a 4 MB graph + + +class _PortionSizer: + """Resize the remaining portions from the cost the engine has shown so far. + + Sizing starts from `MAX_SINK_VISITS_PER_BATCH`, and the observed seconds per sink visit replaces that + guess as soon as one portion has answered. A portion that did not answer is evidence too, of the + harshest kind: the budget is halved rather than fitted, because the true cost is unknown and above the + ceiling. `pending` is edited in place, so a portion cut down to fit puts its remainder back at the front + of the queue rather than losing it. + """ + + def __init__(self, weights: Mapping[str, int]) -> None: + self.weights = weights + self.budget = float(MAX_SINK_VISITS_PER_BATCH) + + def weight(self, chunk: Mapping[str, object]) -> int: + return sum(self.weights.get(rid, 1) for rid in chunk) + + def fit( + self, chunk: dict[str, Mapping[str, object]], *, pending: list[dict[str, Mapping[str, object]]] + ) -> dict[str, Mapping[str, object]]: + """Shrink this portion to the budget, or grow it from the ones behind it while it still fits.""" + if self.weight(chunk) > self.budget and len(chunk) > 1: + kept: dict[str, Mapping[str, object]] = {} + carried = 0 + for rid, req in chunk.items(): + if kept and carried + self.weights.get(rid, 1) > self.budget: + break + kept[rid] = req + carried += self.weights.get(rid, 1) + pending.insert(0, {rid: req for rid, req in chunk.items() if rid not in kept}) + return kept + while pending and self.weight(chunk) + self.weight(pending[0]) <= self.budget: + chunk = {**chunk, **pending.pop(0)} + return chunk + + def observe(self, chunk: Mapping[str, object], seconds: float, *, answered: bool) -> None: + visits = max(1, self.weight(chunk)) + if not answered: + self.budget = max(1.0, self.budget / 2) + logger.info("cpg portion of %d sink visit(s) did not answer in %.0fs; budget halved to %d", visits, seconds, int(self.budget)) + return + per_visit = max(0.0, seconds - PORTION_FIXED_SECONDS) / visits + if per_visit > 0: + self.budget = max(1.0, (PORTION_TARGET_SECONDS - PORTION_FIXED_SECONDS) / per_visit) + logger.info("cpg portion of %d sink visit(s) answered in %.0fs; next budget %d visit(s)", visits, seconds, int(self.budget)) + + def _collect( regions: Sequence[ScanRegion], *, execution_budget: ExecutionBudget | None = None ) -> list[tuple[str, ScanRegion, ArbiterSpec]]: diff --git a/tests/test_cpg_backend.py b/tests/test_cpg_backend.py index 386ca32..8892e30 100644 --- a/tests/test_cpg_backend.py +++ b/tests/test_cpg_backend.py @@ -1230,3 +1230,42 @@ def test_an_engine_failure_reports_the_line_that_names_the_cause(tmp_path, monke assert _engine_failure(None) == "timeout" assert "no output" in _engine_failure(subprocess.CompletedProcess(["joern"], 3, "", "")) assert isinstance(JoernBackend(), JoernBackend) + + +def test_a_streamed_payload_assembles_and_a_killed_one_keeps_what_finished() -> None: + """A kill keeps every answer that had streamed; a half-written line is dropped, not guessed at.""" + from openultrasast.cpg.backend import BEGIN, END, PARTIAL, extract_payload + + whole = f'{BEGIN}\n{{"__census__": [{{"methods": "5"}}]}}\n{{"id": "0", "rows": [1]}}\n{{"id": "1", "rows": []}}\n{END}\n' + assembled = extract_payload(whole) + assert assembled == {"__census__": [{"methods": "5"}], "0": [1], "1": []} + + killed = f'{BEGIN}\n[INFO ] engine chatter\n{{"__census__": []}}\n{{"id": "0", "rows": [1]}}\n{{"id": "1", "ro' + harvest = extract_payload(killed) + assert harvest is not None and harvest.get(PARTIAL) is True + assert harvest["0"] == [1] and "1" not in harvest + + legacy = f'{BEGIN}\n{{"0": [1], "__census__": []}}\n{END}\n' + assert extract_payload(legacy) == {"0": [1], "__census__": []} + # Half of a single document is still nothing. + assert extract_payload(f'{BEGIN}\n{{"0": [1], "__cen') is None + + +def test_a_killed_batch_retries_only_the_remainder(tmp_path: Path) -> None: + """The retry asks what was NOT answered, so a kill costs the unfinished requests and nothing else.""" + from openultrasast.cpg.backend import JoernBackend + + backend = JoernBackend(runner=lambda *a, **k: None) + calls: list[list[str]] = [] + + def once(cpg_path: Path, query: str, requests: dict[str, dict[str, object]]) -> dict[str, list[object]] | None: + calls.append(sorted(requests)) + if len(calls) == 1: + return {"__census__": [], "a": [1]} # killed mid-way: a answered, b and c did not + return {rid: [rid] for rid in requests} + + backend._batch_once = once # type: ignore[method-assign] + answered = backend.query_batch(tmp_path / "cpg.bin", "taint", {"a": {}, "b": {}, "c": {}}) + assert answered is not None and answered["a"] == [1] and answered["b"] == ["b"] and answered["c"] == ["c"] + assert calls[0] == ["a", "b", "c"] + assert all("a" not in retry for retry in calls[1:]), calls diff --git a/tests/test_cpg_queries_compile.py b/tests/test_cpg_queries_compile.py index 6133805..9db5370 100644 --- a/tests/test_cpg_queries_compile.py +++ b/tests/test_cpg_queries_compile.py @@ -575,3 +575,49 @@ def test_prototype_pollution_is_established_on_the_shape_it_declares(tmp_path: P flows = [row for row in (payload.get("0") or []) if row.get("sink")] assert flows, "a request body merged into an object was not reported" assert all("merge" in str(row["sink"]) for row in flows), f"the Express setting was reported: {flows}" + + +@pytest.mark.skipif(_joern() is None, reason="joern is not installed on this machine") +def test_the_taint_query_streams_one_line_per_request(tmp_path: Path) -> None: + """Answers arrive as they finish, so a kill keeps what finished. The fence is unchanged.""" + from openultrasast.cpg.backend import BEGIN, END, extract_payload + + frontend = __import__("shutil").which("jssrc2cpg") + if frontend is None: + pytest.skip("jssrc2cpg is not installed") + (tmp_path / "a.js").write_text("function handler(req, res) { return eval(req.body.x); }\n") + cpg = tmp_path / "cpg.bin" + subprocess.run([frontend, str(tmp_path), "-o", str(cpg)], capture_output=True, text=True, timeout=600, check=False) + if not cpg.is_file(): + pytest.skip("could not build a sample cpg") + requests = tmp_path / "requests.json" + requests.write_text( + json.dumps( + { + "0": {"sources": "req.body", "sinks": "eval", "function": "handler"}, + "1": {"sources": "req.body", "sinks": "eval", "function": "nothing"}, + } + ) + ) + done = subprocess.run( + [ + _joern() or "joern", + "--script", + str((QUERIES / "taint.sc").resolve()), + "--param", + f"cpgFile={cpg}", + "--param", + f"requestsFile={requests}", + ], + capture_output=True, + text=True, + timeout=900, + check=False, + ) + assert done.returncode == 0, (done.stderr or done.stdout or "")[-500:] + body = done.stdout[done.stdout.find(BEGIN) + len(BEGIN) : done.stdout.find(END)] + lines = [line for line in body.splitlines() if line.strip().startswith("{")] + assert len(lines) >= 3, lines # census, then one per request + assert json.loads(lines[0]).keys() == {"__census__"} + assembled = extract_payload(done.stdout) + assert assembled is not None and set(assembled) >= {"0", "1", "__census__"} and "__partial__" not in assembled diff --git a/tests/test_model_scope.py b/tests/test_model_scope.py index af9a39c..bb3d48a 100644 --- a/tests/test_model_scope.py +++ b/tests/test_model_scope.py @@ -320,3 +320,21 @@ def test_expired_planning_keeps_census_without_more_fact_reads(tmp_path, monkeyp assert time.monotonic() < deadline + 0.5 assert len(result.scope.selected) + len(result.scope.deferred) == 200 assert "deadline_exhausted" in result.scope.unresolved_boundaries + + +def test_a_portion_sizer_shrinks_to_budget_and_learns_from_what_answered() -> None: + """Portions follow the engine's observed cost; a portion that did not answer halves the budget.""" + from openultrasast.model import scan + + weights = {"w": 500, "x": 400, "y": 100, "z": 100} + sizer = scan._PortionSizer(weights) + sizer.budget = 600.0 + pending = [{"y": {}}, {"z": {}}] + fitted = sizer.fit({"w": {}, "x": {}}, pending=pending) + assert sorted(fitted) == ["w"] and pending[0] == {"x": {}}, (fitted, pending) # x pushed to the front, not lost + grown = sizer.fit({"y": {}}, pending=[{"z": {}}]) + assert sorted(grown) == ["y", "z"] + sizer.observe({"w": {}}, 300.0, answered=False) + assert sizer.budget == 300.0 + sizer.observe({"y": {}, "z": {}}, scan.PORTION_FIXED_SECONDS + 20.0, answered=True) # 0.1 s per visit + assert 450 < sizer.budget < 550, sizer.budget