fix: harden benchmark output reservation

Problem: Sidecar locks did not own the final report paths, argparse still echoed malformed value details, and missing or extra runner results could be reported as successful.

Risk: Final-path placeholders and inode-checked cleanup change concurrent publication behavior; collision, external replacement, temp failure, thread/process stress, and cardinality regressions verify that external files are preserved and incomplete runs exit 2.

src/agent_lab/infrastructure/benchmark_reporting.py

@@ -105,8 +105,8 @@ class BenchmarkReportWriteError(RuntimeError):
 class _ReportReservation:
     json_path: Path
     markdown_path: Path
-    lock_path: Path
-    lock_fd: int
+    json_identity: tuple[int, int]
+    markdown_identity: tuple[int, int]
 
 
 def percentile(values: Sequence[int | float], quantile: float) -> float | None:
@@ -510,9 +510,7 @@ def write_benchmark_report(
     timestamp: str | None = None,
 ) -> tuple[Path, Path]:
     directory = Path(output_dir)
-    temp_paths: list[Path] = []
-    published_paths: list[Path] = []
-    reservation: _ReportReservation | None = None
+    owned_paths: dict[Path, tuple[int, int]] = {}
     try:
         directory.mkdir(parents=True, exist_ok=True)
         stem = timestamp or report.generated_at.astimezone(timezone.utc).strftime(
@@ -521,36 +519,55 @@ def write_benchmark_report(
         reservation = _reserve_report_paths(directory, stem)
         json_path = reservation.json_path
         markdown_path = reservation.markdown_path
+        owned_paths[json_path] = reservation.json_identity
+        owned_paths[markdown_path] = reservation.markdown_identity
         nonce = uuid4().hex
         json_temp = directory / f".{json_path.name}.{nonce}.tmp"
         markdown_temp = directory / f".{markdown_path.name}.{nonce}.tmp"
-        temp_paths.extend([json_temp, markdown_temp])
-        _write_text(json_temp, report_to_json(report))
-        _write_text(markdown_temp, render_markdown(report))
-        os.replace(json_temp, json_path)
-        published_paths.append(json_path)
-        os.replace(markdown_temp, markdown_path)
-        published_paths.append(markdown_path)
+        json_temp_identity = _write_text(json_temp, report_to_json(report))
+        owned_paths[json_temp] = json_temp_identity
+        markdown_temp_identity = _write_text(
+            markdown_temp,
+            render_markdown(report),
+        )
+        owned_paths[markdown_temp] = markdown_temp_identity
+        _replace_owned_placeholder(
+            json_temp,
+            json_path,
+            reservation.json_identity,
+        )
+        owned_paths.pop(json_temp)
+        owned_paths[json_path] = json_temp_identity
+        _replace_owned_placeholder(
+            markdown_temp,
+            markdown_path,
+            reservation.markdown_identity,
+        )
+        owned_paths.pop(markdown_temp)
+        owned_paths[markdown_path] = markdown_temp_identity
         return json_path, markdown_path
     except Exception as exc:
-        for path in temp_paths + published_paths:
-            try:
-                path.unlink(missing_ok=True)
-            except OSError:
-                pass
+        for path, identity in owned_paths.items():
+            _unlink_if_owned(path, identity)
         raise BenchmarkReportWriteError(
             "failed to write benchmark reports"
         ) from exc
-    finally:
-        if reservation is not None:
-            _release_reservation(reservation)
 
 
-def _write_text(path: Path, content: str) -> None:
-    with path.open("x", encoding="utf-8") as handle:
-        handle.write(content)
-        handle.flush()
-        os.fsync(handle.fileno())
+def _write_text(path: Path, content: str) -> tuple[int, int]:
+    identity: tuple[int, int] | None = None
+    try:
+        with path.open("x", encoding="utf-8") as handle:
+            file_stat = os.fstat(handle.fileno())
+            identity = file_stat.st_dev, file_stat.st_ino
+            handle.write(content)
+            handle.flush()
+            os.fsync(handle.fileno())
+        return identity
+    except BaseException:
+        if identity is not None:
+            _unlink_if_owned(path, identity)
+        raise
 
 
 def _reserve_report_paths(directory: Path, stem: str) -> _ReportReservation:
@@ -559,49 +576,59 @@ def _reserve_report_paths(directory: Path, stem: str) -> _ReportReservation:
         candidate = stem if suffix == 0 else f"{stem}-{suffix}"
         json_path = directory / f"{candidate}.json"
         markdown_path = directory / f"{candidate}.md"
-        lock_path = directory / f".{candidate}.lock"
         try:
-            lock_fd = os.open(
-                lock_path,
-                os.O_CREAT | os.O_EXCL | os.O_WRONLY,
-                0o600,
-            )
+            json_identity = _create_placeholder(json_path)
         except FileExistsError:
             suffix += 1
             continue
-        reservation = _ReportReservation(
-            json_path=json_path,
-            markdown_path=markdown_path,
-            lock_path=lock_path,
-            lock_fd=lock_fd,
-        )
         try:
-            paths_are_available = (
-                not json_path.exists() and not markdown_path.exists()
-            )
+            markdown_identity = _create_placeholder(markdown_path)
+        except FileExistsError:
+            _unlink_if_owned(json_path, json_identity)
+            suffix += 1
+            continue
         except BaseException:
-            _release_reservation(reservation)
+            _unlink_if_owned(json_path, json_identity)
             raise
-        if paths_are_available:
-            return reservation
-        _release_reservation(reservation)
-        suffix += 1
+        return _ReportReservation(
+            json_path=json_path,
+            markdown_path=markdown_path,
+            json_identity=json_identity,
+            markdown_identity=markdown_identity,
+        )
 
 
-def _release_reservation(reservation: _ReportReservation) -> None:
+def _create_placeholder(path: Path) -> tuple[int, int]:
+    file_descriptor = os.open(
+        path,
+        os.O_CREAT | os.O_EXCL | os.O_WRONLY,
+        0o600,
+    )
     try:
-        try:
-            path_stat = reservation.lock_path.stat()
-            fd_stat = os.fstat(reservation.lock_fd)
-        except OSError:
-            return
-        if (path_stat.st_dev, path_stat.st_ino) == (
-            fd_stat.st_dev,
-            fd_stat.st_ino,
-        ):
-            reservation.lock_path.unlink(missing_ok=True)
+        file_stat = os.fstat(file_descriptor)
+        return file_stat.st_dev, file_stat.st_ino
     finally:
-        try:
-            os.close(reservation.lock_fd)
-        except OSError:
-            pass
+        os.close(file_descriptor)
+
+
+def _path_identity(path: Path) -> tuple[int, int]:
+    path_stat = path.stat()
+    return path_stat.st_dev, path_stat.st_ino
+
+
+def _replace_owned_placeholder(
+    source: Path,
+    destination: Path,
+    placeholder_identity: tuple[int, int],
+) -> None:
+    if _path_identity(destination) != placeholder_identity:
+        raise FileExistsError("reserved report path ownership changed")
+    os.replace(source, destination)
+
+
+def _unlink_if_owned(path: Path, identity: tuple[int, int]) -> None:
+    try:
+        if _path_identity(path) == identity:
+            path.unlink(missing_ok=True)
+    except OSError:
+        pass

src/agent_lab/presentation/benchmark_cli.py

@@ -32,8 +32,12 @@ DEFAULT_OUTPUT_DIR = Path("outputs/benchmarks")
 
 class _RedactingArgumentParser(argparse.ArgumentParser):
     def error(self, message: str) -> NoReturn:
+        del message
         self.print_usage(sys.stderr)
-        self.exit(2, f"{self.prog}: error: {redact_text(message)}\n")
+        self.exit(
+            2,
+            f"{self.prog}: error: invalid command-line arguments\n",
+        )
 
 
 @dataclass(frozen=True)
@@ -126,6 +130,19 @@ def run_cli(
     except Exception as exc:
         runner_error = exc
 
+    execution_error = runner_error
+    if execution_error is None:
+        expected_result_count = (
+            len(config.cases)
+            * len(config.modes)
+            * config.runs_per_case
+        )
+        if len(results) != expected_result_count:
+            execution_error = RuntimeError(
+                "runner result count mismatch: "
+                f"expected {expected_result_count}, got {len(results)}"
+            )
+
     try:
         report = deps.report_builder(
             config,
@@ -134,8 +151,8 @@ def run_cli(
             generated_at=deps.now(),
             secrets=(api_key,),
             system_errors=(
-                (str(runner_error),)
-                if runner_error is not None
+                (str(execution_error),)
+                if execution_error is not None
                 else ()
             ),
         )
@@ -156,8 +173,8 @@ def run_cli(
         deps.stdout,
         secrets=(api_key,),
     )
-    if runner_error is not None:
-        _print_error(runner_error, deps.stderr, secrets=(api_key,))
+    if execution_error is not None:
+        _print_error(execution_error, deps.stderr, secrets=(api_key,))
         return 2
     return 1 if report.overall.failed else 0
 

tests/test_benchmark_cli.py

@@ -122,9 +122,15 @@ def test_cli_overrides_json_and_repeatable_case_mode_replace_lists(tmp_path: Pat
         cli,
         results=[
             _run_result(
-                case_id=BenchmarkCaseId.SESSION_TERMINATE,
+                case_id=case_id,
                 mode=BenchmarkMode.CHAT_AGENT_TOOLS,
+                iteration=iteration,
             )
+            for case_id in (
+                BenchmarkCaseId.ORDINARY_CHAT,
+                BenchmarkCaseId.SESSION_TERMINATE,
+            )
+            for iteration in range(1, 4)
         ],
         api_key="env-key",
         runner_calls=runner_calls,
@@ -205,7 +211,10 @@ def test_mock_run_needs_no_api_key_and_exits_zero(tmp_path: Path):
 def test_partial_failed_run_exits_one_writes_redacted_reports(tmp_path: Path):
     cli = _cli_module()
     secret = "sk-cli-real-secret"
-    config_path = _write_config(tmp_path / "config.json")
+    config_path = _write_config(
+        tmp_path / "config.json",
+        runs_per_case=2,
+    )
     output_dir = tmp_path / "reports"
     stdout = StringIO()
     stderr = StringIO()
@@ -415,12 +424,72 @@ def test_unexpected_runner_failure_still_attempts_report_write(tmp_path: Path):
     assert writes[0].overall.attempted == 1
     assert writes[0].overall.failed == 1
     assert writes[0].overall.system_errors == ["Bearer [REDACTED]"]
+    assert "result count mismatch" not in writes[0].overall.system_errors[0]
     assert "attempted=1 passed=0 failed=1" in stdout.getvalue()
     assert secret not in stdout.getvalue()
     assert secret not in stderr.getvalue()
     assert "Bearer [REDACTED]" in stderr.getvalue()
 
 
+@pytest.mark.parametrize(
+    ("results", "attempted", "passed", "failed", "actual_count"),
+    [
+        ([], 1, 0, 1, 0),
+        ([_run_result(iteration=1), _run_result(iteration=2)], 3, 2, 1, 2),
+    ],
+)
+def test_runner_result_cardinality_mismatch_is_reported_and_exits_two(
+    tmp_path: Path,
+    results: list[BenchmarkRunResult],
+    attempted: int,
+    passed: int,
+    failed: int,
+    actual_count: int,
+):
+    cli = _cli_module()
+    config_path = _write_config(tmp_path / "config.json")
+    output_dir = tmp_path / "reports"
+    stdout = StringIO()
+    stderr = StringIO()
+    dependencies = _dependencies(
+        cli,
+        results=results,
+        stdout=stdout,
+        stderr=stderr,
+    )
+
+    exit_code = cli.run_cli(
+        [
+            "--config",
+            str(config_path),
+            "--mock",
+            "--output-dir",
+            str(output_dir),
+        ],
+        dependencies=dependencies,
+    )
+
+    assert exit_code == 2
+    json_path = next(output_dir.glob("*.json"))
+    markdown_path = next(output_dir.glob("*.md"))
+    payload = json.loads(json_path.read_text())
+    expected_error = (
+        f"runner result count mismatch: expected 1, got {actual_count}"
+    )
+    assert payload["overall"] == {
+        "attempted": attempted,
+        "passed": passed,
+        "failed": failed,
+        "warnings": payload["overall"]["warnings"],
+        "system_errors": [expected_error],
+    }
+    assert expected_error in markdown_path.read_text()
+    assert f"attempted={attempted} passed={passed} failed={failed}" in (
+        stdout.getvalue()
+    )
+    assert expected_error in stderr.getvalue()
+
+
 def test_help_and_required_config_use_argparse(capsys: pytest.CaptureFixture[str]):
     cli = _cli_module()
 
@@ -434,24 +503,37 @@ def test_help_and_required_config_use_argparse(capsys: pytest.CaptureFixture[str
     assert missing_exit.value.code == 2
 
 
-def test_argparse_errors_are_redacted(capsys: pytest.CaptureFixture[str]):
+@pytest.mark.parametrize(
+    "invalid_args",
+    [
+        ["--runs", "api_key=runs-secret"],
+        ["--case", "api_key=case-secret"],
+    ],
+)
+def test_argparse_errors_use_only_generic_safe_message(
+    capsys: pytest.CaptureFixture[str],
+    invalid_args: list[str],
+):
     cli = _cli_module()
-    secret = "argument-secret"
 
     with pytest.raises(SystemExit) as exc_info:
         cli.run_cli(
             [
                 "--config",
                 "unused.json",
-                "--mode",
-                f"api_key={secret}",
+                *invalid_args,
             ]
         )
 
     assert exc_info.value.code == 2
     stderr = capsys.readouterr().err
-    assert secret not in stderr
-    assert "[REDACTED]" in stderr
+    assert "runs-secret" not in stderr
+    assert "case-secret" not in stderr
+    assert "api_key" not in stderr
+    assert "invalid command-line arguments" in stderr
+    assert "invalid int value" not in stderr
+    assert "invalid choice" not in stderr
+    assert "choose from" not in stderr
 
 
 def test_main_raises_system_exit(monkeypatch: pytest.MonkeyPatch):

tests/test_benchmark_reporting.py

@@ -125,12 +125,13 @@ def _process_report_writer(
     original_write_text = reporting._write_text
     first_write = True
 
-    def synchronized_write(path: Path, content: str) -> None:
+    def synchronized_write(path: Path, content: str) -> tuple[int, int]:
         nonlocal first_write
-        original_write_text(path, content)
+        identity = original_write_text(path, content)
         if first_write:
             first_write = False
             first_temp_barrier.wait()
+        return identity
 
     reporting._write_text = synchronized_write
     try:
@@ -588,6 +589,27 @@ def test_atomic_write_creates_matching_unique_json_and_markdown_paths(tmp_path:
     assert list(tmp_path.glob("*.tmp")) == []
 
 
+def test_second_placeholder_collision_cleans_first_and_uses_suffix(
+    tmp_path: Path,
+):
+    reporting = _reporting_module()
+    timestamp = "20260714T011223Z"
+    external_markdown = tmp_path / f"{timestamp}.md"
+    external_markdown.write_text("external markdown")
+
+    json_path, markdown_path = reporting.write_benchmark_report(
+        _marker_report("second-placeholder-collision"),
+        tmp_path,
+        timestamp=timestamp,
+    )
+
+    assert external_markdown.read_text() == "external markdown"
+    assert not (tmp_path / f"{timestamp}.json").exists()
+    assert json_path.name == f"{timestamp}-1.json"
+    assert markdown_path.name == f"{timestamp}-1.md"
+    assert list(tmp_path.glob("*.lock")) == []
+
+
 def test_atomic_write_removes_temps_and_published_half_on_replace_failure(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ):
@@ -620,19 +642,46 @@ def test_atomic_write_removes_temps_and_published_half_on_replace_failure(
     assert list(tmp_path.iterdir()) == []
 
 
-def test_reservation_failure_after_lock_creation_removes_own_lock(
+def test_temp_fsync_failure_removes_owned_temp_and_placeholders(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+):
+    reporting = _reporting_module()
+
+    def failing_fsync(file_descriptor: int) -> None:
+        del file_descriptor
+        raise OSError("simulated temp fsync failure")
+
+    monkeypatch.setattr(reporting.os, "fsync", failing_fsync)
+
+    with pytest.raises(
+        reporting.BenchmarkReportWriteError,
+        match="failed to write benchmark reports",
+    ):
+        reporting.write_benchmark_report(
+            _marker_report("temp-fsync-failure"),
+            tmp_path,
+            timestamp="20260714T012334Z",
+        )
+
+    assert list(tmp_path.iterdir()) == []
+
+
+def test_second_placeholder_failure_removes_first_placeholder(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
 ):
     reporting = _reporting_module()
-    original_exists = Path.exists
+    original_open = reporting.os.open
+    timestamp = "20260714T040506Z"
+    markdown_path = tmp_path / f"{timestamp}.md"
 
-    def failing_exists(path: Path) -> bool:
-        if path.parent == tmp_path and path.suffix == ".json":
-            raise OSError("simulated reservation path check failure")
-        return original_exists(path)
+    def failing_open(path, flags, mode=0o777):
+        if Path(path) == markdown_path:
+            raise OSError("simulated second placeholder failure")
+        return original_open(path, flags, mode)
 
-    monkeypatch.setattr(Path, "exists", failing_exists)
+    monkeypatch.setattr(reporting.os, "open", failing_open)
 
     with pytest.raises(
         reporting.BenchmarkReportWriteError,
@@ -641,12 +690,108 @@ def test_reservation_failure_after_lock_creation_removes_own_lock(
         reporting.write_benchmark_report(
             _marker_report("reservation-failure"),
             tmp_path,
-            timestamp="20260714T040506Z",
+            timestamp=timestamp,
         )
 
     assert list(tmp_path.iterdir()) == []
 
 
+def test_reservation_uses_final_placeholders_without_sidecar_lock(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+):
+    reporting = _reporting_module()
+    timestamp = "20260714T050607Z"
+    first_temp_written = threading.Event()
+    allow_publish = threading.Event()
+    original_write_text = reporting._write_text
+    first_write = True
+
+    def paused_write(path: Path, content: str) -> tuple[int, int]:
+        nonlocal first_write
+        identity = original_write_text(path, content)
+        if first_write:
+            first_write = False
+            first_temp_written.set()
+            assert allow_publish.wait(timeout=10)
+        return identity
+
+    monkeypatch.setattr(reporting, "_write_text", paused_write)
+    with ThreadPoolExecutor(max_workers=1) as executor:
+        future = executor.submit(
+            reporting.write_benchmark_report,
+            _marker_report("placeholder-owner"),
+            tmp_path,
+            timestamp=timestamp,
+        )
+        assert first_temp_written.wait(timeout=10)
+        json_placeholder = tmp_path / f"{timestamp}.json"
+        markdown_placeholder = tmp_path / f"{timestamp}.md"
+        try:
+            assert json_placeholder.exists()
+            assert markdown_placeholder.exists()
+            assert json_placeholder.read_bytes() == b""
+            assert markdown_placeholder.read_bytes() == b""
+            assert list(tmp_path.glob("*.lock")) == []
+        finally:
+            allow_publish.set()
+        json_path, markdown_path = future.result(timeout=10)
+
+    assert json_path == json_placeholder
+    assert markdown_path == markdown_placeholder
+
+
+@pytest.mark.parametrize("competed_suffix", [".json", ".md"])
+def test_external_replacement_of_reserved_final_is_not_overwritten_or_deleted(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    competed_suffix: str,
+):
+    reporting = _reporting_module()
+    timestamp = "20260714T060708Z"
+    first_temp_written = threading.Event()
+    allow_publish = threading.Event()
+    original_write_text = reporting._write_text
+    first_write = True
+
+    def paused_write(path: Path, content: str) -> tuple[int, int]:
+        nonlocal first_write
+        identity = original_write_text(path, content)
+        if first_write:
+            first_write = False
+            first_temp_written.set()
+            assert allow_publish.wait(timeout=10)
+        return identity
+
+    monkeypatch.setattr(reporting, "_write_text", paused_write)
+    competed_path = tmp_path / f"{timestamp}{competed_suffix}"
+    competitor_temp = tmp_path / f"competitor{competed_suffix}"
+    competitor_content = b"external competitor content"
+
+    with ThreadPoolExecutor(max_workers=1) as executor:
+        future = executor.submit(
+            reporting.write_benchmark_report,
+            _marker_report("must-not-overwrite"),
+            tmp_path,
+            timestamp=timestamp,
+        )
+        assert first_temp_written.wait(timeout=10)
+        competitor_temp.write_bytes(competitor_content)
+        reporting.os.replace(competitor_temp, competed_path)
+        allow_publish.set()
+        with pytest.raises(
+            reporting.BenchmarkReportWriteError,
+            match="failed to write benchmark reports",
+        ):
+            future.result(timeout=10)
+
+    assert competed_path.read_bytes() == competitor_content
+    other_suffix = ".md" if competed_suffix == ".json" else ".json"
+    assert not (tmp_path / f"{timestamp}{other_suffix}").exists()
+    assert list(tmp_path.glob("*.lock")) == []
+    assert list(tmp_path.glob("*.tmp")) == []
+
+
 def test_concurrent_thread_writers_reserve_distinct_matching_pairs(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
@@ -658,11 +803,12 @@ def test_concurrent_thread_writers_reserve_distinct_matching_pairs(
     original_write_text = reporting._write_text
     thread_state = threading.local()
 
-    def synchronized_write(path: Path, content: str) -> None:
-        original_write_text(path, content)
+    def synchronized_write(path: Path, content: str) -> tuple[int, int]:
+        identity = original_write_text(path, content)
         if not getattr(thread_state, "first_write_done", False):
             thread_state.first_write_done = True
             first_temp_barrier.wait()
+        return identity
 
     monkeypatch.setattr(reporting, "_write_text", synchronized_write)