From 0b1ccf84dc19c519af5a0ef7028fd2ab17f40c33 Mon Sep 17 00:00:00 2001 From: Aleksei Menshutin Date: Sat, 17 Oct 2026 12:10:47 +0300 Subject: [PATCH] Verify complete class inputs after campaign budget expiry --- usvm-ts-fast-check/scripts/pbt_experiment.py | 106 +++++++- .../scripts/test_pbt_experiment.py | 245 +++++++++++++++++- 2 files changed, 339 insertions(+), 12 deletions(-) diff --git a/usvm-ts-fast-check/scripts/pbt_experiment.py b/usvm-ts-fast-check/scripts/pbt_experiment.py index 4586dbd766..2f2132f8ad 100644 --- a/usvm-ts-fast-check/scripts/pbt_experiment.py +++ b/usvm-ts-fast-check/scripts/pbt_experiment.py @@ -211,6 +211,104 @@ def replay_inputs(artifact): return batches, omissions +def verify_campaign_identity(request, artifact): + expected_property = dict(request["property"], assertions=request["property"].get("assertions", [])) + for key in ("predicate", "precondition"): + entry = expected_property.get(key) + if entry is not None: + expected_property[key] = dict(entry, executionKind=entry.get("executionKind", "sync")) + expected_configuration = {"examples": [], "exactReplay": False, **request["configuration"]} + expected_configuration["coverageRequest"] = { + "scopes": ["source_under_test"], "includePatterns": [], "excludePatterns": [], + "includeUnexecutedSources": False, **request["configuration"]["coverageRequest"], + } + observation = expected_configuration.get("observationRequest") + if observation is not None: + expected_configuration["observationRequest"] = {"maxPointsPerInvocation": 8, **observation} + same_request = artifact["property"] == expected_property and artifact["configuration"] == expected_configuration \ + and artifact["metadata"]["sourceRoot"] == request["sourceRoot"] and artifact["result"]["mode"] == request["mode"] \ + and artifact["metadata"]["requestedNodeCommand"] == request["nodeExecutable"] + same_controls = all(artifact.get(key) == request.get(key) for key in ( + "validationReserveMillis", "pbtAllocationMillis", "completedInputLimit", "shrinkWitness", + )) + if not same_request or not same_controls: + raise ValueError("Campaign artifact does not match the requested property, source or controls") + + result = artifact["result"] + property_id = request["property"]["id"] + if result["propertyId"] != property_id: + raise ValueError("Campaign result belongs to another property") + capture_request = request["configuration"].get("inputExportRequest") or observation + expected_sources = {source["module"]: source["sha256"] for source in capture_request["sources"]} + for part in (result.get("concrete"), result.get("symbolic")): + if part is not None and part["propertyId"] != property_id: + raise ValueError("Campaign phase belongs to another property") + concrete = result.get("concrete") + if concrete is not None: + capture = concrete.get("inputExport") or concrete.get("observations") or {} + captured_sources = {source["module"]: source["sha256"] for source in capture.get("verifiedSources", [])} + if concrete["seed"] != request["configuration"]["seed"] or captured_sources != expected_sources: + raise ValueError("Captured PBT inputs do not match the requested seed and sources") + for replay in result.get("symbolicInputReplays", []): + candidate = replay["candidate"] + sources = {source["module"]: source["sha256"] for source in candidate["sources"]} + if candidate["propertyId"] != property_id or sources != expected_sources: + raise ValueError("Native symbolic input belongs to another property or source snapshot") + + +def expected_budget_stop(result): + concrete = result.get("concrete") + failure_kind = ((concrete or {}).get("failure") or {}).get("kind") + concrete_timeout = concrete is not None and concrete.get("status") == "failure" and failure_kind == "timeout" + concrete_success = concrete is not None and concrete.get("status") == "success" and failure_kind is None + mode = result["mode"] + symbolic = result.get("symbolic") + if result.get("confirmation") is not None: + return False + if mode == "PBT_ONLY": + return concrete_timeout and symbolic is None and result["stop"] == "DEADLINE" and result.get("stoppedAt") == "PBT" + if mode == "SYMBOLIC_ONLY": + if concrete is not None: + return False + elif mode not in ("SEQUENTIAL", "COVERAGE_FEEDBACK") or not (concrete_timeout or concrete_success): + return False + if symbolic is None or symbolic.get("diagnostics") or symbolic.get("target") is not None: + return False + if symbolic["status"] == "TIMEOUT": + return result["stop"] == "DEADLINE" and result.get("stoppedAt") in ("SYMBOLIC", "CONFIRMATION") + return concrete_timeout and result["stop"] == "COMPLETED" and symbolic["status"] == "NO_VIOLATION_REACHED" + + +def budget_prefix_complete(artifact, omissions): + if any(omissions.values()): + return False + if not expected_budget_stop(artifact["result"]): + return False + result = artifact["result"] + concrete = result.get("concrete") or {} + capture = concrete.get("inputExport") or concrete.get("observations") or {} + if capture.get("pendingInvocations", 0) or capture.get("unsupportedInputs", 0): + return False + symbolic = result.get("symbolic") or {} + input_key = lambda inputs: json.dumps(inputs, sort_keys=True) + expected_outcomes = {} + for item in symbolic.get("completedInputs", []): + expected_outcomes.setdefault(input_key(item["inputs"]), set()).add(item["expectedOutcome"]) + replays = result.get("symbolicInputReplays", []) + replayed = {input_key(item["candidate"]["inputs"]) for item in replays} + if set(expected_outcomes) != replayed: + return False + for replay in replays: + predictions = set(replay.get("predictedOutcomes", [])) + input_predictions = expected_outcomes[input_key(replay["candidate"]["inputs"])] + verified = predictions == input_predictions == {"HOLDS"} and replay["completedWithinDeadline"] is True \ + and replay["replay"]["admission"] == "ADMITTED" and replay["replay"]["outcome"] == "HOLDS" \ + and not replay["replay"].get("diagnostic") + if not verified: + return False + return True + + def run_cell(frozen, case, seed, mode, output): directory = output / case["id"] / str(seed) / mode directory.mkdir(parents=True) @@ -284,15 +382,17 @@ def run_cell(frozen, case, seed, mode, output): stopped_on_pbt_fault = result.get("symbolic") is None and failure.get("kind") == "property" summary["analysisSupported"] = (mode == "PBT_ONLY" or symbolic_supported or stopped_on_pbt_fault) \ and not summary["symbolicDiagnostics"] + batches, omissions = replay_inputs(artifact) + summary["omissions"] = omissions + verify_campaign_identity(request, artifact) runner_completed = runner_process["exitCode"] == 0 \ - or (runner_process["exitCode"] == 1 and summary["confirmedViolation"]) + or (runner_process["exitCode"] == 1 and summary["confirmedViolation"]) \ + or (runner_process["exitCode"] == 2 and budget_prefix_complete(artifact, omissions)) runner_process["creditedWithinDeadline"] = runner_timely and runner_completed if summary["confirmedViolation"]: summary["findingReturnedByMillis"] = int((time.monotonic() - started) * 1000) + preparation_cost summary["confirmedViolationWithinBudget"] = summary["findingReturnedByMillis"] <= frozen["totalBudgetMillis"] \ and runner_process["creditedWithinDeadline"] - batches, omissions = replay_inputs(artifact) - summary["omissions"] = omissions if not runner_process["creditedWithinDeadline"]: raise ValueError("Runner did not complete successfully within its deadline; raw artifact and output are retained") prepared = read_json(output / case["id"] / "prepared.json") diff --git a/usvm-ts-fast-check/scripts/test_pbt_experiment.py b/usvm-ts-fast-check/scripts/test_pbt_experiment.py index b071a53d07..d41b2f6197 100644 --- a/usvm-ts-fast-check/scripts/test_pbt_experiment.py +++ b/usvm-ts-fast-check/scripts/test_pbt_experiment.py @@ -150,7 +150,7 @@ def communicate(*args, **kwargs): self.assertTrue(process["timedOut"]) - def campaign(self, *, fault=False): + def campaign(self, *, fault=False, request=None): inputs = [number(1)] artifact = {"result": {"mode": "COVERAGE_FEEDBACK", "stop": "COMPLETED", "concrete": {"status": "success", "numRuns": 12, @@ -158,7 +158,8 @@ def campaign(self, *, fault=False): {"phase": "generation", "admission": "admitted", "outcome": "holds", "input": {"status": "captured", "value": {"kind": "array", "elements": inputs}}}, ]}}, - "symbolic": {"status": "NO_VIOLATION_REACHED", "completedInputs": [inputs]}, + "symbolic": {"status": "NO_VIOLATION_REACHED", + "completedInputs": [{"inputs": inputs, "expectedOutcome": "HOLDS"}]}, "symbolicInputReplays": [ {"candidate": {"inputs": inputs}, "completedWithinDeadline": True, "predictedOutcomes": ["HOLDS"], "replay": {"admission": "ADMITTED", "outcome": "HOLDS"}}, @@ -172,12 +173,46 @@ def campaign(self, *, fault=False): result["symbolic"] = None result.pop("symbolicInputReplays") + if request is not None: + artifact.update({key: request[key] for key in ( + "property", "configuration", "validationReserveMillis", "completedInputLimit", "shrinkWitness", + )}) + artifact["property"] = dict(request["property"], assertions=[]) + artifact["property"]["predicate"] = dict(request["property"]["predicate"], executionKind="sync") + artifact["configuration"] = dict(request["configuration"], examples=[], exactReplay=False) + artifact["configuration"]["observationRequest"] = dict( + request["configuration"]["observationRequest"], maxPointsPerInvocation=8, + ) + artifact["configuration"]["coverageRequest"] = { + "scopes": ["source_under_test"], "includePatterns": [], "excludePatterns": [], + "includeUnexecutedSources": False, "admittedOnly": True, + } + artifact["metadata"] = {"sourceRoot": request["sourceRoot"], "requestedNodeCommand": request["nodeExecutable"]} + if "pbtAllocationMillis" in request: + artifact["pbtAllocationMillis"] = request["pbtAllocationMillis"] + result = artifact["result"] + result["mode"] = request["mode"] + result["propertyId"] = request["property"]["id"] + concrete = result["concrete"] + concrete["propertyId"] = result["propertyId"] + concrete["seed"] = request["configuration"]["seed"] + concrete["observations"]["verifiedSources"] = request["configuration"]["observationRequest"]["sources"] + if result["symbolic"] is not None: + result["symbolic"]["propertyId"] = result["propertyId"] + for replay in result["symbolicInputReplays"]: + replay["candidate"].update(propertyId=result["propertyId"], sources=concrete["observations"]["verifiedSources"]) + if request["mode"] == "PBT_ONLY": + result["symbolic"] = None + result.pop("symbolicInputReplays", None) + if request["mode"] == "SYMBOLIC_ONLY": + result["concrete"] = None + return artifact def complete_process(self, command, directory, stem, timeout_seconds, *, deadline=None): self.process_calls.append({"stem": stem, "timeout": timeout_seconds, "deadline": deadline}) - response = self.campaign() if stem == "runner" else {"records": [ + response = self.campaign(request=driver.read_json(directory / "request.json")) if stem == "runner" else {"records": [ {"verified": True, "creditedStatementIds": [stem]}, ]} driver.write_json(command[-1], response) @@ -185,7 +220,28 @@ def complete_process(self, command, directory, stem, timeout_seconds, *, deadlin return {"exitCode": 0, "timedOut": False} - def invoke_cell(self, process, *, wall_time=None): + def budget_process(self, command, directory, stem, timeout_seconds, **kwargs): + process = self.complete_process(command, directory, stem, timeout_seconds, **kwargs) + if stem != "runner": + return process + + artifact = driver.read_json(command[-1]) + result = artifact["result"] + if result["concrete"] is not None: + result["concrete"]["status"] = "failure" + result["concrete"]["failure"] = {"kind": "timeout"} + if result["mode"] == "PBT_ONLY": + result.update(stop="DEADLINE", stoppedAt="PBT") + elif result["mode"] == "SYMBOLIC_ONLY": + result.update(stop="DEADLINE", stoppedAt="CONFIRMATION") + result["symbolic"]["status"] = "TIMEOUT" + else: + result["diagnostic"] = "PBT allocation ended; completed prefix retained" + driver.write_json(command[-1], artifact) + return dict(process, exitCode=2) + + + def invoke_cell(self, process, *, wall_time=None, mode="COVERAGE_FEEDBACK"): self.cell_count += 1 output = self.root / f"output-{self.cell_count}" prepared = output / "subject" / "prepared.json" @@ -195,19 +251,21 @@ def invoke_cell(self, process, *, wall_time=None): source.write_text("export function holds(value: number) { return value === value; }", encoding="utf-8") case = {"id": "subject", "project": "fixture", "population": "correct-control", "sourceRoot": str(self.root), "sourceSha256": {"Subject.ts": driver.digest(source)}, - "property": {"id": "subject.holds"}, + "property": {"id": "subject.holds", + "inputs": [{"name": "value", "domain": {"kind": "integer", "min": -10, "max": 10}}], + "predicate": {"module": "Subject.ts", "exportName": "holds"}}, "preparation": {"elapsedMillis": 2000, "nativeSupported": True, "frontendSupported": True}} frozen = {"totalBudgetMillis": 30000, "classReplayReserveMillis": 8000, "validationReserveMillis": 1500, "pbtAllocationMillis": 4000, "numRuns": 12, "pbtInputLimit": 64, "symbolicInputLimit": 16, "maxArrayElements": 64, "maxObservationBytes": 65536, "nodeCommand": "node", "distribution": "/distribution", "adapter": "/adapter", "runtimeSha256": {}} - self.cell_directory = output / "subject" / "37" / "COVERAGE_FEEDBACK" + self.cell_directory = output / "subject" / "37" / mode wall_clock = self.clock.time if wall_time is None else wall_time with patch.object(driver.time, "monotonic", self.clock.monotonic), \ patch.object(driver.time, "time", wall_clock), \ patch.object(driver, "run_process", side_effect=process): - return driver.run_cell(frozen, case, seed=37, mode="COVERAGE_FEEDBACK", output=output) + return driver.run_cell(frozen, case, seed=37, mode=mode, output=output) def test_cell_shares_cutoffs_and_passes_reserves_without_rewriting_controls(self): @@ -249,7 +307,7 @@ def test_confirmed_original_fault_exit_one_remains_eligible(self): def process(command, directory, stem, timeout_seconds, **kwargs): result = self.complete_process(command, directory, stem, timeout_seconds, **kwargs) if stem == "runner": - driver.write_json(command[-1], self.campaign(fault=True)) + driver.write_json(command[-1], self.campaign(fault=True, request=driver.read_json(directory / "request.json"))) result["exitCode"] = 1 return result @@ -259,6 +317,175 @@ def process(command, directory, stem, timeout_seconds, **kwargs): self.assertTrue(summary["confirmedViolationWithinBudget"]) + def test_pbt_deadline_with_complete_prefix_reaches_native_class_verification(self): + summary = self.invoke_cell(self.budget_process, mode="PBT_ONLY") + + self.assertEqual(summary["campaignStop"], "DEADLINE") + self.assertEqual(summary["concreteFailureKind"], "timeout") + self.assertEqual(summary["runnerProcess"]["exitCode"], 2) + self.assertEqual([call["stem"] for call in self.process_calls], ["runner", "pbt"]) + self.assertTrue(summary["primaryEligible"], summary) + self.assertEqual(summary["coveredStatementIds"], ["pbt"]) + self.assertFalse(summary["confirmedViolation"]) + + + def test_hybrid_pbt_timeout_keeps_diagnostic_and_checks_both_prefixes(self): + summary = self.invoke_cell(self.budget_process) + + self.assertEqual(summary["campaignStop"], "COMPLETED") + self.assertEqual(summary["campaignDiagnostic"], "PBT allocation ended; completed prefix retained") + self.assertEqual(summary["concreteFailureKind"], "timeout") + self.assertEqual([call["deadline"] for call in self.process_calls], [120, 128, 128]) + self.assertTrue(summary["primaryEligible"], summary) + self.assertEqual(summary["coveredStatementIds"], ["pbt", "symbolic"]) + + + def test_symbolic_timeout_with_checked_completed_inputs_keeps_timeout_disposition(self): + summary = self.invoke_cell(self.budget_process, mode="SYMBOLIC_ONLY") + + self.assertEqual(summary["campaignStop"], "DEADLINE") + self.assertEqual(summary["symbolicStatus"], "TIMEOUT") + self.assertEqual([call["stem"] for call in self.process_calls], ["runner", "symbolic"]) + self.assertTrue(summary["primaryEligible"], summary) + self.assertEqual(summary["coveredStatementIds"], ["symbolic"]) + + + def test_hybrid_query_timeout_with_complete_prefixes_is_verified(self): + def process(command, directory, stem, timeout_seconds, **kwargs): + receipt = self.budget_process(command, directory, stem, timeout_seconds, **kwargs) + if stem == "runner": + artifact = driver.read_json(command[-1]) + artifact["result"].update(stop="DEADLINE", stoppedAt="CONFIRMATION") + artifact["result"]["symbolic"]["status"] = "TIMEOUT" + driver.write_json(command[-1], artifact) + return receipt + + summary = self.invoke_cell(process) + + self.assertEqual(summary["symbolicStatus"], "TIMEOUT") + self.assertEqual(summary["concreteFailureKind"], "timeout") + self.assertTrue(summary["coverageComplete"], summary) + self.assertTrue(summary["primaryEligible"], summary) + + + def test_budget_stop_does_not_admit_unsafe_or_mismatched_evidence(self): + mutations = ( + ("execution failure", lambda item: item["result"].update(stop="EXECUTION_FAILURE")), + ("unknown query", lambda item: item["result"]["symbolic"].update(status="SOLVER_UNKNOWN")), + ("unsupported query", lambda item: item["result"]["symbolic"].update(status="UNSUPPORTED")), + ("symbolic diagnostic", lambda item: item["result"]["symbolic"].update(diagnostics=["unsupported operation"])), + ("contract failure", lambda item: item["result"]["concrete"].update(failure={"kind": "contract"})), + ("unchecked target", lambda item: item["result"]["symbolic"].update(target={"id": "pending"})), + ("unconfirmed witness", lambda item: item["result"].update(confirmation={"status": "UNCONFIRMED"})), + ("dropped inputs", lambda item: item["result"]["concrete"]["observations"].update(droppedInvocations=1)), + ("pending inputs", lambda item: item["result"]["concrete"]["observations"].update(pendingInvocations=1)), + ("unsupported inputs", lambda item: item["result"]["concrete"]["observations"].update(unsupportedInputs=1)), + ("foreign property", lambda item: item["property"].update(id="other.property")), + ("foreign result", lambda item: item["result"].update(propertyId="other.property")), + ("foreign source root", lambda item: item["metadata"].update(sourceRoot="/another/source")), + ("foreign Node command", lambda item: item["metadata"].update(requestedNodeCommand="another-node")), + ("changed allocation", lambda item: item["configuration"].update(timeoutMillis=30000)), + ("foreign capture", lambda item: item["result"]["concrete"]["observations"].update(verifiedSources=[])), + ("foreign native source", lambda item: item["result"]["symbolicInputReplays"][0]["candidate"].update(sources=[])), + ("unknown native admission", lambda item: item["result"]["symbolicInputReplays"][0]["replay"].update(admission="UNKNOWN")), + ("native mismatch", lambda item: item["result"]["symbolicInputReplays"][0]["replay"].update(outcome="FALSE")), + ("changed symbolic prediction", lambda item: item["result"]["symbolic"]["completedInputs"][0].update(expectedOutcome="FALSE")), + ("late native receipt", lambda item: item["result"]["symbolicInputReplays"][0].update(completedWithinDeadline=False)), + ) + for reason, mutate in mutations: + with self.subTest(reason=reason): + self.clock = FakeClock() + self.process_calls = [] + + def process(command, directory, stem, timeout_seconds, **kwargs): + receipt = self.budget_process(command, directory, stem, timeout_seconds, **kwargs) + if stem == "runner": + artifact = driver.read_json(command[-1]) + mutate(artifact) + driver.write_json(command[-1], artifact) + return receipt + + summary = self.invoke_cell(process) + + self.assertFalse(summary["primaryEligible"], summary) + self.assertFalse(summary["coverageComplete"]) + self.assertEqual(summary["coveredStatementIds"], []) + self.assertEqual([call["stem"] for call in self.process_calls], ["runner"]) + + + def test_query_timeout_with_unchecked_completed_inputs_is_rejected(self): + for wrong_replay_input in (False, True): + with self.subTest(wrong_replay_input=wrong_replay_input): + self.clock = FakeClock() + self.process_calls = [] + + def process(command, directory, stem, timeout_seconds, **kwargs): + receipt = self.budget_process(command, directory, stem, timeout_seconds, **kwargs) + if stem == "runner": + artifact = driver.read_json(command[-1]) + if wrong_replay_input: + artifact["result"]["symbolicInputReplays"][0]["candidate"]["inputs"] = [number(2)] + else: + artifact["result"]["symbolic"]["completedInputs"].append({"inputs": [number(2)], "expectedOutcome": "HOLDS"}) + driver.write_json(command[-1], artifact) + return receipt + + summary = self.invoke_cell(process, mode="SYMBOLIC_ONLY") + + self.assertFalse(summary["primaryEligible"], summary) + self.assertEqual(summary["coveredStatementIds"], []) + self.assertEqual([call["stem"] for call in self.process_calls], ["runner"]) + + + def test_expected_budget_stop_still_needs_timely_successful_class_receipts(self): + for failure in ("late", "nonzero", "mismatch"): + with self.subTest(failure=failure): + self.clock = FakeClock() + self.process_calls = [] + + def process(command, directory, stem, timeout_seconds, **kwargs): + receipt = self.budget_process(command, directory, stem, timeout_seconds, **kwargs) + if stem == "pbt": + if failure == "late": + self.clock.advance(27) + elif failure == "nonzero": + receipt["exitCode"] = 2 + else: + driver.write_json(command[-1], {"records": [{"verified": False, "creditedStatementIds": []}]}) + return receipt + + summary = self.invoke_cell(process, mode="PBT_ONLY") + + self.assertFalse(summary["primaryEligible"], summary) + self.assertFalse(summary["coverageComplete"]) + self.assertEqual(summary["coveredStatementIds"], []) + self.assertEqual([call["deadline"] for call in self.process_calls], [120, 128]) + + + def test_budget_artifacts_from_late_killed_or_missing_process_results_are_not_admitted(self): + for failure in ("late", "killed", "missing"): + with self.subTest(failure=failure): + self.clock = FakeClock() + self.process_calls = [] + + def process(command, directory, stem, timeout_seconds, **kwargs): + receipt = self.budget_process(command, directory, stem, timeout_seconds, **kwargs) + if stem == "runner": + if failure == "late": + self.clock.advance(20) + elif failure == "killed": + receipt.update(timedOut=True, exitCode=-9) + else: + command[-1].unlink() + return receipt + + summary = self.invoke_cell(process, mode="PBT_ONLY") + + self.assertFalse(summary["primaryEligible"], summary) + self.assertEqual(summary["coveredStatementIds"], []) + self.assertEqual([call["stem"] for call in self.process_calls], ["runner"]) + + def test_incomplete_killed_and_late_artifacts_keep_diagnostics_without_credit(self): for exit_code, timed_out, elapsed in ((2, False, 0), (-9, True, 0), (0, False, 20), (1, False, 20)): with self.subTest(exit_code=exit_code, timed_out=timed_out, elapsed=elapsed): @@ -267,7 +494,7 @@ def test_incomplete_killed_and_late_artifacts_keep_diagnostics_without_credit(se def process(command, directory, stem, timeout_seconds, **kwargs): result = self.complete_process(command, directory, stem, timeout_seconds, **kwargs) - artifact = self.campaign(fault=True) + artifact = self.campaign(fault=True, request=driver.read_json(directory / "request.json")) artifact["result"]["diagnostic"] = {"message": "retained reason"} driver.write_json(command[-1], artifact) self.clock.advance(elapsed)