Use compact native inputs and strict coverage evidence in experiments

usvm-ts-fast-check/experiments/two-phase-smoke/manifest.json

@@ -1,6 +1,7 @@
 {
   "schemaVersion": 1,
   "experimentId": "two-phase-functional-smoke-v1",
+  "comparisonKind": "functional-quota",
   "seeds": [
     37
   ],

usvm-ts-fast-check/scripts/pbt_experiment.py

@@ -66,6 +66,8 @@ def verify_runtime(frozen):
 def validate_manifest(manifest):
     if manifest["schemaVersion"] != 1 or not manifest["experimentId"]:
         raise ValueError("Experiment identity or schema version is invalid")
+    if manifest.get("comparisonKind") not in ("functional-quota", "equal-budget"):
+        raise ValueError("Declare whether this is a functional quota check or an equal-budget comparison")
     seeds = manifest["seeds"]
     valid_seeds = seeds and len(set(seeds)) == len(seeds) and all(-(2 ** 31) <= seed < 2 ** 31 for seed in seeds)
     if not valid_seeds:
@@ -77,11 +79,11 @@ def validate_manifest(manifest):
         and 0 < manifest["pbtAllocationMillis"] < budget - replay_reserve - validation_reserve
     if not valid_budget:
         raise ValueError("PBT, runner validation and class replay allocations must fit the total budget")
-    valid_limits = 0 < manifest["numRuns"] <= manifest["pbtInputLimit"] <= 64 \
+    valid_limits = manifest["numRuns"] > 0 and 0 < manifest["pbtInputLimit"] <= 100_000 \
         and manifest["symbolicInputLimit"] > 0 and 1 <= manifest["maxArrayElements"] <= 64 \
-        and 1024 <= manifest["maxObservationBytes"] <= 65_536
+        and 1024 <= manifest["maxObservationBytes"] <= 2 * 1024 * 1024
     if not valid_limits:
-        raise ValueError("Frozen limits must fit the adapter's 64-invocation/64-element/65536-byte capture contract")
+        raise ValueError("Frozen limits must fit the compact export's 100000-record/64-element/2MiB contract")
     unique_cases = manifest["cases"] and len({case["id"] for case in manifest["cases"]}) == len(manifest["cases"])
     if not unique_cases:
         raise ValueError("The manifest must contain distinct case identities")
@@ -155,17 +157,59 @@ def freeze(manifest_path, output, distribution, adapter, node):
     return frozen
 
 
+def native_coverage_matches(candidate, native):
+    coverage = native.get("coverage") or {}
+    identity = coverage.get("identity") or {}
+    return native.get("admission") == "ADMITTED" and not native.get("diagnostic") \
+        and coverage.get("propertyId") == candidate.get("propertyId") \
+        and coverage.get("admittedInvocations") == 1 \
+        and coverage.get("provenance", {}).get("request", {}).get("admittedOnly") is True \
+        and bool(identity) and identity.get("manifestSha256") == candidate.get("manifestSha256")
+
+
+def confirmed_symbolic_input(artifact):
+    """Consume the existing original-property confirmation; do not confirm raw faults here."""
+    result = artifact["result"]
+    confirmation = result.get("confirmation") or {}
+    candidate = confirmation.get("candidate") or {}
+    native = confirmation.get("originalReplay") or {}
+    target = candidate.get("target") or {}
+    symbolic = result.get("symbolic") or {}
+    assertions = artifact["property"].get("assertions", [])
+    named_ids = {assertion["id"] for assertion in assertions}
+    target_id = target.get("id")
+    original_check_matches = target_id in named_ids and (native.get("failure") or {}).get("checkId") == target_id \
+        if assertions else target_id == artifact["property"]["predicate"]["exportName"]
+    valid = confirmation.get("status") == "CONFIRMED_PROPERTY_VIOLATION" \
+        and not confirmation.get("diagnostic") and candidate.get("propertyId") == result.get("propertyId") \
+        and target.get("role", "PROPERTY") == "PROPERTY" \
+        and original_check_matches and native.get("outcome") in ("FALSE", "THREW") \
+        and target.get("expectedOutcome") == native.get("outcome") \
+        and native.get("targetMatched") is True and candidate.get("inputs") == symbolic.get("inputs") \
+        and candidate.get("inputs") is not None and native_coverage_matches(candidate, native)
+    if not valid:
+        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, "symbolicOmittedInputs": 0,
+    omissions = {"pbtDroppedInvocations": 0, "pbtUnavailableInputs": 0, "pbtUnsupportedInputs": 0,
+                 "pbtPendingInvocations": 0, "symbolicOmittedInputs": 0,
                  "symbolicUncheckedInputs": 0}
     batches = []
     if concrete is not None:
+        export = concrete.get("inputExport")
         observation = concrete.get("observations")
-        omissions["pbtDroppedInvocations"] = observation.get("droppedInvocations", 0) if observation else 1
+        capture = export if export is not None else observation
+        omissions["pbtDroppedInvocations"] = capture.get("droppedInvocations", 0) if capture else 1
+        if export is not None:
+            omissions["pbtUnsupportedInputs"] = export["unsupportedInputs"]
+            omissions["pbtPendingInvocations"] = export["pendingInvocations"]
+        records = (export or {}).get("records", []) if export is not None else (observation or {}).get("invocations", [])
         invocations = []
-        for invocation in (observation or {}).get("invocations", []):
+        for invocation in records:
             if invocation["phase"] == "shrink":
                 continue
             value = invocation["input"]
@@ -184,23 +228,40 @@ def replay_inputs(artifact):
     if symbolic is not None:
         omissions["symbolicOmittedInputs"] = symbolic.get("omittedCompletedInputs", 0)
         replays = result.get("symbolicInputReplays", [])
-        omissions["symbolicUncheckedInputs"] = len(symbolic.get("completedInputs", [])) - len(replays)
+        input_key = lambda inputs: json.dumps(inputs, sort_keys=True)
+        completed_keys = {input_key(item["inputs"]) for item in symbolic.get("completedInputs", [])}
+        replay_keys = {input_key(item["candidate"]["inputs"]) for item in replays}
+        omissions["symbolicUncheckedInputs"] = len(completed_keys - replay_keys)
+        confirmed = confirmed_symbolic_input(artifact)
         invocations = []
         for replay in replays:
             native = replay["replay"]
-            matches = replay.get("predictedOutcomes") == [native["outcome"]]
-            eligible = replay["completedWithinDeadline"] and not native.get("diagnostic") and matches
+            inputs = replay["candidate"]["inputs"]
+            if confirmed is not None and inputs == confirmed["inputs"]:
+                continue
+            eligible = replay["completedWithinDeadline"] and replay.get("predictedOutcomes") == ["HOLDS"] \
+                and native["outcome"] == "HOLDS" and native_coverage_matches(replay["candidate"], native)
             if not eligible:
                 omissions["symbolicUncheckedInputs"] += 1
                 continue
             invocations.append({"inputs": replay["candidate"]["inputs"],
                                 "admission": native["admission"], "outcome": native["outcome"]})
+        if confirmed is not None:
+            invocations.append(confirmed)
+        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})
     if concrete is None and result["mode"] != "SYMBOLIC_ONLY":
         omissions["pbtUnavailableInputs"] += 1
     return batches, omissions
 
 
+def full_budget_eligible(summary):
+    quota_stopped_pbt = summary["mode"] == "PBT_ONLY" and summary.get("pbtQuotaReached") \
+        and not summary["confirmedViolation"]
+    return summary["comparisonKind"] == "equal-budget" and summary["primaryEligible"] and not quota_stopped_pbt
+
+
 def run_cell(frozen, case, seed, mode, output):
     directory = output / case["id"] / str(seed) / mode
     directory.mkdir(parents=True)
@@ -211,6 +272,7 @@ def run_cell(frozen, case, seed, mode, output):
     runner_budget = remaining_budget - frozen["classReplayReserveMillis"]
     summary = {"caseId": case["id"], "project": case["project"], "population": case["population"],
                "seed": seed, "mode": mode, "stopCriterion": STOP_CRITERION, "preparationChargedMillis": preparation_cost,
+               "comparisonKind": frozen["comparisonKind"],
                "declaredTotalBudgetMillis": frozen["totalBudgetMillis"], "pbtRunQuota": frozen["numRuns"],
                "totalStatements": case.get("totalStatements"), "universeSha256": case.get("universeSha256"),
                "coveredStatementIds": [], "coverageComplete": False, "confirmedViolation": False}
@@ -222,7 +284,7 @@ def run_cell(frozen, case, seed, mode, output):
         sources = [{"module": relative, "sha256": sha} for relative, sha in case["sourceSha256"].items()]
         configuration = {"seed": seed, "numRuns": frozen["numRuns"], "timeoutMillis": runner_budget,
                          "coverageRequest": {"admittedOnly": True},
-                         "observationRequest": {"points": [], "sources": sources,
+                         "inputExportRequest": {"sources": sources,
                                                 "maxInvocations": frozen["pbtInputLimit"],
                                                 "maxArrayElements": frozen["maxArrayElements"],
                                                 "maxBytes": frozen["maxObservationBytes"]}}
@@ -242,6 +304,7 @@ def run_cell(frozen, case, seed, mode, output):
             raise ValueError("Runner produced no completed artifact; raw process output is retained")
         artifact = read_json(directory / "campaign.json")
         result = artifact["result"]
+        summary["campaignElapsedMillis"] = result["elapsedMillis"]
         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"
@@ -251,6 +314,8 @@ def run_cell(frozen, case, seed, mode, output):
         summary["campaignDiagnostic"] = result.get("diagnostic")
         summary["concreteFailureKind"] = failure.get("kind")
         concrete = result.get("concrete") or {}
+        summary["compactCaptureMillis"] = (concrete.get("inputExport") or {}).get("captureTimeMillis", 0)
+        summary["pointCaptureMillis"] = (concrete.get("observations") or {}).get("captureTimeMillis", 0)
         summary["pbtRuns"] = concrete.get("numRuns")
         summary["pbtSkips"] = concrete.get("numSkips")
         summary["pbtQuotaReached"] = concrete.get("status") == "success" and concrete.get("numRuns") == frozen["numRuns"]
@@ -315,6 +380,8 @@ def run_cell(frozen, case, seed, mode, output):
     if summary["overrunMillis"]:
         summary["primaryEligible"] = False
         summary["coverageComplete"] = False
+    summary["nativeReplayMillis"] = sum(process["elapsedMillis"] for process in summary.get("replayProcesses", []))
+    summary["fullBudgetEligible"] = full_budget_eligible(summary)
     write_json(directory / "cell.json", summary)
     return summary
 

usvm-ts-fast-check/scripts/test_pbt_experiment.py

@@ -1,4 +1,5 @@
 import importlib.util
+import copy
 import os
 from pathlib import Path
 import struct
@@ -16,6 +17,20 @@ def number(value):
     return {"kind": "number", "value": "finite", "bits": struct.pack(">d", value).hex()}
 
 
+def symbolic_artifact(outcome="HOLDS"):
+    candidate = {"propertyId": "p", "manifestSha256": "hash", "inputs": [number(1)],
+                 "target": {"id": "predicate", "role": "PROPERTY", "expectedOutcome": outcome}}
+    native = {"admission": "ADMITTED", "outcome": outcome, "targetMatched": True,
+              "coverage": {"propertyId": "p", "admittedInvocations": 1,
+                           "identity": {"manifestSha256": "hash"},
+                           "provenance": {"request": {"admittedOnly": True}}}}
+    return {"property": {"predicate": {"exportName": "predicate"}},
+            "result": {"propertyId": "p", "mode": "SYMBOLIC_ONLY", "symbolic": {
+                "inputs": candidate["inputs"], "completedInputs": [{"inputs": candidate["inputs"]}]},
+                "symbolicInputReplays": [{"candidate": candidate, "replay": native,
+                                         "completedWithinDeadline": True, "predictedOutcomes": [outcome]}]}}
+
+
 class ExperimentTest(unittest.TestCase):
     def setUp(self):
         self.temporary = tempfile.TemporaryDirectory()
@@ -49,6 +64,94 @@ def test_frozen_runtime_changes_are_rejected_before_treatment(self):
             driver.verify_runtime(frozen)
 
 
+    def test_compact_export_takes_priority_and_preserves_all_omission_kinds(self):
+        concrete = {"inputExport": {"droppedInvocations": 3, "unsupportedInputs": 2,
+                                   "pendingInvocations": 1, "records": [
+            {"phase": "generation", "input": {"status": "captured", "value": {"kind": "array", "elements": [number(1)]}},
+             "admission": "admitted", "outcome": "holds"}]},
+                    "observations": {"droppedInvocations": 99, "invocations": []}}
+
+        batches, omissions = driver.replay_inputs({"result": {"mode": "PBT_ONLY", "concrete": concrete}})
+
+        self.assertEqual(len(batches[0]["inputs"]), 1)
+        self.assertEqual(omissions["pbtDroppedInvocations"], 3)
+        self.assertEqual(omissions["pbtUnsupportedInputs"], 2)
+        self.assertEqual(omissions["pbtPendingInvocations"], 1)
+
+
+    def test_raw_false_or_exception_does_not_bypass_original_confirmation(self):
+        for outcome in ("FALSE", "THREW"):
+            artifact = symbolic_artifact(outcome)
+
+            batches, omissions = driver.replay_inputs(artifact)
+
+            self.assertEqual(batches[0]["inputs"], [])
+            self.assertEqual(omissions["symbolicUncheckedInputs"], 1)
+
+
+    def test_healthy_symbolic_coverage_requires_the_existing_admitted_receipt_contract(self):
+        artifact = symbolic_artifact()
+        batches, omissions = driver.replay_inputs(artifact)
+        self.assertEqual(len(batches[0]["inputs"]), 1)
+        self.assertFalse(any(omissions.values()))
+
+        for change in ("late", "unadmitted", "missing-identity", "different-manifest", "diagnostic"):
+            invalid = copy.deepcopy(artifact)
+            replay = invalid["result"]["symbolicInputReplays"][0]
+            if change == "late":
+                replay["completedWithinDeadline"] = False
+            elif change == "unadmitted":
+                replay["replay"]["coverage"]["admittedInvocations"] = 0
+            elif change == "missing-identity":
+                replay["replay"]["coverage"].pop("identity")
+            elif change == "different-manifest":
+                replay["replay"]["coverage"]["identity"]["manifestSha256"] = "other"
+            else:
+                replay["replay"]["diagnostic"] = "mismatch"
+
+            batches, omissions = driver.replay_inputs(invalid)
+
+            self.assertEqual(batches[0]["inputs"], [], change)
+            self.assertEqual(omissions["symbolicUncheckedInputs"], 1, change)
+
+
+    def test_confirmed_named_fault_uses_the_original_input_and_check_receipt(self):
+        artifact = symbolic_artifact("FALSE")
+        artifact["property"]["assertions"] = [{"id": "named"}]
+        raw = artifact["result"]["symbolicInputReplays"][0]
+        candidate = copy.deepcopy(raw["candidate"])
+        candidate["target"]["id"] = "named"
+        native = copy.deepcopy(raw["replay"])
+        native["failure"] = {"checkId": "named"}
+        confirmation = {"status": "CONFIRMED_PROPERTY_VIOLATION", "candidate": candidate, "originalReplay": native}
+        artifact["result"]["confirmation"] = confirmation
+
+        batches, omissions = driver.replay_inputs(artifact)
+
+        self.assertEqual(batches[0]["inputs"], [{"inputs": candidate["inputs"], "admission": "ADMITTED", "outcome": "FALSE"}])
+        self.assertFalse(any(omissions.values()))
+
+        confirmation["originalReplay"]["failure"]["checkId"] = "different"
+
+        batches, omissions = driver.replay_inputs(artifact)
+
+        self.assertEqual(batches[0]["inputs"], [])
+        self.assertEqual(omissions["symbolicUncheckedInputs"], 1)
+
+
+    def test_functional_quota_and_early_success_are_not_full_budget_results(self):
+        row = {"comparisonKind": "functional-quota", "primaryEligible": True, "mode": "PBT_ONLY",
+               "pbtQuotaReached": True, "confirmedViolation": False}
+
+        self.assertFalse(driver.full_budget_eligible(row))
+        row["comparisonKind"] = "equal-budget"
+        self.assertFalse(driver.full_budget_eligible(row))
+        row["confirmedViolation"] = True
+        self.assertTrue(driver.full_budget_eligible(row))
+        row.update(mode="SYMBOLIC_ONLY", confirmedViolation=False)
+        self.assertTrue(driver.full_budget_eligible(row))
+
+
     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,

usvm-ts-fast-check/src/main/kotlin/org/usvm/ts/pbt/campaign/PropertyCampaignRunner.kt

@@ -287,7 +287,7 @@ class PropertyCampaignRunner(
         val request = configuration.coverageRequest ?: PropertyCoverageRequest()
         return configuration.copy(
             coverageRequest = request.copy(admittedOnly = true),
-            observationRequest = if (guided) {
+            observationRequest = if (guided && configuration.inputExportRequest == null) {
                 configuration.observationRequest ?: PropertyObservationRequest(points = emptyList(), sources = sources)
             } else {
                 configuration.observationRequest

usvm-ts-fast-check/src/test/kotlin/org/usvm/ts/pbt/campaign/GuidedPropertyCampaignTest.kt

@@ -4,6 +4,7 @@
 import org.usvm.ts.pbt.backend.CoverageScope
 import org.usvm.ts.pbt.backend.PropertyCoverageRequest
 import org.usvm.ts.pbt.backend.PropertyFailureKind
+import org.usvm.ts.pbt.backend.PropertyInputExportRequest
 import org.usvm.ts.pbt.backend.PropertyRunConfiguration
 import org.usvm.ts.pbt.backend.PropertyRunStatus
 import org.usvm.ts.pbt.backend.WitnessReductionStatus
@@ -15,6 +16,7 @@
 import org.usvm.ts.pbt.model.PropertyId
 import org.usvm.ts.pbt.model.PropertyInput
 import org.usvm.ts.pbt.model.TypeScriptEntryPoint
+import org.usvm.ts.pbt.observation.PropertyObservationSource
 import org.usvm.ts.pbt.testResourcesRoot
 import org.usvm.ts.pbt.usvm.UsvmEdgeReplayStatus
 import org.usvm.ts.pbt.usvm.UsvmPropertySearchStatus
@@ -61,6 +63,28 @@
         assertEquals(UsvmPropertySearchStatus.NO_VIOLATION_REACHED, artifact.result.symbolic?.status)
     }
 
+    @Test
+    fun `compact export supplies native inputs without allocating automatic point observations`() {
+        val configuration = configuration.copy(
+            examples = examples(1),
+            inputExportRequest = PropertyInputExportRequest(
+                sources = listOf(PropertyObservationSource(module = MODULE, sha256 = campaignSourceSha256(root.resolve(MODULE)))),
+            ),
+        )
+
+        val artifact = runner.runArtifact(
+            property = property(exportName = "branchHolds"),
+            mode = PropertyCampaignMode.COVERAGE_FEEDBACK,
+            configuration = configuration,
+            pbtAllocationMillis = 15_000,
+        )
+
+        assertEquals(1, artifact.result.concrete?.inputExport?.records?.size)
+        assertNull(artifact.result.concrete?.observations)
+        assertNotNull(artifact.result.symbolic?.edgeGuidance)
+        assertTrue(assertNotNull(artifact.result.concrete?.coverage).provenance.request.admittedOnly)
+    }
+
     @Test
     fun `saturated coverage still finds and natively confirms an original defect and saves the full trace`() {
         val artifact = run(exportName = "branchDifferent", examples = listOf(-1, 3), shrinkWitness = false)