From 0282e92dda7c420d04d1551bc1e1356e3f5605bd Mon Sep 17 00:00:00 2001 From: Aleksei Menshutin Date: Sat, 17 Oct 2026 00:56:07 +0300 Subject: [PATCH 1/2] Credit checked class coverage from confirmed auxiliary witnesses --- usvm-ts-fast-check/scripts/pbt_experiment.py | 59 ++++++++++++++-- .../scripts/test_pbt_experiment.py | 68 +++++++++++++++++++ 2 files changed, 123 insertions(+), 4 deletions(-) diff --git a/usvm-ts-fast-check/scripts/pbt_experiment.py b/usvm-ts-fast-check/scripts/pbt_experiment.py index cfd46872e8..83d86689a5 100644 --- a/usvm-ts-fast-check/scripts/pbt_experiment.py +++ b/usvm-ts-fast-check/scripts/pbt_experiment.py @@ -192,12 +192,37 @@ def confirmed_symbolic_input(artifact): return {"inputs": candidate["inputs"], "admission": "ADMITTED", "outcome": native["outcome"]} +def confirmed_auxiliary_input(artifact, attempt): + """Consume a complete original-oracle confirmation, preserving the original input rather than a shrink.""" + confirmation = attempt.get("confirmation") or {} + candidate = confirmation.get("candidate") or {} + native = confirmation.get("originalReplay") or {} + target = candidate.get("target") or {} + status = confirmation.get("status") + original_fault = status == "CONFIRMED_PROPERTY_VIOLATION" and native.get("outcome") in ("FALSE", "THREW") + hypothesis_refuted = status == "CONFIRMED_HYPOTHESIS_REFUTATION" and native.get("outcome") == "HOLDS" \ + and native.get("targetMatched") is True \ + and any(event.get("binding") == target.get("binding") and event.get("guard") is True + and event.get("holds") is False for event in confirmation.get("contextualRefutations", [])) + declared_checks = {assertion["id"] for assertion in artifact["property"].get("assertions", [])} + original_check_matches = not declared_checks or (native.get("failure") or {}).get("checkId") in declared_checks + eligible = attempt.get("completedWithinDeadline") is True and attempt.get("searchStatus") == "REFUTATION_REACHED" \ + and candidate == attempt.get("witness") and candidate.get("inputs") is not None \ + and candidate.get("propertyId") == artifact["result"].get("propertyId") \ + and target.get("role") == "HYPOTHESIS" and not confirmation.get("diagnostic") \ + and native_coverage_matches(candidate, native) \ + and (hypothesis_refuted or original_fault and original_check_matches) + if not eligible: + return None + return {"inputs": candidate["inputs"], "admission": "ADMITTED", "outcome": native["outcome"]} + + def replay_inputs(artifact): result = artifact["result"] concrete = result.get("concrete") omissions = {"pbtDroppedInvocations": 0, "pbtUnavailableInputs": 0, "pbtUnsupportedInputs": 0, "pbtPendingInvocations": 0, "symbolicOmittedInputs": 0, - "symbolicUncheckedInputs": 0} + "symbolicUncheckedInputs": 0, "auxiliaryUncheckedInputs": 0} batches = [] if concrete is not None: export = concrete.get("inputExport") @@ -251,6 +276,19 @@ def replay_inputs(artifact): elif symbolic.get("target") is not None and input_key(symbolic.get("inputs")) not in replay_keys: omissions["symbolicUncheckedInputs"] += 1 batches.append({"id": "symbolic", "inputs": invocations, "isolateInputs": True}) + auxiliary = (result.get("relationGuidance") or {}).get("attempts", []) + if auxiliary: + invocations = [] + for attempt in auxiliary: + has_candidate = attempt.get("witness") is not None or attempt.get("searchStatus") == "REFUTATION_REACHED" + if not has_candidate: + continue + confirmed = confirmed_auxiliary_input(artifact, attempt) + if confirmed is None: + omissions["auxiliaryUncheckedInputs"] += 1 + else: + invocations.append(confirmed) + batches.append({"id": "auxiliary", "inputs": invocations, "isolateInputs": True}) if concrete is None and result["mode"] != "SYMBOLIC_ONLY": omissions["pbtUnavailableInputs"] += 1 return batches, omissions @@ -305,6 +343,15 @@ def run_cell(frozen, case, seed, mode, output): artifact = read_json(directory / "campaign.json") result = artifact["result"] summary["campaignElapsedMillis"] = result["elapsedMillis"] + relation_trace = result.get("relationGuidance") or {} + relation_attempts = relation_trace.get("attempts", []) + summary["relationElapsedMillis"] = relation_trace.get("elapsedMillis", 0) + summary["originalQueryElapsedMillis"] = relation_trace.get("originalElapsedMillis") + summary["relationStatuses"] = [attempt["searchStatus"] for attempt in relation_attempts] + summary["relationDiagnostics"] = relation_trace.get("diagnostics", []) + [ + diagnostic for attempt in relation_attempts + for diagnostic in attempt.get("diagnostics", []) + attempt.get("bindingDiagnostics", []) + ] confirmation = result.get("confirmation") or {} failure = (result.get("concrete") or {}).get("failure") or {} summary["confirmedViolation"] = failure.get("kind") == "property" or confirmation.get("status") == "CONFIRMED_PROPERTY_VIOLATION" @@ -322,9 +369,11 @@ def run_cell(frozen, case, seed, mode, output): symbolic_supported = summary["symbolicStatus"] in ( "VIOLATION_REACHED", "NO_VIOLATION_REACHED", "PRECONDITION_REJECTED", "TIMEOUT", ) - 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"] + stopped_on_original_fault = result.get("symbolic") is None and summary["confirmedViolation"] + relation_supported = all(status in ("REFUTATION_REACHED", "NO_REFUTATION_REACHED", "PRECONDITION_REJECTED", "TIMEOUT") + for status in summary["relationStatuses"]) + summary["analysisSupported"] = (mode == "PBT_ONLY" or symbolic_supported or stopped_on_original_fault) \ + and not summary["symbolicDiagnostics"] and relation_supported and not summary["relationDiagnostics"] if summary["confirmedViolation"]: summary["findingReturnedByMillis"] = int((time.monotonic() - started) * 1000) + preparation_cost summary["confirmedViolationWithinBudget"] = summary["findingReturnedByMillis"] <= frozen["totalBudgetMillis"] \ @@ -417,6 +466,8 @@ def summarize(output): row.pop("replayProcesses", None) row["omissions"] = json.dumps(row.get("omissions", {}), sort_keys=True) row["symbolicDiagnostics"] = json.dumps(row.get("symbolicDiagnostics", []), sort_keys=True) + row["relationDiagnostics"] = json.dumps(row.get("relationDiagnostics", []), sort_keys=True) + row["relationStatuses"] = json.dumps(row.get("relationStatuses", [])) rows.append(row) columns = sorted(set().union(*(row.keys() for row in rows))) with (output / "summary.csv").open("w", newline="", encoding="utf-8") as file: diff --git a/usvm-ts-fast-check/scripts/test_pbt_experiment.py b/usvm-ts-fast-check/scripts/test_pbt_experiment.py index 80c293d750..ba3dada6e4 100644 --- a/usvm-ts-fast-check/scripts/test_pbt_experiment.py +++ b/usvm-ts-fast-check/scripts/test_pbt_experiment.py @@ -31,6 +31,22 @@ def symbolic_artifact(outcome="HOLDS"): "completedWithinDeadline": True, "predictedOutcomes": [outcome]}]}} +def auxiliary_artifact(): + artifact = symbolic_artifact() + raw = artifact["result"]["symbolicInputReplays"][0] + candidate = copy.deepcopy(raw["candidate"]) + candidate["target"] = {"id": "relation", "role": "HYPOTHESIS", "binding": {"pointId": "relation", "revision": "hash"}} + native = copy.deepcopy(raw["replay"]) + confirmation = {"status": "CONFIRMED_HYPOTHESIS_REFUTATION", "candidate": candidate, "originalReplay": native, + "contextualRefutations": [{"binding": candidate["target"]["binding"], "guard": True, "holds": False}], + "reduction": {"inputs": [number(0)]}} + artifact["result"].pop("symbolicInputReplays") + artifact["result"].pop("symbolic") + artifact["result"]["relationGuidance"] = {"attempts": [{"searchStatus": "REFUTATION_REACHED", "witness": candidate, + "confirmation": confirmation, "completedWithinDeadline": True}]} + return artifact + + class ExperimentTest(unittest.TestCase): def setUp(self): self.temporary = tempfile.TemporaryDirectory() @@ -164,6 +180,58 @@ def test_budget_driven_quota_is_independent_of_the_explicit_large_capture_limit( driver.validate_manifest(manifest) + def test_confirmed_auxiliary_refutation_with_passing_original_retains_its_original_input(self): + artifact = auxiliary_artifact() + + batches, omissions = driver.replay_inputs(artifact) + + self.assertEqual(batches[0]["id"], "auxiliary") + self.assertEqual(batches[0]["inputs"], [{"inputs": [number(1)], "admission": "ADMITTED", "outcome": "HOLDS"}]) + self.assertTrue(batches[0]["isolateInputs"]) + self.assertFalse(any(omissions.values())) + + + def test_unchecked_late_or_wrong_identity_auxiliary_inputs_never_become_class_coverage(self): + for change in ("late", "raw", "missing-identity", "missing-original", "different-input", "no-context"): + artifact = auxiliary_artifact() + attempt = artifact["result"]["relationGuidance"]["attempts"][0] + if change == "late": + attempt["completedWithinDeadline"] = False + elif change == "raw": + attempt["confirmation"]["status"] = "PROPERTY_HOLDS" + elif change == "missing-identity": + attempt["confirmation"]["originalReplay"]["coverage"].pop("identity") + elif change == "missing-original": + attempt["confirmation"].pop("originalReplay") + elif change == "different-input": + attempt["confirmation"]["candidate"] = dict(attempt["witness"], inputs=[number(2)]) + else: + attempt["confirmation"]["contextualRefutations"] = [] + + batches, omissions = driver.replay_inputs(artifact) + + self.assertEqual(batches[0]["inputs"], [], change) + self.assertEqual(omissions["auxiliaryUncheckedInputs"], 1, change) + + + def test_auxiliary_original_fault_requires_its_named_original_check_and_preserves_false(self): + artifact = auxiliary_artifact() + artifact["property"]["assertions"] = [{"id": "bound"}] + confirmation = artifact["result"]["relationGuidance"]["attempts"][0]["confirmation"] + confirmation["status"] = "CONFIRMED_PROPERTY_VIOLATION" + confirmation["originalReplay"].update(outcome="FALSE", failure={"checkId": "bound"}, targetMatched=False) + + batches, omissions = driver.replay_inputs(artifact) + + self.assertEqual(batches[0]["inputs"][0]["outcome"], "FALSE") + self.assertFalse(any(omissions.values())) + + confirmation["originalReplay"]["failure"]["checkId"] = "other" + batches, omissions = driver.replay_inputs(artifact) + self.assertEqual(batches[0]["inputs"], []) + self.assertEqual(omissions["auxiliaryUncheckedInputs"], 1) + + def test_process_timeout_retains_raw_output_and_failure(self): process = driver.run_process( ["node", "-e", "process.stdout.write('started'); setInterval(() => {}, 1000)"], self.root, "slow", 0.2, From ec2a4dc5a8f52781fb3cdbcc8aa56cb3ab63a994 Mon Sep 17 00:00:00 2001 From: Aleksei Menshutin Date: Sat, 17 Oct 2026 01:44:25 +0300 Subject: [PATCH 2/2] Validate auxiliary original receipts before experiment coverage replay --- usvm-ts-fast-check/scripts/pbt_experiment.py | 162 +++++++++++++++--- .../scripts/test_pbt_experiment.py | 148 ++++++++++++++-- 2 files changed, 268 insertions(+), 42 deletions(-) diff --git a/usvm-ts-fast-check/scripts/pbt_experiment.py b/usvm-ts-fast-check/scripts/pbt_experiment.py index 83d86689a5..1536bad25c 100644 --- a/usvm-ts-fast-check/scripts/pbt_experiment.py +++ b/usvm-ts-fast-check/scripts/pbt_experiment.py @@ -192,31 +192,124 @@ def confirmed_symbolic_input(artifact): return {"inputs": candidate["inputs"], "admission": "ADMITTED", "outcome": native["outcome"]} -def confirmed_auxiliary_input(artifact, attempt): - """Consume a complete original-oracle confirmation, preserving the original input rather than a shrink.""" +def _same_auxiliary_registration(artifact, candidate): + property = artifact.get("property") or {} + property_id = property.get("id") + manifest = dict(property) + manifest.pop("id", None) + manifest["propertyId"] = property_id + metadata = artifact.get("metadata") or {} + configured = ((artifact.get("configuration") or {}).get("observationRequest") or {}).get("sources") + sources = candidate.get("sources") + if not property_id or candidate.get("propertyId") != property_id \ + or artifact["result"].get("propertyId") != property_id or candidate.get("manifest") != manifest \ + or not isinstance(candidate.get("inputs"), list) or len(candidate["inputs"]) != len(property.get("inputs", [])) \ + or not isinstance(sources, list) or not sources or sources != configured: + return False + valid_sources = all(isinstance(source, dict) and isinstance(source.get("module"), str) + and isinstance(source.get("sha256"), str) for source in sources) + if not valid_sources: + return False + pins = {source["module"]: source["sha256"] for source in sources} + selected = metadata.get("sources") or [] + return len(pins) == len(sources) and bool(selected) and isinstance(metadata.get("sourceRoot"), str) \ + and bool(metadata["sourceRoot"]) \ + and all(isinstance(source, dict) and pins.get(source.get("module")) == source.get("sha256") for source in selected) + + +def _same_auxiliary_target(artifact, attempt, candidate): + trace = artifact["result"].get("relationGuidance") or {} + selection = trace.get("selection") or {} + inference = selection.get("inference") or {} + selected = attempt.get("selected") or {} + relation = selected.get("relation") or {} + pair = relation.get("pair") or {} + controls = artifact.get("relationGuidanceControls") or {} + left, right = pair.get("left") or {}, pair.get("right") or {} + target = candidate.get("target") or {} + binding = target.get("binding") or {} + target_id = f'{left.get("assertionId")}:{left.get("id")}:{relation.get("template")}:{right.get("id")}' + exact = attempt.get("bindingStatus") == "EXACT" and not attempt.get("bindingDiagnostics") \ + and selected in selection.get("candidates", []) and relation in inference.get("relations", []) \ + and relation.get("status") == "SAMPLED" and pair in controls.get("pairs", []) \ + and inference.get("property") == artifact.get("property") \ + and inference.get("propertyId") == candidate.get("propertyId") \ + and inference.get("sources") == candidate.get("sources") \ + and target.get("role") == "HYPOTHESIS" and bool(binding) and target.get("id") == target_id \ + and binding.get("pointId") == target_id and binding.get("occurrence") == 0 \ + and binding.get("revision") == candidate.get("manifestSha256") and bool(binding.get("revision")) + if not exact: + return False + if left.get("kind") == right.get("kind") == "entry-argument": + expected = {"leftInputIndex": left.get("inputIndex"), "rightInputIndex": right.get("inputIndex"), + "template": relation.get("template")} + return target.get("entryRelation") == expected and target.get("dslNumericLocalRelation") is None \ + and binding.get("contextId") == "original-entry" + result, original = (left, right) if left.get("compilerLocal") else (right, left) + compiler = result.get("compilerLocal") + expected = {"compiler": compiler, "inputIndex": original.get("inputIndex"), + "resultIsLeft": result == left, "template": relation.get("template")} + return result.get("kind") == "return" and original.get("kind") == "entry-argument" and bool(compiler) \ + and controls.get("compilerRetained") is True \ + and target.get("dslNumericLocalRelation") == expected and target.get("entryRelation") is None \ + and binding.get("contextId") == "original-compiler-local" + + +def _confirmed_auxiliary_replay(artifact, attempt): confirmation = attempt.get("confirmation") or {} candidate = confirmation.get("candidate") or {} native = confirmation.get("originalReplay") or {} target = candidate.get("target") or {} - status = confirmation.get("status") - original_fault = status == "CONFIRMED_PROPERTY_VIOLATION" and native.get("outcome") in ("FALSE", "THREW") - hypothesis_refuted = status == "CONFIRMED_HYPOTHESIS_REFUTATION" and native.get("outcome") == "HOLDS" \ - and native.get("targetMatched") is True \ - and any(event.get("binding") == target.get("binding") and event.get("guard") is True - and event.get("holds") is False for event in confirmation.get("contextualRefutations", [])) + binding = target.get("binding") + context_matches = bool(binding) and any( + event.get("binding") == binding and event.get("guard") is True and event.get("holds") is False + for event in confirmation.get("contextualRefutations", []) + ) and any(event.get("binding") == binding and event.get("guard") is True and event.get("holds") is False + for event in native.get("events", [])) declared_checks = {assertion["id"] for assertion in artifact["property"].get("assertions", [])} original_check_matches = not declared_checks or (native.get("failure") or {}).get("checkId") in declared_checks + original_fault = confirmation.get("status") == "CONFIRMED_PROPERTY_VIOLATION" \ + and native.get("outcome") in ("FALSE", "THREW") and original_check_matches + hypothesis_refuted = confirmation.get("status") == "CONFIRMED_HYPOTHESIS_REFUTATION" and native.get("outcome") == "HOLDS" eligible = attempt.get("completedWithinDeadline") is True and attempt.get("searchStatus") == "REFUTATION_REACHED" \ and candidate == attempt.get("witness") and candidate.get("inputs") is not None \ - and candidate.get("propertyId") == artifact["result"].get("propertyId") \ - and target.get("role") == "HYPOTHESIS" and not confirmation.get("diagnostic") \ - and native_coverage_matches(candidate, native) \ - and (hypothesis_refuted or original_fault and original_check_matches) - if not eligible: + and _same_auxiliary_registration(artifact, candidate) and _same_auxiliary_target(artifact, attempt, candidate) \ + and native.get("admission") == "ADMITTED" and native.get("targetMatched") is True \ + and not native.get("diagnostic") and context_matches and (hypothesis_refuted or original_fault) + return (candidate, native) if eligible else None + + +def confirmed_auxiliary_input(artifact, attempt): + """Use the original confirmed input; the later original/counter replay measures its class coverage.""" + confirmed = _confirmed_auxiliary_replay(artifact, attempt) + if confirmed is None or (attempt.get("confirmation") or {}).get("diagnostic"): + return None + candidate, native = confirmed + coverage = native.get("coverage") or {} + metadata = artifact["metadata"] + source_root = str(Path(metadata["sourceRoot"]).resolve()) + hashes = (coverage.get("identity") or {}).get("sourceSha256") or {} + sources_match = all(hashes.get(str((Path(source_root) / source["module"]).resolve())) == source["sha256"] + for source in candidate["sources"]) + roots_match = coverage.get("provenance", {}).get("sourceRoots") == [source_root] + if not native_coverage_matches(candidate, native) or not sources_match or not roots_match: return None return {"inputs": candidate["inputs"], "admission": "ADMITTED", "outcome": native["outcome"]} +def confirmed_original_violation(artifact): + result = artifact["result"] + if ((result.get("concrete") or {}).get("failure") or {}).get("kind") == "property": + return True + canonical = result.get("confirmation") or {} + if canonical.get("status") != "CONFIRMED_PROPERTY_VIOLATION": + return False + if (canonical.get("candidate") or {}).get("target", {}).get("role") != "HYPOTHESIS": + return True # Existing ordinary-control confirmation contract. + return any(attempt.get("confirmation") == canonical and _confirmed_auxiliary_replay(artifact, attempt) is not None + for attempt in (result.get("relationGuidance") or {}).get("attempts", [])) + + def replay_inputs(artifact): result = artifact["result"] concrete = result.get("concrete") @@ -280,7 +373,8 @@ def replay_inputs(artifact): if auxiliary: invocations = [] for attempt in auxiliary: - has_candidate = attempt.get("witness") is not None or attempt.get("searchStatus") == "REFUTATION_REACHED" + has_candidate = attempt.get("witness") is not None or attempt.get("searchStatus") == "REFUTATION_REACHED" \ + or (attempt.get("confirmation") or {}).get("candidate") is not None if not has_candidate: continue confirmed = confirmed_auxiliary_input(artifact, attempt) @@ -289,6 +383,10 @@ def replay_inputs(artifact): else: invocations.append(confirmed) batches.append({"id": "auxiliary", "inputs": invocations, "isolateInputs": True}) + canonical = result.get("confirmation") or {} + canonical_auxiliary = (canonical.get("candidate") or {}).get("target", {}).get("role") == "HYPOTHESIS" + if canonical_auxiliary and not any(attempt.get("confirmation") == canonical for attempt in auxiliary): + omissions["auxiliaryUncheckedInputs"] += 1 if concrete is None and result["mode"] != "SYMBOLIC_ONLY": omissions["pbtUnavailableInputs"] += 1 return batches, omissions @@ -300,6 +398,28 @@ def full_budget_eligible(summary): return summary["comparisonKind"] == "equal-budget" and summary["primaryEligible"] and not quota_stopped_pbt +def _relation_summary(result): + trace = result.get("relationGuidance") or {} + attempts = trace.get("attempts", []) + inference = (trace.get("selection") or {}).get("inference") or {} + diagnostics = list(trace.get("diagnostics", [])) + for attempt in attempts: + diagnostics.extend(attempt.get("diagnostics", []) + attempt.get("bindingDiagnostics", [])) + diagnostic = (attempt.get("confirmation") or {}).get("diagnostic") + if diagnostic: + diagnostics.append(diagnostic) + relations = inference.get("relations", []) + inference.get("domainImpliedRelations", []) + return {"relationElapsedMillis": trace.get("elapsedMillis", 0), + "originalQueryElapsedMillis": trace.get("originalElapsedMillis"), + "relationStatuses": [attempt.get("searchStatus") for attempt in attempts], + "relationDiagnostics": diagnostics, + "relationLimitations": sorted({item for relation in relations for item in relation.get("limitations", [])}), + "relationOmittedPairs": inference.get("omittedPairs", 0), + "relationOmittedInvocations": inference.get("omittedInvocations", 0), + "relationDroppedInvocations": inference.get("droppedInvocations", 0), + "relationDroppedPoints": inference.get("droppedPoints", 0)} + + def run_cell(frozen, case, seed, mode, output): directory = output / case["id"] / str(seed) / mode directory.mkdir(parents=True) @@ -343,18 +463,9 @@ def run_cell(frozen, case, seed, mode, output): artifact = read_json(directory / "campaign.json") result = artifact["result"] summary["campaignElapsedMillis"] = result["elapsedMillis"] - relation_trace = result.get("relationGuidance") or {} - relation_attempts = relation_trace.get("attempts", []) - summary["relationElapsedMillis"] = relation_trace.get("elapsedMillis", 0) - summary["originalQueryElapsedMillis"] = relation_trace.get("originalElapsedMillis") - summary["relationStatuses"] = [attempt["searchStatus"] for attempt in relation_attempts] - summary["relationDiagnostics"] = relation_trace.get("diagnostics", []) + [ - diagnostic for attempt in relation_attempts - for diagnostic in attempt.get("diagnostics", []) + attempt.get("bindingDiagnostics", []) - ] - confirmation = result.get("confirmation") or {} + summary.update(_relation_summary(result)) failure = (result.get("concrete") or {}).get("failure") or {} - summary["confirmedViolation"] = failure.get("kind") == "property" or confirmation.get("status") == "CONFIRMED_PROPERTY_VIOLATION" + summary["confirmedViolation"] = confirmed_original_violation(artifact) summary["campaignStop"] = result["stop"] summary["symbolicStatus"] = (result.get("symbolic") or {}).get("status") summary["symbolicDiagnostics"] = (result.get("symbolic") or {}).get("diagnostics", []) @@ -468,6 +579,7 @@ def summarize(output): row["symbolicDiagnostics"] = json.dumps(row.get("symbolicDiagnostics", []), sort_keys=True) row["relationDiagnostics"] = json.dumps(row.get("relationDiagnostics", []), sort_keys=True) row["relationStatuses"] = json.dumps(row.get("relationStatuses", [])) + row["relationLimitations"] = json.dumps(row.get("relationLimitations", [])) rows.append(row) columns = sorted(set().union(*(row.keys() for row in rows))) with (output / "summary.csv").open("w", newline="", encoding="utf-8") as file: diff --git a/usvm-ts-fast-check/scripts/test_pbt_experiment.py b/usvm-ts-fast-check/scripts/test_pbt_experiment.py index ba3dada6e4..717e7c1e24 100644 --- a/usvm-ts-fast-check/scripts/test_pbt_experiment.py +++ b/usvm-ts-fast-check/scripts/test_pbt_experiment.py @@ -1,5 +1,6 @@ import importlib.util import copy +import json import os from pathlib import Path import struct @@ -32,19 +33,42 @@ def symbolic_artifact(outcome="HOLDS"): def auxiliary_artifact(): - artifact = symbolic_artifact() - raw = artifact["result"]["symbolicInputReplays"][0] - candidate = copy.deepcopy(raw["candidate"]) - candidate["target"] = {"id": "relation", "role": "HYPOTHESIS", "binding": {"pointId": "relation", "revision": "hash"}} - native = copy.deepcopy(raw["replay"]) + source = {"module": "p.ts", "line": 1, "column": 1} + sources = [{"module": "p.ts", "sha256": "a" * 64}] + left = {"id": "left", "assertionId": "bound", "operandId": "left", "source": source, + "callSite": source, "kind": "entry-argument", "inputIndex": 0} + right = dict(left, id="right", operandId="right", inputIndex=1) + relation = {"pair": {"left": left, "right": right}, "template": "STRICT_EQUAL", "status": "SAMPLED"} + selected = {"relation": relation, "relevance": "UNKNOWN"} + property = {"id": "p", "predicate": {"module": "p.ts", "exportName": "predicate", "executionKind": "sync"}, + "inputs": [{"name": name, "domain": {"kind": "integer", "min": 0, "max": 10}} for name in ("a", "b")], + "assertions": [{"id": "bound", "source": source, "testedCall": source, + "operands": [{"id": name, "source": source} for name in ("left", "right")]}]} + manifest = dict(property, propertyId="p") + manifest.pop("id") + target_id = "bound:left:STRICT_EQUAL:right" + binding = {"pointId": target_id, "contextId": "original-entry", "revision": "b" * 64, "occurrence": 0} + candidate = {"propertyId": "p", "manifestSha256": "b" * 64, "manifest": manifest, + "sources": sources, "inputs": [number(1), number(2)], + "target": {"id": target_id, "role": "HYPOTHESIS", "binding": binding, + "entryRelation": {"leftInputIndex": 0, "rightInputIndex": 1, "template": "STRICT_EQUAL"}}} + event = {"binding": binding, "guard": True, "holds": False} + native = {"admission": "ADMITTED", "outcome": "HOLDS", "targetMatched": True, "events": [event], + "coverage": {"propertyId": "p", "admittedInvocations": 1, + "identity": {"manifestSha256": "b" * 64, "sourceSha256": {"/fixture/p.ts": "a" * 64}}, + "provenance": {"sourceRoots": ["/fixture"], "request": {"admittedOnly": True}}}} confirmation = {"status": "CONFIRMED_HYPOTHESIS_REFUTATION", "candidate": candidate, "originalReplay": native, - "contextualRefutations": [{"binding": candidate["target"]["binding"], "guard": True, "holds": False}], - "reduction": {"inputs": [number(0)]}} - artifact["result"].pop("symbolicInputReplays") - artifact["result"].pop("symbolic") - artifact["result"]["relationGuidance"] = {"attempts": [{"searchStatus": "REFUTATION_REACHED", "witness": candidate, - "confirmation": confirmation, "completedWithinDeadline": True}]} - return artifact + "contextualRefutations": [event], "reduction": {"inputs": [number(0)]}} + artifact = {"property": property, "metadata": {"sourceRoot": "/fixture", "sources": sources}, + "configuration": {"observationRequest": {"sources": sources}}, + "relationGuidanceControls": {"pairs": [relation["pair"]]}, + "result": {"propertyId": "p", "mode": "SEQUENTIAL", "relationGuidance": { + "selection": {"candidates": [selected], "inference": { + "propertyId": "p", "property": property, "sources": sources, "relations": [relation]}}, + "attempts": [{"selected": selected, "bindingStatus": "EXACT", "bindingDiagnostics": [], + "searchStatus": "REFUTATION_REACHED", "witness": candidate, + "confirmation": confirmation, "completedWithinDeadline": True}]}}} + return json.loads(json.dumps(artifact)) class ExperimentTest(unittest.TestCase): @@ -186,9 +210,9 @@ def test_confirmed_auxiliary_refutation_with_passing_original_retains_its_origin batches, omissions = driver.replay_inputs(artifact) self.assertEqual(batches[0]["id"], "auxiliary") - self.assertEqual(batches[0]["inputs"], [{"inputs": [number(1)], "admission": "ADMITTED", "outcome": "HOLDS"}]) + self.assertEqual(batches[0]["inputs"], [{"inputs": [number(1), number(2)], "admission": "ADMITTED", "outcome": "HOLDS"}]) self.assertTrue(batches[0]["isolateInputs"]) - self.assertFalse(any(omissions.values())) + self.assertEqual(omissions["auxiliaryUncheckedInputs"], 0) def test_unchecked_late_or_wrong_identity_auxiliary_inputs_never_become_class_coverage(self): @@ -216,22 +240,112 @@ def test_unchecked_late_or_wrong_identity_auxiliary_inputs_never_become_class_co def test_auxiliary_original_fault_requires_its_named_original_check_and_preserves_false(self): artifact = auxiliary_artifact() - artifact["property"]["assertions"] = [{"id": "bound"}] confirmation = artifact["result"]["relationGuidance"]["attempts"][0]["confirmation"] confirmation["status"] = "CONFIRMED_PROPERTY_VIOLATION" - confirmation["originalReplay"].update(outcome="FALSE", failure={"checkId": "bound"}, targetMatched=False) + confirmation["originalReplay"].update(outcome="FALSE", failure={"checkId": "bound"}, targetMatched=True) batches, omissions = driver.replay_inputs(artifact) self.assertEqual(batches[0]["inputs"][0]["outcome"], "FALSE") - self.assertFalse(any(omissions.values())) + self.assertEqual(omissions["auxiliaryUncheckedInputs"], 0) + artifact["result"]["confirmation"] = confirmation + self.assertTrue(driver.confirmed_original_violation(artifact)) + confirmation["originalReplay"]["targetMatched"] = False + self.assertFalse(driver.confirmed_original_violation(artifact)) + self.assertIsNone(driver.confirmed_auxiliary_input(artifact, artifact["result"]["relationGuidance"]["attempts"][0])) + confirmation["originalReplay"]["targetMatched"] = True confirmation["originalReplay"]["failure"]["checkId"] = "other" batches, omissions = driver.replay_inputs(artifact) self.assertEqual(batches[0]["inputs"], []) self.assertEqual(omissions["auxiliaryUncheckedInputs"], 1) + def test_auxiliary_receipts_require_trusted_registration_selected_point_and_complete_binding(self): + changes = ("manifest", "source", "missing-binding", "wrong-binding", "ambiguous", "wrong-target", + "unselected", "unconfigured-pair", "wrong-point", "missing-native-event", "wrong-input") + for change in changes: + with self.subTest(change=change): + artifact = auxiliary_artifact() + attempt = artifact["result"]["relationGuidance"]["attempts"][0] + confirmation = attempt["confirmation"] + candidate = confirmation["candidate"] + native = confirmation["originalReplay"] + if change == "manifest": + candidate["manifest"]["predicate"]["exportName"] = "other" + candidate["manifestSha256"] = "c" * 64 + native["coverage"]["identity"]["manifestSha256"] = "c" * 64 + candidate["target"]["binding"]["revision"] = "c" * 64 + elif change == "source": + candidate["sources"][0]["sha256"] = "c" * 64 + native["coverage"]["identity"]["sourceSha256"]["/fixture/p.ts"] = "c" * 64 + elif change == "missing-binding": + candidate["target"]["binding"] = None + native["events"][0]["binding"] = None + confirmation["contextualRefutations"][0]["binding"] = None + elif change == "wrong-binding": + candidate["target"]["binding"]["contextId"] = "helper" + elif change == "ambiguous": + attempt["bindingStatus"] = "AMBIGUOUS" + elif change == "wrong-target": + candidate["target"]["entryRelation"]["leftInputIndex"] = 1 + elif change == "unselected": + artifact["result"]["relationGuidance"]["selection"]["candidates"] = [] + elif change == "unconfigured-pair": + artifact["relationGuidanceControls"]["pairs"] = [] + elif change == "wrong-point": + attempt["selected"]["relation"]["pair"]["left"]["id"] = "other" + elif change == "missing-native-event": + native["events"] = [] + else: + candidate["inputs"] = [] + attempt["witness"] = copy.deepcopy(candidate) + + batches, omissions = driver.replay_inputs(artifact) + + self.assertEqual(batches[0]["inputs"], []) + self.assertEqual(omissions["auxiliaryUncheckedInputs"], 1) + + + def test_canonical_auxiliary_fault_keeps_its_role_and_requires_the_timely_named_attempt(self): + artifact = auxiliary_artifact() + attempt = artifact["result"]["relationGuidance"]["attempts"][0] + attempt["confirmation"]["status"] = "CONFIRMED_PROPERTY_VIOLATION" + attempt["confirmation"]["originalReplay"].update(outcome="FALSE", failure={"checkId": "bound"}) + artifact["result"]["confirmation"] = copy.deepcopy(attempt["confirmation"]) + artifact["result"]["symbolic"] = None + + self.assertTrue(driver.confirmed_original_violation(artifact)) + self.assertEqual(artifact["result"]["confirmation"]["candidate"]["target"]["role"], "HYPOTHESIS") + + artifact["result"]["relationGuidance"]["attempts"] = [] + self.assertFalse(driver.confirmed_original_violation(artifact)) + _, omissions = driver.replay_inputs(artifact) + self.assertEqual(omissions["auxiliaryUncheckedInputs"], 1) + + + def test_relation_summary_retains_capture_limits_costs_and_failed_coverage_confirmation(self): + artifact = auxiliary_artifact() + trace = artifact["result"]["relationGuidance"] + trace.update(elapsedMillis=4980, originalElapsedMillis=70, diagnostics=["trace-note"]) + trace["selection"]["inference"].update(droppedInvocations=56, droppedPoints=4, omittedInvocations=2, omittedPairs=1) + trace["selection"]["inference"]["relations"][0]["limitations"] = ["INCOMPLETE_CAPTURE", "TRUNCATED_INVOCATION"] + trace["attempts"][0]["confirmation"]["diagnostic"] = "backend.process.timeout" + + summary = driver._relation_summary(artifact["result"]) + + self.assertEqual(summary["relationElapsedMillis"], 4980) + self.assertEqual(summary["originalQueryElapsedMillis"], 70) + self.assertEqual(summary["relationDiagnostics"], ["trace-note", "backend.process.timeout"]) + self.assertEqual(summary["relationDroppedInvocations"], 56) + self.assertEqual(summary["relationDroppedPoints"], 4) + self.assertEqual(summary["relationOmittedInvocations"], 2) + self.assertEqual(summary["relationOmittedPairs"], 1) + self.assertEqual(summary["relationLimitations"], ["INCOMPLETE_CAPTURE", "TRUNCATED_INVOCATION"]) + _, omissions = driver.replay_inputs(artifact) + self.assertEqual(omissions["auxiliaryUncheckedInputs"], 1) + + def test_process_timeout_retains_raw_output_and_failure(self): process = driver.run_process( ["node", "-e", "process.stdout.write('started'); setInterval(() => {}, 1000)"], self.root, "slow", 0.2,