Merge pull request #511 from Anionix/codex/issue-301-worker-rlimits

Cap bounded robustness worker resources

Author: Anionix

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.
 

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)]

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,

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

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__":