Harden robustness boundary evidence

Author: Anionix

src/format_bench/robustness/cases.py

@@ -74,7 +74,8 @@ def generated_cases(seed: int, count: int) -> tuple[CaseSpec, ...]:
             "rows": rng.randint(0, 2049),
             "string_seed": rng.getrandbits(32),
         }
-        parameters["cardinality"] = rng.randint(1, max(parameters["rows"], 1))
+        rows = parameters["rows"]
+        parameters["cardinality"] = 0 if rows == 0 else rng.randint(1, rows)
         encoded = json.dumps(parameters, sort_keys=True, separators=(",", ":")).encode()
         suffix = hashlib.sha256(encoded).hexdigest()[:10]
         cases.append(

src/format_bench/robustness/native.py

@@ -210,6 +210,12 @@ def _corpus_source(run_dir: Path, run: dict, target: NativeTarget) -> Path:
     return _safe_run_file(run_root, run["input"]["source"], "native corpus")
 
 
+def _corpus_directory(run_dir: Path, run: dict, target: NativeTarget) -> Path:
+    # Resolve the seed before _process changes the child's cwd. libFuzzer treats
+    # a directory argument as a fuzzing corpus and a file argument as replay.
+    return _corpus_source(run_dir, run, target).parent
+
+
 def _safe_run_file(run_root: Path, value: object, label: str) -> Path:
     if not isinstance(value, str):
         raise ValueError(f"{label} path must be a string")
@@ -303,7 +309,11 @@ def _case(
             },
         }
     try:
-        source = _corpus_source(run_dir, run, target) if target.engine == "coverage-guided" and target.binary_name else None
+        corpus = (
+            _corpus_directory(run_dir, run, target)
+            if target.engine == "coverage-guided" and target.binary_name
+            else None
+        )
     except ValueError as error:
         return {
             **base,
@@ -322,11 +332,11 @@ def _case(
         if target.engine == "coverage-guided" and target.binary_name is None:
             command = ["cargo", "fuzz", "run", target.official_target, "--", *limits]
         elif target.engine == "coverage-guided":
-            assert source is not None
+            assert corpus is not None
             command = [
                 str(binary), "-print_final_stats=1",
                 f"-timeout={_fuzz_seconds(duration_seconds)}",
-                *limits, source.as_posix(),
+                *limits,
             ]
         else:
             command = [
@@ -337,6 +347,8 @@ def _case(
             ]
         if target.engine == "coverage-guided" and fixture:
             command.append("-runs=1")
+        if corpus is not None:
+            command.append(corpus.as_posix())
         process, stdout, stderr = _process(
             command, work_dir, duration_seconds + 5, output_budget_bytes=_NATIVE_OUTPUT_BUDGET_BYTES
         )
@@ -357,8 +369,8 @@ def _case(
         }
         if binary is not None:
             details["binary_sha256"] = _sha256(binary)
-        if source is not None:
-            details["corpus"] = source.relative_to(run_root).as_posix()
+        if corpus is not None:
+            details["corpus"] = corpus.relative_to(run_root).as_posix()
         result = {
             **base,
             "observed": observed,

src/format_bench/robustness/runner.py

@@ -113,7 +113,13 @@ def _worker_response(stdout: str) -> WorkerResponse:
 
 
 def _bounded_details(details: dict[str, object]) -> dict[str, object]:
-    encoded = (json.dumps(details, indent=2, sort_keys=True) + "\n").encode()
+    try:
+        encoded = (json.dumps(details, indent=2, sort_keys=True) + "\n").encode()
+    except (RecursionError, TypeError, ValueError) as error:
+        return {
+            "truncated": True,
+            "error_type": type(error).__name__,
+        }
     if len(encoded) <= MAX_WORKER_DETAILS_BYTES:
         return details
     return {
@@ -333,7 +339,7 @@ def _outcome(
     try:
         response = _worker_response(stdout)
         return ObservedOutcome(response["observed"]), response["details"]
-    except (json.JSONDecodeError, KeyError, ValueError, TypeError):
+    except (json.JSONDecodeError, KeyError, RecursionError, ValueError, TypeError):
         return ObservedOutcome.HARNESS_FAILED, {}
 
 

src/format_bench/robustness/worker.py

@@ -70,8 +70,14 @@ def run_request(request_path: Path) -> dict:
                     "observed": ObservedOutcome.UNSUPPORTED,
                     "details": {"error_type": type(error).__name__},
                 }
-            except Exception as error:
+            except ValueError as error:
                 observed = ObservedOutcome.VALUE_MISMATCH
+                details = {
+                    "error_type": type(error).__name__,
+                    "message": str(error)[-500:],
+                }
+            except Exception as error:
+                observed = ObservedOutcome.HARNESS_FAILED
                 details = {"error_type": type(error).__name__, "message": str(error)[-500:]}
         else:
             effective_manifest = json.loads(manifest.read_text(encoding="utf-8"))

tests/test_robustness_cases.py

@@ -57,6 +57,8 @@ def test_generated_cases_preserve_requested_rows() -> None:
         assert table.num_rows == case.options["rows"]
         if case.options["rows"]:
             assert case.options["cardinality"] <= case.options["rows"]
+        else:
+            assert case.options["cardinality"] == 0
 
 
 def test_malformed_case_has_no_valid_arrow_materialization() -> None:

tests/test_robustness_native.py

@@ -87,6 +87,41 @@ def test_native_suite_records_arrow_target_and_process_evidence(
     assert not Path(case["stdout"]).is_absolute()
 
 
+def test_native_suite_passes_absolute_directory_corpus_for_relative_run_dir(
+    tmp_path: Path, monkeypatch
+) -> None:
+    root = Path(__file__).parents[1]
+    monkeypatch.chdir(root)
+    run_dir = Path(os.path.relpath(tmp_path / "run", root))
+    build_dir = tmp_path / "build"
+    _verified_fixture(root, run_dir)
+    _fake_target(build_dir, "arrow-csv-fuzz", "print('native smoke')")
+    commands: list[list[str]] = []
+
+    def fake_process(command, cwd, timeout, output_budget_bytes):
+        commands.append(list(command))
+        return {"timed_out": False, "signal": None, "exit_code": 0}, "", ""
+
+    monkeypatch.setattr(native, "_process", fake_process)
+    result_path = run_native(
+        root,
+        run_dir,
+        duration_seconds=1,
+        artifact_budget_mib=4,
+        targets=(replace(ARROW_NATIVE_TARGETS[0], source_commit=None),),
+        build_dir=build_dir,
+    )
+
+    case = json.loads(result_path.read_text())["results"]["robustness_v1"]["cases"][0]
+    corpus = Path(commands[0][-1])
+    assert corpus.is_absolute()
+    assert corpus.is_dir()
+    assert (corpus / "source.csv").is_file()
+    assert case["details"]["corpus"] == "input"
+    assert case["observed"] == ObservedOutcome.ACCEPTED.value
+    assert case["verdict"] == RobustnessVerdict.PASS.value
+
+
 def test_native_suite_keeps_missing_target_as_unsupported(tmp_path: Path) -> None:
     root = Path(__file__).parents[1]
     run_dir = tmp_path / "run"

tests/test_robustness_runner.py

@@ -56,6 +56,14 @@ def test_runner_classifies_signal_and_timeout(tmp_path: Path) -> None:
     assert timed_out["process"]["signal"] is None
 
 
+def test_runner_classifies_timeout_after_bytes_output(tmp_path: Path) -> None:
+    code = "import os,time; os.write(1,b'worker-bytes\\n'); time.sleep(5)"
+    timed_out = _run(tmp_path, code)
+    assert timed_out["observed"] is ObservedOutcome.TIMED_OUT
+    assert timed_out["process"]["signal"] is None
+    assert (tmp_path / timed_out["stdout"]).read_bytes() == b"worker-bytes\n"
+
+
 def test_process_reaps_descendant_after_leader_exit(tmp_path: Path) -> None:
     pid_path = tmp_path / "descendant.pid"
     ready_path = tmp_path / "descendant.ready"
@@ -152,6 +160,22 @@ def test_runner_classifies_invalid_output_and_valid_roundtrip_failure(tmp_path:
     assert valid["verdict"] is RobustnessVerdict.FAIL
 
 
+def test_runner_bounds_deep_worker_details_without_escaping(tmp_path: Path) -> None:
+    code = (
+        "depth=1000; "
+        "payload='{" + '\"observed\":\"REJECTED\",\"details\":' + "' "
+        "+ '{\"nested\":' * depth + '0' + '}' * depth + '}'"
+        "; print(payload)"
+    )
+    result = _run(tmp_path, code)
+    assert result["observed"] is ObservedOutcome.REJECTED
+    assert result["verdict"] is RobustnessVerdict.PASS
+    assert result["details"] == {
+        "truncated": True,
+        "error_type": "RecursionError",
+    }
+
+
 def test_runner_reuses_unused_stream_budget(tmp_path: Path) -> None:
     code = "import json; print(json.dumps({'observed':'REJECTED','details':{'message':'x'*200}}))"
     result = _run(tmp_path, code, output_budget_bytes=512)

tests/test_robustness_targets.py

@@ -152,7 +152,7 @@ def test_worker_reports_raised_builtin_verification_as_value_mismatch(
     assert result["details"]["error_type"] == "ValueError"
 
 
-def test_worker_reports_non_value_verification_errors_as_value_mismatch(
+def test_worker_reports_non_value_verification_errors_as_harness_failure(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:
     class BrokenVerifier:
@@ -176,5 +176,5 @@ def verify_roundtrip(self, path, manifest):
 
     result = run_request(Path("request.json"))
 
-    assert result["observed"] is ObservedOutcome.VALUE_MISMATCH
+    assert result["observed"] is ObservedOutcome.HARNESS_FAILED
     assert result["details"]["error_type"] == "TypeError"