diff --git a/README.md b/README.md index 7d60d4d..287a0a9 100644 --- a/README.md +++ b/README.md @@ -76,7 +76,7 @@ uv run --frozen format-bench report --run-dir runs/fair-local The native robustness suite records pinned Arrow, Vortex, and FastLanes targets. Arrow requires a checkout at the recorded source commit plus binaries in `native/arrow/build`; Vortex and FastLanes require a checkout whose `HEAD` matches the recorded source commit. FastLanes is recorded as project-seeded rather than coverage-guided. Lance, object JSONL, and TsFile have no confirmed official native target and are retained as `UNSUPPORTED` evidence. Missing binaries or mismatched source checkouts never become a silent pass. Select targets with repeated `--target` options and set the run budget with `--duration-seconds` and `--artifact-budget-mib`. -Robustness reports also aggregate each target's case denominator, pass/fail outcomes, crash and timeout counts, incomplete reasons, duration p50, and artifact/source identities. For generated artifact mutations only, reports add deterministic descriptive counts for the denominator, completed cases, failures, crashes, timeouts, unsupported cases, incomplete cases, and completed percentage. Named boundary cases are excluded; this is not a conventional mutation score and has no ranking or gate effect. The percentage is `N/A` when no generated artifact mutations are present. The manual [`Coverage branch report`](.github/workflows/coverage.yml) layers a pinned coverage.py release over the frozen uv project environment; see the primary [coverage.py branch measurement documentation](https://coverage.readthedocs.io/en/7.10.7/branch.html) and [uv one-off dependency contract](https://docs.astral.sh/uv/reference/cli/#uv-run--with). +Robustness reports also aggregate each target's case denominator, pass/fail outcomes, crash and timeout counts, incomplete reasons, duration p50, and artifact/source identities. Before importing the heavyweight bounded-worker graph, a lightweight bootstrap lowers non-raiseable Unix limits to at most 1 GiB per file, 256 open files, and a 512-process ceiling for the worker's real user; Linux workers also receive an 8 GiB address-space cap. `RLIMIT_NPROC` is a real-user aggregate, not a per-worker child count, so the finite ceiling leaves headroom for the runner's existing processes and library threads. Root and Linux processes with exempt effective capabilities record this ceiling as `UNSUPPORTED` because the kernel does not enforce it there, per the [Linux `getrlimit(2)` contract](https://man7.org/linux/man-pages/man2/getrlimit.2.html). Tighter inherited limits win, and requested, effective, and unsupported caps are recorded in the run plus each launched case. The supported macOS runner rejects finite `RLIMIT_AS` values, so its address-space cap is explicitly `UNSUPPORTED` rather than falsely recorded as enforced; maintainers track the remaining containment risk in [#510](https://github.com/Anionix/data-format-lab/issues/510). These controls follow the primary [Python `resource` contract](https://docs.python.org/3.12/library/resource.html) and complement the existing timeout, output-retention, and total artifact-budget bounds. For generated artifact mutations only, reports add deterministic descriptive counts for the denominator, completed cases, failures, crashes, timeouts, unsupported cases, incomplete cases, and completed percentage. Named boundary cases are excluded; this is not a conventional mutation score and has no ranking or gate effect. The percentage is `N/A` when no generated artifact mutations are present. The manual [`Coverage branch report`](.github/workflows/coverage.yml) layers a pinned coverage.py release over the frozen uv project environment; see the primary [coverage.py branch measurement documentation](https://coverage.readthedocs.io/en/7.10.7/branch.html) and [uv one-off dependency contract](https://docs.astral.sh/uv/reference/cli/#uv-run--with). Coverage thresholds and source-code mutation scores are intentionally not enforced yet, so regressions can remain ungated. Repository maintainers own this accepted risk under [#506](https://github.com/Anionix/data-format-lab/issues/506); it expires after three successful reports on distinct merged `main` commits or on 2026-08-07, whichever comes first. That issue fixes the evidence fields, numeric threshold decision, serial source-mutation pilot, 20-minute budget, and closeout criteria. diff --git a/src/format_bench/report.py b/src/format_bench/report.py index 0bd0b27..f8e59c1 100644 --- a/src/format_bench/report.py +++ b/src/format_bench/report.py @@ -474,6 +474,34 @@ def _robustness(results: dict) -> list[str]: ["Case timeout seconds", config["case_timeout_seconds"]], ["Artifact budget MiB", config["artifact_budget_mib"]], ] + for label, key in ( + ("Requested", "worker_resource_limits_requested"), + ("Effective", "worker_resource_limits_effective"), + ): + worker_limits = config.get(key) + if isinstance(worker_limits, dict): + config_rows.extend( + [ + [f"{label} worker address-space cap bytes", worker_limits["address_space_bytes"]], + [f"{label} worker file-size cap bytes", worker_limits["file_size_bytes"]], + [f"{label} worker open-file cap", worker_limits["open_files"]], + [ + f"{label} real-user process ceiling", + worker_limits["real_user_processes"], + ], + ] + ) + unsupported_limits = config.get("worker_resource_limits_unsupported") + if isinstance(unsupported_limits, list): + config_rows.append( + ["Unsupported worker resource caps", ", ".join(unsupported_limits) or "None"] + ) + config_rows.append( + [ + "Worker resource cap application", + config.get("worker_resource_limits_application"), + ] + ) summary_rows = [ [verdict, evidence["summary"].get(verdict.value, 0)] diff --git a/src/format_bench/robustness/profile.py b/src/format_bench/robustness/profile.py index b0acda7..a353fcc 100644 --- a/src/format_bench/robustness/profile.py +++ b/src/format_bench/robustness/profile.py @@ -29,6 +29,11 @@ encode_malformed, encode_valid, ) +from format_bench.worker_limits import ( + DEFAULT_WORKER_RESOURCE_LIMITS, + EffectiveWorkerResourceLimits, + effective_worker_resource_limits, +) _PER_CASE_OUTPUT_BUDGET_BYTES = 1024 * 1024 @@ -97,6 +102,7 @@ def _case_result_reserve( mutation: Mapping[str, object] | None, timeout: float, output_budget_bytes: int, + worker_resource_limits_effective: EffectiveWorkerResourceLimits, ) -> int: placeholder = { "case_id": case_id, @@ -127,6 +133,12 @@ def _case_result_reserve( "input_canonical_hash": "0" * 64, "input_arrow": _record(input_record), "artifact_records": [_record(item) for item in artifact_records], + "worker_resource_limits_requested": DEFAULT_WORKER_RESOURCE_LIMITS.evidence(), + "worker_resource_limits_effective": worker_resource_limits_effective.evidence(), + "worker_resource_limits_unsupported": list( + worker_resource_limits_effective.unsupported_resources + ), + "worker_resource_limits_application": "APPLIED", **({"mutation": mutation} if mutation is not None else {}), } return len(_json(placeholder)) @@ -146,7 +158,11 @@ def _execute( mutation_index: int | None = None, mutation_count: int = 0, seed: int = 0, + worker_resource_limits_effective: EffectiveWorkerResourceLimits | None = None, ) -> dict[str, object]: + worker_resource_limits_effective = ( + worker_resource_limits_effective or effective_worker_resource_limits() + ) with ( tempfile.TemporaryDirectory() as temporary, tempfile.TemporaryDirectory(dir=run_dir) as process_directory, @@ -218,6 +234,7 @@ def _execute( mutation, timeout, _PER_CASE_OUTPUT_BUDGET_BYTES, + worker_resource_limits_effective, ) if remaining < result_reserve: raise ArtifactBudgetExceeded( @@ -305,6 +322,7 @@ def run_bounded( ] mutation_count = min(mutations_per_target, 1) if run["fixture"] else mutations_per_target store = EvidenceStore(run_dir / "robustness", artifact_budget_mib * 1024 * 1024) + worker_resource_limits_effective = effective_worker_resource_limits() observations: list[dict] = [] exhausted = False for target in targets or core_targets(): @@ -347,6 +365,7 @@ def run_bounded( mutation_index=mutation_index, mutation_count=mutation_count, seed=seed, + worker_resource_limits_effective=worker_resource_limits_effective, ) ) except ArtifactBudgetExceeded as error: @@ -370,6 +389,30 @@ def run_bounded( summary = {verdict.value: 0 for verdict in RobustnessVerdict} for item in observations: summary[item["verdict"].value] += 1 + launched = [ + item + for item in observations + if "worker_resource_limits_requested" in item + ] + applied = [ + item + for item in launched + if item.get("worker_resource_limits_application") == "APPLIED" + ] + if launched and len(applied) == len(launched): + resource_limit_application = "APPLIED" + elif applied: + resource_limit_application = "PARTIAL" + else: + resource_limit_application = "UNCONFIRMED" + effective_limit_evidence = ( + applied[0]["worker_resource_limits_effective"] + if applied + else { + key: None + for key in DEFAULT_WORKER_RESOURCE_LIMITS.evidence() + } + ) evidence = { "robustness_v1": { "contract_version": "1", @@ -385,6 +428,16 @@ def run_bounded( "effective_mutations_per_target": mutation_count, "case_timeout_seconds": timeout_seconds, "artifact_budget_mib": artifact_budget_mib, + "worker_resource_limits_requested": ( + DEFAULT_WORKER_RESOURCE_LIMITS.evidence() + ), + "worker_resource_limits_effective": ( + effective_limit_evidence + ), + "worker_resource_limits_unsupported": list( + worker_resource_limits_effective.unsupported_resources + ), + "worker_resource_limits_application": resource_limit_application, }, "cases": observations, "summary": summary, diff --git a/src/format_bench/robustness/runner.py b/src/format_bench/robustness/runner.py index 6df3412..61df4a9 100644 --- a/src/format_bench/robustness/runner.py +++ b/src/format_bench/robustness/runner.py @@ -19,6 +19,10 @@ ) from format_bench.json_contract import strict_json_dumps, strict_json_loads from format_bench.robustness.paths import reject_symlink_tree +from format_bench.worker_limits import ( + DEFAULT_WORKER_RESOURCE_LIMITS, + effective_worker_resource_limits, +) MAX_WORKER_DETAILS_BYTES = 4096 DEFAULT_OUTPUT_RETENTION_BYTES = 1024 * 1024 @@ -52,6 +56,8 @@ class RequestPayload(TypedDict): class WorkerResponse(TypedDict): observed: str details: dict[str, object] + worker_resource_limits_effective: NotRequired[dict[str, int | None]] + worker_resource_limits_unsupported: NotRequired[list[str]] class CaseResult(TypedDict): @@ -71,6 +77,10 @@ class CaseResult(TypedDict): input_arrow: NotRequired[object] artifact_records: NotRequired[object] mutation: NotRequired[object] + worker_resource_limits_requested: NotRequired[dict[str, int]] + worker_resource_limits_effective: NotRequired[dict[str, int | None]] + worker_resource_limits_unsupported: NotRequired[list[str]] + worker_resource_limits_application: NotRequired[str] def _json_object(value: object, label: str) -> dict[str, object]: @@ -107,10 +117,49 @@ def _worker_response(stdout: str) -> WorkerResponse: value = strict_json_loads(stdout.strip()) response = _json_object(value, "worker response") details: object = response.get("details", {}) - return { + parsed: WorkerResponse = { "observed": _string_field(response, "observed"), "details": _bounded_details(_json_object(details, "worker response details")), } + effective = response.get("worker_resource_limits_effective") + unsupported = response.get("worker_resource_limits_unsupported") + if effective is None and unsupported is None: + return parsed + if not isinstance(unsupported, list): + raise TypeError("worker unsupported resource limits must be strings") + unsupported_strings: list[str] = [] + for item in cast(list[object], unsupported): + if not isinstance(item, str): + raise TypeError("worker unsupported resource limits must be strings") + unsupported_strings.append(item) + effective_object = _json_object( + effective, "worker effective resource limits" + ) + fields = ( + "address_space_bytes", + "file_size_bytes", + "open_files", + "real_user_processes", + ) + if set(effective_object) != set(fields): + raise TypeError("worker effective resource limit fields do not match") + if not all( + value is None + or ( + isinstance(value, int) + and not isinstance(value, bool) + and value >= 0 + ) + for value in effective_object.values() + ): + raise TypeError("worker effective resource limits must be non-negative integers") + if not set(unsupported_strings).issubset(fields): + raise TypeError("worker unsupported resource limit is unknown") + parsed["worker_resource_limits_effective"] = cast( + dict[str, int | None], effective_object + ) + parsed["worker_resource_limits_unsupported"] = unsupported_strings + return parsed def _bounded_details(details: dict[str, object]) -> dict[str, object]: @@ -318,30 +367,30 @@ def cleanup_group() -> None: def _outcome( process: ProcessEvidence, stdout: str -) -> tuple[ObservedOutcome, dict[str, object]]: +) -> tuple[ObservedOutcome, dict[str, object], WorkerResponse | None]: if process["timed_out"]: details: dict[str, object] = ( {"cleanup_incomplete": True} if process["cleanup_incomplete"] else {} ) - return ObservedOutcome.TIMED_OUT, details + return ObservedOutcome.TIMED_OUT, details, None signal_number = process["signal"] if signal_number is not None: try: name = signal.Signals(signal_number).name except ValueError: name = None - return ObservedOutcome.CRASHED, {"signal_name": name} + return ObservedOutcome.CRASHED, {"signal_name": name}, None if process["output_exhausted"]: - return ObservedOutcome.BUDGET_EXHAUSTED, {} + return ObservedOutcome.BUDGET_EXHAUSTED, {}, None if process["exit_code"] != 0: - return ObservedOutcome.HARNESS_FAILED, {} + return ObservedOutcome.HARNESS_FAILED, {}, None try: response = _worker_response(stdout) - return ObservedOutcome(response["observed"]), response["details"] + return ObservedOutcome(response["observed"]), response["details"], response except (json.JSONDecodeError, KeyError, RecursionError, ValueError, TypeError): - return ObservedOutcome.HARNESS_FAILED, {} + return ObservedOutcome.HARNESS_FAILED, {}, None def run_case( @@ -359,17 +408,55 @@ def run_case( _relative(root, payload["manifest"]) _relative(root, payload["artifact"]) expectation = RobustnessExpectation(payload["expectation"]) - command = command or ( - sys.executable, "-m", "format_bench.robustness.worker", "--request", Path(request).as_posix() - ) + default_worker = command is None + planned_resource_limits = None + if default_worker: + planned_resource_limits = effective_worker_resource_limits() + command = ( + sys.executable, + "-m", + "format_bench.robustness_worker_bootstrap", + "--request", + Path(request).as_posix(), + ) + assert command is not None process, stdout, stderr = _process( command, root, timeout_seconds, output_budget_bytes ) stdout_path = _save(root, Path(output_dir) / "stdout.txt", stdout) stderr_path = _save(root, Path(output_dir) / "stderr.txt", stderr) - observed, details = _outcome(process, stdout) + outcome = _outcome(process, stdout) + observed = outcome[0] + details: dict[str, object] = outcome[1] + worker_response = outcome[2] + resource_limits_confirmed = False + confirmed_effective_limits: dict[str, int | None] | None = None + confirmed_unsupported_limits: list[str] | None = None + if default_worker and worker_response is not None: + assert planned_resource_limits is not None + response_effective_limits = worker_response.get( + "worker_resource_limits_effective" + ) + response_unsupported_limits = worker_response.get( + "worker_resource_limits_unsupported" + ) + resource_limits_confirmed = ( + response_effective_limits == planned_resource_limits.evidence() + and response_unsupported_limits + == list(planned_resource_limits.unsupported_resources) + ) + if resource_limits_confirmed: + assert response_effective_limits is not None + assert response_unsupported_limits is not None + confirmed_effective_limits = response_effective_limits + confirmed_unsupported_limits = response_unsupported_limits + else: + observed = ObservedOutcome.HARNESS_FAILED + details = { + "error_type": "WorkerResourceLimitEvidenceMismatch", + } verdict = robustness_verdict(expectation, observed) - return { + result: CaseResult = { "case_id": payload["case_id"], "target": payload["target"], "expectation": expectation, @@ -380,3 +467,26 @@ def run_case( "stdout": stdout_path, "stderr": stderr_path, } + if default_worker: + assert planned_resource_limits is not None + result["worker_resource_limits_requested"] = ( + DEFAULT_WORKER_RESOURCE_LIMITS.evidence() + ) + if resource_limits_confirmed: + assert confirmed_effective_limits is not None + assert confirmed_unsupported_limits is not None + result["worker_resource_limits_effective"] = confirmed_effective_limits + result["worker_resource_limits_unsupported"] = ( + confirmed_unsupported_limits + ) + result["worker_resource_limits_application"] = "APPLIED" + else: + result["worker_resource_limits_effective"] = { + key: None + for key in DEFAULT_WORKER_RESOURCE_LIMITS.evidence() + } + result["worker_resource_limits_unsupported"] = list( + planned_resource_limits.unsupported_resources + ) + result["worker_resource_limits_application"] = "UNCONFIRMED" + return result diff --git a/src/format_bench/robustness/worker.py b/src/format_bench/robustness/worker.py index 55fc97a..eda9dae 100644 --- a/src/format_bench/robustness/worker.py +++ b/src/format_bench/robustness/worker.py @@ -13,6 +13,10 @@ read_target, target_map, ) +from format_bench.worker_limits import ( + EffectiveWorkerResourceLimits, + apply_worker_resource_limits, +) # LLM contract: DISCOVERED -> ENCODED -> ROUNDTRIP_VERIFIED -> BENCHMARKED -> REPORTED. @@ -131,11 +135,17 @@ def run_request(request_path: Path) -> dict: return {"schema_version": "1", "case_id": case_id, "observed": observed, "details": details} -def main() -> None: +def main(effective_limits: EffectiveWorkerResourceLimits | None = None) -> None: parser = argparse.ArgumentParser() parser.add_argument("--request", type=Path, required=True) args = parser.parse_args() - print(strict_json_dumps(run_request(args.request), separators=(",", ":"))) + effective_limits = effective_limits or apply_worker_resource_limits() + response = run_request(args.request) + response["worker_resource_limits_effective"] = effective_limits.evidence() + response["worker_resource_limits_unsupported"] = list( + effective_limits.unsupported_resources + ) + print(strict_json_dumps(response, separators=(",", ":"))) if __name__ == "__main__": diff --git a/src/format_bench/robustness_worker_bootstrap.py b/src/format_bench/robustness_worker_bootstrap.py new file mode 100644 index 0000000..1096244 --- /dev/null +++ b/src/format_bench/robustness_worker_bootstrap.py @@ -0,0 +1,15 @@ +from __future__ import annotations + +from format_bench.worker_limits import apply_worker_resource_limits + + +def main() -> None: + effective_limits = apply_worker_resource_limits() + # Import the heavyweight adapter and robustness graph only after hard caps apply. + from format_bench.robustness.worker import main as worker_main + + worker_main(effective_limits) + + +if __name__ == "__main__": + main() diff --git a/src/format_bench/worker_limits.py b/src/format_bench/worker_limits.py new file mode 100644 index 0000000..823c308 --- /dev/null +++ b/src/format_bench/worker_limits.py @@ -0,0 +1,144 @@ +from __future__ import annotations + +import os +import resource +import sys +from dataclasses import dataclass +from pathlib import Path + +_CAP_SYS_ADMIN = 21 +_CAP_SYS_RESOURCE = 24 + + +@dataclass(frozen=True) +class WorkerResourceLimits: + address_space_bytes: int + file_size_bytes: int + open_files: int + real_user_processes: int + + def evidence(self) -> dict[str, int]: + return { + "address_space_bytes": self.address_space_bytes, + "file_size_bytes": self.file_size_bytes, + "open_files": self.open_files, + "real_user_processes": self.real_user_processes, + } + + +@dataclass(frozen=True) +class EffectiveWorkerResourceLimits: + address_space_bytes: int | None + file_size_bytes: int + open_files: int + real_user_processes: int | None + unsupported_resources: tuple[str, ...] = () + + def evidence(self) -> dict[str, int | None]: + return { + "address_space_bytes": self.address_space_bytes, + "file_size_bytes": self.file_size_bytes, + "open_files": self.open_files, + "real_user_processes": self.real_user_processes, + } + + +DEFAULT_WORKER_RESOURCE_LIMITS = WorkerResourceLimits( + address_space_bytes=8 * 1024**3, + file_size_bytes=1024**3, + open_files=256, + real_user_processes=512, +) + + +def _effective_limit(resource_id: int, requested: int) -> int: + inherited_soft, inherited_hard = resource.getrlimit(resource_id) + effective = requested + for inherited in (inherited_soft, inherited_hard): + if inherited != resource.RLIM_INFINITY: + effective = min(effective, inherited) + return effective + + +def _linux_effective_capabilities() -> int | None: + try: + status = Path("/proc/self/status").read_text(encoding="utf-8") + except OSError: + return None + for line in status.splitlines(): + name, separator, value = line.partition(":") + if name == "CapEff" and separator: + try: + return int(value.strip(), 16) + except ValueError: + return None + return None + + +def _real_user_process_limit_supported() -> bool: + # Linux getrlimit contract: + # https://man7.org/linux/man-pages/man2/getrlimit.2.html + if os.getuid() == 0: + return False + if sys.platform != "linux": + return True + capabilities = _linux_effective_capabilities() + exempt = (1 << _CAP_SYS_ADMIN) | (1 << _CAP_SYS_RESOURCE) + return capabilities is not None and capabilities & exempt == 0 + + +def effective_worker_resource_limits( + limits: WorkerResourceLimits = DEFAULT_WORKER_RESOURCE_LIMITS, +) -> EffectiveWorkerResourceLimits: + """Derive caps without relaxing tighter inherited soft or hard limits.""" + + # Python resource contract: https://docs.python.org/3.12/library/resource.html + address_space_supported = sys.platform != "darwin" + process_limit_supported = _real_user_process_limit_supported() + unsupported_resources = [] + if not address_space_supported: + unsupported_resources.append("address_space_bytes") + if not process_limit_supported: + unsupported_resources.append("real_user_processes") + return EffectiveWorkerResourceLimits( + address_space_bytes=( + _effective_limit(resource.RLIMIT_AS, limits.address_space_bytes) + if address_space_supported + else None + ), + file_size_bytes=_effective_limit( + resource.RLIMIT_FSIZE, limits.file_size_bytes + ), + open_files=_effective_limit(resource.RLIMIT_NOFILE, limits.open_files), + real_user_processes=( + _effective_limit(resource.RLIMIT_NPROC, limits.real_user_processes) + if process_limit_supported + else None + ), + unsupported_resources=tuple(unsupported_resources), + ) + + +def apply_worker_resource_limits( + limits: WorkerResourceLimits = DEFAULT_WORKER_RESOURCE_LIMITS, +) -> EffectiveWorkerResourceLimits: + """Install non-raiseable caps and return their effective values.""" + + effective = effective_worker_resource_limits(limits) + resource_limits = [ + (resource.RLIMIT_FSIZE, effective.file_size_bytes), + (resource.RLIMIT_NOFILE, effective.open_files), + ] + if effective.address_space_bytes is not None: + resource_limits.insert( + 0, (resource.RLIMIT_AS, effective.address_space_bytes) + ) + if effective.real_user_processes is not None: + resource_limits.append( + (resource.RLIMIT_NPROC, effective.real_user_processes) + ) + for resource_id, value in resource_limits: + resource.setrlimit(resource_id, (value, value)) + # LLM contract: limit setup failure becomes HARNESS_FAILED, incomplete, + # non-ranking evidence; successful setup never advances lifecycle state. + return effective diff --git a/tests/test_robustness_limits.py b/tests/test_robustness_limits.py new file mode 100644 index 0000000..fa25f2d --- /dev/null +++ b/tests/test_robustness_limits.py @@ -0,0 +1,143 @@ +import resource + +import pytest + +import format_bench.worker_limits as worker_limits +from format_bench.worker_limits import ( + EffectiveWorkerResourceLimits, + WorkerResourceLimits, + apply_worker_resource_limits, +) + + +def test_worker_limits_lower_hard_caps_and_preserve_tighter_inheritance( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(worker_limits.sys, "platform", "linux") + monkeypatch.setattr( + worker_limits, "_real_user_process_limit_supported", lambda: True + ) + requested = WorkerResourceLimits( + address_space_bytes=800, + file_size_bytes=400, + open_files=200, + real_user_processes=20, + ) + inherited = { + resource.RLIMIT_AS: (resource.RLIM_INFINITY, resource.RLIM_INFINITY), + resource.RLIMIT_FSIZE: (resource.RLIM_INFINITY, 300), + resource.RLIMIT_NOFILE: (50, 100), + resource.RLIMIT_NPROC: (resource.RLIM_INFINITY, 10), + } + applied: dict[int, tuple[int, int]] = {} + monkeypatch.setattr( + resource, + "getrlimit", + lambda resource_id: inherited[resource_id], + ) + monkeypatch.setattr( + resource, + "setrlimit", + lambda resource_id, limits: applied.__setitem__(resource_id, limits), + ) + + effective = apply_worker_resource_limits(requested) + + assert effective == EffectiveWorkerResourceLimits(800, 300, 50, 10) + assert applied == { + resource.RLIMIT_AS: (800, 800), + resource.RLIMIT_FSIZE: (300, 300), + resource.RLIMIT_NOFILE: (50, 50), + resource.RLIMIT_NPROC: (10, 10), + } + + +def test_worker_limits_record_darwin_address_space_as_unsupported( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(worker_limits.sys, "platform", "darwin") + monkeypatch.setattr( + worker_limits, "_real_user_process_limit_supported", lambda: True + ) + requested = WorkerResourceLimits(800, 400, 200, 20) + inherited = { + resource.RLIMIT_FSIZE: (resource.RLIM_INFINITY, 300), + resource.RLIMIT_NOFILE: (50, 100), + resource.RLIMIT_NPROC: (resource.RLIM_INFINITY, 10), + } + applied: dict[int, tuple[int, int]] = {} + monkeypatch.setattr( + resource, + "getrlimit", + lambda resource_id: inherited[resource_id], + ) + monkeypatch.setattr( + resource, + "setrlimit", + lambda resource_id, limits: applied.__setitem__(resource_id, limits), + ) + + effective = apply_worker_resource_limits(requested) + + assert effective == EffectiveWorkerResourceLimits( + None, 300, 50, 10, ("address_space_bytes",) + ) + assert resource.RLIMIT_AS not in applied + assert applied == { + resource.RLIMIT_FSIZE: (300, 300), + resource.RLIMIT_NOFILE: (50, 50), + resource.RLIMIT_NPROC: (10, 10), + } + + +def test_worker_limits_do_not_claim_privileged_nproc_enforcement( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(worker_limits.sys, "platform", "linux") + monkeypatch.setattr( + worker_limits, "_real_user_process_limit_supported", lambda: False + ) + requested = WorkerResourceLimits(800, 400, 200, 20) + inherited = { + resource.RLIMIT_AS: (resource.RLIM_INFINITY, resource.RLIM_INFINITY), + resource.RLIMIT_FSIZE: (resource.RLIM_INFINITY, 300), + resource.RLIMIT_NOFILE: (50, 100), + } + applied: dict[int, tuple[int, int]] = {} + monkeypatch.setattr( + resource, + "getrlimit", + lambda resource_id: inherited[resource_id], + ) + monkeypatch.setattr( + resource, + "setrlimit", + lambda resource_id, limits: applied.__setitem__(resource_id, limits), + ) + + effective = apply_worker_resource_limits(requested) + + assert effective == EffectiveWorkerResourceLimits( + 800, 300, 50, None, ("real_user_processes",) + ) + assert resource.RLIMIT_NPROC not in applied + + +def test_privileged_linux_context_does_not_support_nproc_enforcement( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(worker_limits.sys, "platform", "linux") + monkeypatch.setattr(worker_limits.os, "getuid", lambda: 501) + monkeypatch.setattr( + worker_limits, + "_linux_effective_capabilities", + lambda: 1 << worker_limits._CAP_SYS_RESOURCE, + ) + + assert worker_limits._real_user_process_limit_supported() is False + + monkeypatch.setattr(worker_limits.os, "getuid", lambda: 0) + monkeypatch.setattr( + worker_limits, "_linux_effective_capabilities", lambda: 0 + ) + assert worker_limits._real_user_process_limit_supported() is False diff --git a/tests/test_robustness_profile.py b/tests/test_robustness_profile.py index 68ab4f2..2820155 100644 --- a/tests/test_robustness_profile.py +++ b/tests/test_robustness_profile.py @@ -1,4 +1,5 @@ import json +import sys from pathlib import Path import pytest @@ -130,10 +131,55 @@ def test_public_cli_runs_and_reports_bounded_fixture( assert evidence["config"]["mutations_per_target"] == 1 assert evidence["config"]["case_timeout_seconds"] == 5 assert evidence["config"]["artifact_budget_mib"] == 64 + requested_limits = { + "address_space_bytes": 8 * 1024**3, + "file_size_bytes": 1024**3, + "open_files": 256, + "real_user_processes": 512, + } + assert evidence["config"]["worker_resource_limits_requested"] == requested_limits + effective_limits = evidence["config"]["worker_resource_limits_effective"] + unsupported_limits = evidence["config"][ + "worker_resource_limits_unsupported" + ] + assert set(unsupported_limits) <= { + "address_space_bytes", + "real_user_processes", + } + if sys.platform == "darwin": + assert "address_space_bytes" in unsupported_limits + for name, requested in requested_limits.items(): + effective = effective_limits[name] + if name in unsupported_limits: + assert effective is None + else: + assert isinstance(effective, int) + assert effective <= requested + assert evidence["config"]["worker_resource_limits_application"] == "APPLIED" + assert all( + item["worker_resource_limits_requested"] == requested_limits + and item["worker_resource_limits_effective"] == effective_limits + and item["worker_resource_limits_unsupported"] == unsupported_limits + and item["worker_resource_limits_application"] == "APPLIED" + for item in evidence["cases"] + ) valid = next(item for item in evidence["cases"] if "input_arrow" in item) + worker_response = json.loads((run_dir / valid["stdout"]).read_text()) + assert worker_response["worker_resource_limits_effective"] == effective_limits + assert ( + worker_response["worker_resource_limits_unsupported"] + == unsupported_limits + ) assert not Path(valid["input_arrow"]["path"]).is_absolute() cli.main(["report", "--run-dir", str(run_dir)]) first = (run_dir / "report.md").read_text() + assert "| Requested worker address-space cap bytes | 8589934592 |" in first + assert ( + f"| Unsupported worker resource caps | " + f"{', '.join(unsupported_limits) or 'None'} |" + ) in first + assert "| Effective real-user process ceiling |" in first + assert "| Worker resource cap application | APPLIED |" in first cli.main(["report", "--run-dir", str(run_dir)]) assert (run_dir / "report.md").read_text() == first diff --git a/tests/test_robustness_runner.py b/tests/test_robustness_runner.py index 957a3e7..fecf544 100644 --- a/tests/test_robustness_runner.py +++ b/tests/test_robustness_runner.py @@ -7,6 +7,7 @@ import pytest +import format_bench.robustness.runner as robustness_runner from format_bench.model import ObservedOutcome, RobustnessVerdict from format_bench.robustness.runner import ( CaseResult, @@ -168,6 +169,62 @@ def test_runner_classifies_invalid_output_and_valid_roundtrip_failure(tmp_path: assert valid["verdict"] is RobustnessVerdict.FAIL +def test_default_worker_does_not_claim_unconfirmed_resource_limits( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + _request(tmp_path) + process = { + "exit_code": 1, + "signal": None, + "timed_out": False, + "duration_ms": 1.0, + "stdout_bytes": 0, + "stderr_bytes": 16, + "stdout_truncated": False, + "stderr_truncated": False, + "output_budget_bytes": 1024, + "output_exhausted": False, + "cleanup_incomplete": False, + } + monkeypatch.setattr( + robustness_runner, + "_process", + lambda *_args, **_kwargs: (process, "", "setrlimit failed"), + ) + + result = run_case(tmp_path, "request.json", "evidence/case-1") + + assert result["observed"] is ObservedOutcome.HARNESS_FAILED + assert result["worker_resource_limits_application"] == "UNCONFIRMED" + assert set(result["worker_resource_limits_effective"].values()) == {None} + + process["exit_code"] = 0 + contradictory = json.dumps( + { + "observed": "ACCEPTED", + "details": {}, + "worker_resource_limits_effective": { + "address_space_bytes": None, + "file_size_bytes": 0, + "open_files": 999_999, + "real_user_processes": 0, + }, + "worker_resource_limits_unsupported": [], + } + ) + monkeypatch.setattr( + robustness_runner, + "_process", + lambda *_args, **_kwargs: (process, contradictory, ""), + ) + + mismatch = run_case(tmp_path, "request.json", "evidence/case-2") + + assert mismatch["observed"] is ObservedOutcome.HARNESS_FAILED + assert mismatch["worker_resource_limits_application"] == "UNCONFIRMED" + assert set(mismatch["worker_resource_limits_effective"].values()) == {None} + + def test_runner_bounds_deep_worker_details_without_escaping(tmp_path: Path) -> None: code = ( "depth=1000; "