From c74538a2be0c9e63653201ead33b47a8c17535fd Mon Sep 17 00:00:00 2001 From: Joseph Magly <1159087+jmagly@users.noreply.github.com> Date: Sat, 15 Aug 2026 23:05:48 -0400 Subject: [PATCH] test: harden checkpoint transaction reliability --- obliteratus/abliterate.py | 24 +- obliteratus/persistence_contracts.py | 280 +++++++++++++++-- pyproject.toml | 3 + scripts/run_repeat_gate.py | 2 + tests/test_abliterate.py | 14 +- tests/test_abliterate_extended.py | 14 +- tests/test_checkpoint_atomicity.py | 202 ++++++++++++ tests/test_persistence_contracts.py | 449 ++++++++++++++++++++++++++- tests/test_persistence_pipeline.py | 68 +++- tests/test_quality_policy.py | 8 + 10 files changed, 1016 insertions(+), 48 deletions(-) diff --git a/obliteratus/abliterate.py b/obliteratus/abliterate.py index e677d43..fa5fa20 100644 --- a/obliteratus/abliterate.py +++ b/obliteratus/abliterate.py @@ -53,6 +53,7 @@ from obliteratus.persistence_contracts import ( # noqa: E402 ensure_checkpoint_capacity, serialize_checkpoint_metadata, state_dict_size_bytes, + validate_local_checkpoint, ) from obliteratus.runtime_contracts import ( # noqa: E402 attention_projection_names, @@ -6490,10 +6491,17 @@ class AbliterationPipeline: if owns_offload_dir and offload_dir: offload_path = Path(offload_dir) if offload_path.exists(): - size_mb = sum( - f.stat().st_size for f in offload_path.rglob("*") if f.is_file() - ) / (1024 ** 2) - shutil.rmtree(offload_path, ignore_errors=True) + try: + size_mb = sum( + f.stat().st_size for f in offload_path.rglob("*") if f.is_file() + ) / (1024 ** 2) + shutil.rmtree(offload_path) + except OSError as cleanup_error: + self.log( + "Offload cleanup failed; retaining the owned path for retry: " + f"{offload_path} ({cleanup_error})", + ) + return self.log(f"Cleaned up offload dir ({size_mb:.0f} MiB reclaimed)") self.handle._offload_dir = None self.handle._owns_offload_dir = False @@ -6622,7 +6630,13 @@ class AbliterationPipeline: # the complete checkpoint atomically. The offload dir remains intact # until promotion so a failed write can be retried safely. self.log(f"Saving model to {self.output_dir}/") - with _atomic_checkpoint_directory(self.output_dir) as checkpoint_dir: + def validate_checkpoint(checkpoint_dir: Path) -> None: + validate_local_checkpoint(checkpoint_dir, metadata_json) + + with _atomic_checkpoint_directory( + self.output_dir, + validate=validate_checkpoint, + ) as checkpoint_dir: try: disk = shutil.disk_usage(checkpoint_dir) ensure_checkpoint_capacity(disk.free, param_bytes) diff --git a/obliteratus/persistence_contracts.py b/obliteratus/persistence_contracts.py index 9f8adc3..a4e691c 100644 --- a/obliteratus/persistence_contracts.py +++ b/obliteratus/persistence_contracts.py @@ -6,16 +6,22 @@ import json import logging import os import shutil +import stat import tempfile import uuid from collections.abc import Mapping from contextlib import contextmanager from pathlib import Path -from typing import Any, Iterator, Protocol +from typing import Any, Callable, Iterator, Protocol logger = logging.getLogger(__name__) +if os.name == "nt": # pragma: no cover - exercised by the Windows CI lane + _DIRECTORY_SYNC_FLAGS: int | None = None +else: + _DIRECTORY_SYNC_FLAGS = os.O_RDONLY | os.O_DIRECTORY + class SizedTensor(Protocol): """Structural subset used to estimate a serialized state dictionary.""" @@ -55,22 +61,221 @@ def serialize_checkpoint_metadata(metadata: Mapping[str, Any]) -> str: return json.dumps(metadata, indent=2, sort_keys=True, allow_nan=False) +def _read_json_object(path: Path, artifact_name: str) -> Mapping[str, Any]: + """Load a required, non-empty JSON object from a checkpoint.""" + if path.is_symlink() or not path.is_file() or path.stat().st_size == 0: + raise ValueError(f"Checkpoint {artifact_name} is missing or empty: {path}") + try: + value = json.loads(path.read_bytes().decode()) + except (OSError, UnicodeError, json.JSONDecodeError) as error: + raise ValueError(f"Checkpoint {artifact_name} is corrupt: {path}") from error + if not isinstance(value, Mapping): + raise ValueError(f"Checkpoint {artifact_name} must contain a JSON object: {path}") + return value + + +def _require_nonempty_regular_file(path: Path, artifact_name: str) -> None: + """Reject absent, empty, or link-backed checkpoint payloads.""" + if ( + path.is_symlink() + or not path.is_file() + or not stat.S_ISREG(path.stat().st_mode) + or path.stat().st_size == 0 + ): + raise ValueError(f"Checkpoint {artifact_name} is missing or empty: {path}") + + +def validate_local_checkpoint( + checkpoint_dir: Path, + expected_metadata_json: str, +) -> None: + """Validate the minimum artifacts needed for a local Transformers reload.""" + checkpoint_dir = Path(checkpoint_dir) + if checkpoint_dir.is_symlink() or not checkpoint_dir.is_dir(): + raise ValueError(f"Checkpoint staging path is not a directory: {checkpoint_dir}") + + expected_metadata = json.loads(expected_metadata_json) + actual_metadata = _read_json_object( + checkpoint_dir / "abliteration_metadata.json", + "metadata", + ) + if actual_metadata != expected_metadata: + raise ValueError("Checkpoint metadata does not match the prepared transaction") + + _read_json_object(checkpoint_dir / "config.json", "model config") + _read_json_object(checkpoint_dir / "tokenizer_config.json", "tokenizer config") + + direct_weights = [ + checkpoint_dir / "model.safetensors", + checkpoint_dir / "pytorch_model.bin", + ] + for weights_path in direct_weights: + if _path_exists(weights_path): + _require_nonempty_regular_file(weights_path, "weights") + return + + index_paths = [ + checkpoint_dir / "model.safetensors.index.json", + checkpoint_dir / "pytorch_model.bin.index.json", + ] + for index_path in index_paths: + if not _path_exists(index_path): + continue + index = _read_json_object(index_path, "weight index") + weight_map = index.get("weight_map") + if not isinstance(weight_map, Mapping) or not weight_map: + raise ValueError(f"Checkpoint weight index has no weight map: {index_path}") + shard_values = list(weight_map.values()) + if not all(isinstance(name, str) and Path(name).name == name for name in shard_values): + raise ValueError(f"Checkpoint weight index contains an unsafe shard path: {index_path}") + shard_names = set(shard_values) + for shard_name in shard_names: + _require_nonempty_regular_file( + checkpoint_dir / shard_name, + "weight shard", + ) + return + + raise ValueError(f"Checkpoint has no model weights: {checkpoint_dir}") + + +def _path_exists(path: Path) -> bool: + """Return whether a path or dangling symlink occupies ``path``.""" + return path.exists() or path.is_symlink() + + def _remove_checkpoint_path(path: Path) -> None: """Remove a staging/backup path without following directory symlinks.""" if path.is_symlink() or path.is_file(): path.unlink(missing_ok=True) elif path.exists(): - shutil.rmtree(path, ignore_errors=True) + try: + shutil.rmtree(path) + except FileNotFoundError: + pass + + +def _sync_file(path: Path) -> None: + """Flush one regular checkpoint artifact to stable storage.""" + mode = path.lstat().st_mode + if not stat.S_ISREG(mode): + raise OSError(f"Checkpoint artifact is not a regular file: {path}") + descriptor = os.open(path, os.O_RDONLY) + try: + os.fsync(descriptor) + finally: + os.close(descriptor) + + +def _sync_directory(path: Path) -> None: + """Persist directory entries where the platform exposes directory fsync.""" + if _DIRECTORY_SYNC_FLAGS is None: + return + descriptor = os.open(path, _DIRECTORY_SYNC_FLAGS) + try: + os.fsync(descriptor) + finally: + os.close(descriptor) + + +def _sync_checkpoint_tree(checkpoint_dir: Path) -> None: + """Flush checkpoint files and directories from leaves to root.""" + for root, directory_names, file_names in os.walk( + checkpoint_dir, + topdown=False, + followlinks=False, + ): + root_path = Path(root) + for file_name in sorted(file_names): + _sync_file(root_path / file_name) + for directory_name in sorted(directory_names): + directory = root_path / directory_name + if directory.is_symlink(): + raise OSError( + f"Checkpoint directory contains a symlink: {directory}", + ) + _sync_directory(directory) + _sync_directory(root_path) @contextmanager -def atomic_checkpoint_directory(destination: Path) -> Iterator[Path]: +def _checkpoint_lock(destination: Path) -> Iterator[None]: + """Serialize commits to one destination across threads and processes.""" + lock_path = destination.with_name(f".{destination.name}.lock") + with lock_path.open("a+b") as lock_file: + if os.name == "nt": # pragma: no cover - exercised by the Windows CI lane + import msvcrt + + lock_file.seek(0, os.SEEK_END) + if lock_file.tell() == 0: + lock_file.write(b"\0") + lock_file.flush() + lock_file.seek(0) + msvcrt.locking(lock_file.fileno(), msvcrt.LK_LOCK, 1) + try: + yield + finally: + lock_file.seek(0) + msvcrt.locking(lock_file.fileno(), msvcrt.LK_UNLCK, 1) + else: + import fcntl + + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX) + try: + yield + finally: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN) + + +def _rollback_checkpoint_commit( + destination: Path, + staging: Path, + backup: Path | None, + *, + had_destination: bool, +) -> None: + """Restore the pre-commit state after a failed or cancelled commit.""" + if backup is not None and _path_exists(backup): + if _path_exists(destination): + if _path_exists(staging): + raise RuntimeError( + "Checkpoint rollback found both destination and staging; " + f"recover the previous checkpoint from {backup}", + ) + os.replace(destination, staging) + os.replace(backup, destination) + elif not had_destination and _path_exists(destination) and not _path_exists(staging): + os.replace(destination, staging) + _sync_directory(destination.parent) + + +def _cleanup_failed_staging(staging: Path) -> None: + """Remove failed staging without masking the transaction's primary error.""" + try: + _remove_checkpoint_path(staging) + except OSError as cleanup_error: + logger.warning( + "Checkpoint write failed, and staging directory %s could not be " + "removed: %s", + staging, + cleanup_error, + ) + + +@contextmanager +def atomic_checkpoint_directory( + destination: Path, + *, + validate: Callable[[Path], None] | None = None, +) -> Iterator[Path]: """Yield a sibling staging directory and atomically promote it on success. An existing checkpoint is moved to a uniquely named backup immediately - before promotion. If promotion fails, that backup is restored. Exceptions - while writing only remove the staging directory and leave the destination - untouched. + before promotion. Concurrent commits to the same destination are serialized. + Validation and durable file flushes happen before the commit lock is taken. + If commit or parent-directory sync fails, the previous destination is + restored. Cancellation and other ``BaseException`` failures follow the same + rollback path as ordinary exceptions. """ destination = Path(destination) destination.parent.mkdir(parents=True, exist_ok=True) @@ -83,33 +288,46 @@ def atomic_checkpoint_directory(destination: Path) -> Iterator[Path]: backup: Path | None = None try: yield staging - if destination.exists() or destination.is_symlink(): - backup = destination.with_name( - f".{destination.name}.backup-{uuid.uuid4().hex}", - ) - os.replace(destination, backup) - try: - os.replace(staging, destination) - except Exception as promotion_error: - if backup is not None and (backup.exists() or backup.is_symlink()): + if validate is not None: + validate(staging) + _sync_checkpoint_tree(staging) + with _checkpoint_lock(destination): + had_destination = _path_exists(destination) + try: + if had_destination: + backup = destination.with_name( + f".{destination.name}.backup-{uuid.uuid4().hex}", + ) + os.replace(destination, backup) + _sync_directory(destination.parent) + os.replace(staging, destination) + _sync_directory(destination.parent) + except BaseException: try: - os.replace(backup, destination) - except Exception as restore_error: + _rollback_checkpoint_commit( + destination, + staging, + backup, + had_destination=had_destination, + ) + except BaseException as restore_error: + recovery_path = backup if backup is not None else staging raise RuntimeError( "Checkpoint promotion and rollback both failed; " - f"recover the previous checkpoint from {backup}", + f"recover the previous checkpoint from {recovery_path}", ) from restore_error - raise promotion_error - if backup is not None: - try: - _remove_checkpoint_path(backup) - except OSError as cleanup_error: - logger.warning( - "Checkpoint promoted, but previous-checkpoint backup %s " - "could not be removed: %s", - backup, - cleanup_error, - ) - except Exception: - _remove_checkpoint_path(staging) + raise + if backup is not None: + try: + _remove_checkpoint_path(backup) + _sync_directory(destination.parent) + except OSError as cleanup_error: + logger.warning( + "Checkpoint promoted, but previous-checkpoint backup %s " + "could not be removed durably: %s", + backup, + cleanup_error, + ) + except BaseException: + _cleanup_failed_staging(staging) raise diff --git a/pyproject.toml b/pyproject.toml index 9744bc8..9af73e6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -128,6 +128,7 @@ only_mutate = [ required_mutation_targets = [ "obliteratus/analysis/numerical_contracts.py", "obliteratus/analysis/whitened_svd.py", + "obliteratus/persistence_contracts.py", "obliteratus/runtime_contracts.py", ] pytest_add_cli_args = ["--no-cov", "-q"] @@ -138,7 +139,9 @@ pytest_add_cli_args_test_selection = [ "tests/test_projection_math_contracts.py", "tests/test_lm_eval_reporting_contracts.py", "tests/test_numerical_contracts.py", + "tests/test_checkpoint_atomicity.py", "tests/test_persistence_contracts.py", + "tests/test_persistence_pipeline.py", "tests/test_remote_contracts.py", "tests/test_runtime_contracts.py", "tests/test_whitened_svd_oracles.py", diff --git a/scripts/run_repeat_gate.py b/scripts/run_repeat_gate.py index 555b56a..01a12b0 100644 --- a/scripts/run_repeat_gate.py +++ b/scripts/run_repeat_gate.py @@ -16,6 +16,7 @@ from xml.etree import ElementTree DEFAULT_TESTS = ( "tests/test_bayesian_optimizer_contracts.py", + "tests/test_checkpoint_atomicity.py", "tests/test_config.py", "tests/test_config_properties.py", "tests/test_conditional_evidence_freshness.py", @@ -27,6 +28,7 @@ DEFAULT_TESTS = ( "tests/test_numerical_contracts.py", "tests/test_package_export_contracts.py", "tests/test_persistence_contracts.py", + "tests/test_persistence_pipeline.py", "tests/test_property_contracts.py", "tests/test_advanced_metrics.py", "tests/test_metrics.py", diff --git a/tests/test_abliterate.py b/tests/test_abliterate.py index fcfc70f..28f91c8 100644 --- a/tests/test_abliterate.py +++ b/tests/test_abliterate.py @@ -1977,8 +1977,18 @@ class TestRebirth: pipeline._strong_layers = [0] pipeline._quality_metrics = {"perplexity": 8.5, "coherence": 1.0} - handle.model.save_pretrained = MagicMock() - handle.tokenizer.save_pretrained = MagicMock() + handle.model.save_pretrained = MagicMock( + side_effect=lambda path, **_kwargs: ( + (Path(path) / "config.json").write_text("{}", encoding="utf-8"), + (Path(path) / "model.safetensors").write_bytes(b"weights"), + ), + ) + handle.tokenizer.save_pretrained = MagicMock( + side_effect=lambda path: (Path(path) / "tokenizer_config.json").write_text( + "{}", + encoding="utf-8", + ), + ) result_path = pipeline._rebirth() diff --git a/tests/test_abliterate_extended.py b/tests/test_abliterate_extended.py index ec45001..c80162c 100644 --- a/tests/test_abliterate_extended.py +++ b/tests/test_abliterate_extended.py @@ -277,8 +277,18 @@ class TestMetadata: pipeline._strong_layers = [0] pipeline._quality_metrics = {"perplexity": 8.5, "coherence": 1.0} - handle.model.save_pretrained = MagicMock() - handle.tokenizer.save_pretrained = MagicMock() + handle.model.save_pretrained = MagicMock( + side_effect=lambda path, **_kwargs: ( + (Path(path) / "config.json").write_text("{}", encoding="utf-8"), + (Path(path) / "model.safetensors").write_bytes(b"weights"), + ), + ) + handle.tokenizer.save_pretrained = MagicMock( + side_effect=lambda path: (Path(path) / "tokenizer_config.json").write_text( + "{}", + encoding="utf-8", + ), + ) import tempfile from pathlib import Path diff --git a/tests/test_checkpoint_atomicity.py b/tests/test_checkpoint_atomicity.py index 93bb79f..96e0279 100644 --- a/tests/test_checkpoint_atomicity.py +++ b/tests/test_checkpoint_atomicity.py @@ -9,6 +9,7 @@ from unittest.mock import MagicMock import pytest import torch +import obliteratus.abliterate as abliterate import obliteratus.persistence_contracts as persistence from obliteratus.abliterate import AbliterationPipeline, _atomic_checkpoint_directory @@ -16,6 +17,10 @@ from obliteratus.abliterate import AbliterationPipeline, _atomic_checkpoint_dire pytestmark = pytest.mark.cpu +class SimulatedCancellation(BaseException): + """Cancellation signal used to prove BaseException rollback behavior.""" + + def _temporary_artifacts(parent: Path, name: str) -> list[Path]: return [ *parent.glob(f".{name}.staging-*"), @@ -88,6 +93,146 @@ def test_atomic_checkpoint_preserves_destination_on_write_failure(tmp_path): assert _temporary_artifacts(tmp_path, destination.name) == [] +def test_atomic_checkpoint_cleans_staging_on_cancellation_during_write(tmp_path): + destination = tmp_path / "checkpoint" + destination.mkdir() + sentinel = destination / "sentinel.txt" + sentinel.write_text("preserve me", encoding="utf-8") + + with pytest.raises(SimulatedCancellation): + with _atomic_checkpoint_directory(destination) as staging: + (staging / "partial.bin").write_bytes(b"partial") + raise SimulatedCancellation() + + assert sentinel.read_text(encoding="utf-8") == "preserve me" + assert _temporary_artifacts(tmp_path, destination.name) == [] + + +def test_atomic_checkpoint_preserves_destination_on_file_fsync_failure( + tmp_path, + monkeypatch, +): + destination = tmp_path / "checkpoint" + destination.mkdir() + sentinel = destination / "sentinel.txt" + sentinel.write_text("preserve me", encoding="utf-8") + + def fail_sync(_path): + raise OSError("simulated fsync failure") + + monkeypatch.setattr(persistence, "_sync_file", fail_sync) + with pytest.raises(OSError, match="simulated fsync failure"): + with _atomic_checkpoint_directory(destination) as staging: + (staging / "partial.bin").write_bytes(b"partial") + + assert sentinel.read_text(encoding="utf-8") == "preserve me" + assert _temporary_artifacts(tmp_path, destination.name) == [] + + +def test_new_checkpoint_parent_sync_failure_rolls_back_then_allows_retry( + tmp_path, + monkeypatch, +): + destination = tmp_path / "checkpoint" + real_sync_directory = persistence._sync_directory + sync_calls = 0 + failed = False + + def fail_after_promotion(path): + nonlocal sync_calls, failed + sync_calls += 1 + if sync_calls == 2 and not failed: + failed = True + raise OSError("simulated parent fsync failure") + return real_sync_directory(path) + + monkeypatch.setattr(persistence, "_sync_directory", fail_after_promotion) + + with pytest.raises(OSError, match="simulated parent fsync failure"): + with _atomic_checkpoint_directory(destination) as staging: + (staging / "new.txt").write_text("first attempt", encoding="utf-8") + + assert not destination.exists() + assert _temporary_artifacts(tmp_path, destination.name) == [] + + with _atomic_checkpoint_directory(destination) as staging: + (staging / "new.txt").write_text("retry", encoding="utf-8") + + assert (destination / "new.txt").read_text(encoding="utf-8") == "retry" + + +@pytest.mark.parametrize( + "failure_phase", + [ + "tree_sync", + "backup_replace", + "backup_parent_sync", + "promotion_replace", + "promotion_parent_sync", + ], +) +def test_atomic_checkpoint_cancellation_restores_and_unlocks_for_retry( + tmp_path, + monkeypatch, + failure_phase, +): + destination = tmp_path / "checkpoint" + destination.mkdir() + sentinel = destination / "sentinel.txt" + sentinel.write_text("preserve me", encoding="utf-8") + real_replace = os.replace + real_sync_file = persistence._sync_file + real_sync_directory = persistence._sync_directory + replace_calls = 0 + directory_sync_calls = 0 + cancelled = False + + def maybe_cancel_file_sync(path): + nonlocal cancelled + if failure_phase == "tree_sync" and not cancelled: + cancelled = True + raise SimulatedCancellation() + return real_sync_file(path) + + def maybe_cancel_replace(source, target): + nonlocal replace_calls, cancelled + replace_calls += 1 + target_call = 1 if failure_phase == "backup_replace" else 2 + if failure_phase in {"backup_replace", "promotion_replace"}: + if replace_calls == target_call and not cancelled: + cancelled = True + raise SimulatedCancellation() + return real_replace(source, target) + + def maybe_cancel_directory_sync(path): + nonlocal directory_sync_calls, cancelled + directory_sync_calls += 1 + target_call = 2 if failure_phase == "backup_parent_sync" else 3 + if failure_phase in {"backup_parent_sync", "promotion_parent_sync"}: + if directory_sync_calls == target_call and not cancelled: + cancelled = True + raise SimulatedCancellation() + return real_sync_directory(path) + + monkeypatch.setattr(persistence, "_sync_file", maybe_cancel_file_sync) + monkeypatch.setattr(persistence.os, "replace", maybe_cancel_replace) + monkeypatch.setattr(persistence, "_sync_directory", maybe_cancel_directory_sync) + + with pytest.raises(SimulatedCancellation): + with _atomic_checkpoint_directory(destination) as staging: + (staging / "new.txt").write_text("first attempt", encoding="utf-8") + + assert sentinel.read_text(encoding="utf-8") == "preserve me" + assert _temporary_artifacts(tmp_path, destination.name) == [] + + with _atomic_checkpoint_directory(destination) as staging: + (staging / "new.txt").write_text("retry", encoding="utf-8") + + assert not sentinel.exists() + assert (destination / "new.txt").read_text(encoding="utf-8") == "retry" + assert _temporary_artifacts(tmp_path, destination.name) == [] + + def test_atomic_checkpoint_restores_destination_on_promotion_failure( tmp_path, monkeypatch, @@ -139,6 +284,36 @@ def test_atomic_checkpoint_reports_recoverable_backup_when_rollback_fails( assert (backups[0] / "sentinel.txt").read_text() == "recover me" +def test_atomic_checkpoint_reports_staging_cleanup_failure_without_masking_write( + tmp_path, + monkeypatch, + caplog, +): + destination = tmp_path / "checkpoint" + destination.mkdir() + sentinel = destination / "sentinel.txt" + sentinel.write_text("preserve me", encoding="utf-8") + real_remove = persistence._remove_checkpoint_path + + def fail_staging_cleanup(path): + if ".staging-" in path.name: + raise PermissionError("simulated cleanup denial") + return real_remove(path) + + monkeypatch.setattr(persistence, "_remove_checkpoint_path", fail_staging_cleanup) + with pytest.raises(OSError, match="simulated write failure"): + with _atomic_checkpoint_directory(destination) as staging: + (staging / "partial.bin").write_bytes(b"partial") + raise OSError("simulated write failure") + + assert sentinel.read_text(encoding="utf-8") == "preserve me" + assert caplog.messages == [ + "Checkpoint write failed, and staging directory " + f"{staging} could not be removed: simulated cleanup denial", + ] + assert len(list(tmp_path.glob(".checkpoint.staging-*"))) == 1 + + def test_rebirth_failure_preserves_checkpoint_and_owned_offload(tmp_path): destination = tmp_path / "checkpoint" destination.mkdir() @@ -212,3 +387,30 @@ def test_cleanup_clears_stale_owned_offload_reference(tmp_path): assert pipeline.handle._offload_dir is None assert pipeline.handle._owns_offload_dir is False + + +def test_cleanup_failure_retains_owned_offload_reference_for_retry( + tmp_path, + monkeypatch, +): + owned = tmp_path / "pipeline-owned" + owned.mkdir() + (owned / "weight.bin").write_bytes(b"temporary") + messages = [] + pipeline = AbliterationPipeline(model_name="test-model", method="basic") + pipeline._on_log = messages.append + pipeline.handle = MagicMock() + pipeline.handle._offload_dir = str(owned) + pipeline.handle._owns_offload_dir = True + + def deny_cleanup(_path): + raise PermissionError("simulated cleanup denial") + + monkeypatch.setattr(abliterate.shutil, "rmtree", deny_cleanup) + + pipeline._cleanup_offload_dir() + + assert (owned / "weight.bin").read_bytes() == b"temporary" + assert pipeline.handle._offload_dir == str(owned) + assert pipeline.handle._owns_offload_dir is True + assert any("retaining the owned path for retry" in message for message in messages) diff --git a/tests/test_persistence_contracts.py b/tests/test_persistence_contracts.py index 0725146..9454b78 100644 --- a/tests/test_persistence_contracts.py +++ b/tests/test_persistence_contracts.py @@ -4,7 +4,9 @@ from __future__ import annotations import json import os +from concurrent.futures import ThreadPoolExecutor from pathlib import Path +from threading import Barrier from unittest.mock import MagicMock import pytest @@ -15,6 +17,20 @@ import obliteratus.persistence_contracts as persistence pytestmark = pytest.mark.cpu + +def _write_valid_local_checkpoint( + checkpoint_dir: Path, + metadata_json: str = '{"schema": 1}', +) -> None: + (checkpoint_dir / "abliteration_metadata.json").write_text( + metadata_json, + encoding="utf-8", + ) + (checkpoint_dir / "config.json").write_text("{}", encoding="utf-8") + (checkpoint_dir / "tokenizer_config.json").write_text("{}", encoding="utf-8") + (checkpoint_dir / "model.safetensors").write_bytes(b"weights") + + @pytest.mark.parametrize( ("state_dict", "expected"), [ @@ -107,6 +123,215 @@ def test_metadata_serialization_rejects_nonportable_values(value): persistence.serialize_checkpoint_metadata({"invalid": value}) +@pytest.mark.parametrize("weights_name", ["model.safetensors", "pytorch_model.bin"]) +def test_validate_local_checkpoint_accepts_complete_direct_weights(tmp_path, weights_name): + _write_valid_local_checkpoint(tmp_path) + if weights_name != "model.safetensors": + (tmp_path / "model.safetensors").rename(tmp_path / weights_name) + + persistence.validate_local_checkpoint(tmp_path, '{"schema": 1}') + + +def test_validate_local_checkpoint_accepts_complete_sharded_weights(tmp_path): + metadata_json = '{"schema": 1}' + _write_valid_local_checkpoint(tmp_path, metadata_json) + (tmp_path / "model.safetensors").unlink() + (tmp_path / "model.safetensors.index.json").write_text( + json.dumps( + { + "metadata": {"total_size": 2}, + "weight_map": { + "layer.0": "model-00001-of-00002.safetensors", + "layer.1": "model-00002-of-00002.safetensors", + }, + }, + ), + encoding="utf-8", + ) + (tmp_path / "model-00001-of-00002.safetensors").write_bytes(b"a") + (tmp_path / "model-00002-of-00002.safetensors").write_bytes(b"b") + + persistence.validate_local_checkpoint(tmp_path, metadata_json) + + +def test_validate_local_checkpoint_accepts_pytorch_sharded_weights(tmp_path): + _write_valid_local_checkpoint(tmp_path) + (tmp_path / "model.safetensors").unlink() + (tmp_path / "pytorch_model.bin.index.json").write_text( + json.dumps({"weight_map": {"layer": "pytorch_model-00001-of-00001.bin"}}), + encoding="utf-8", + ) + (tmp_path / "pytorch_model-00001-of-00001.bin").write_bytes(b"weights") + + persistence.validate_local_checkpoint(tmp_path, '{"schema": 1}') + + +@pytest.mark.parametrize("kind", ["file", "symlink"]) +def test_validate_local_checkpoint_rejects_non_directory_staging(tmp_path, kind): + staging = tmp_path / "staging" + if kind == "file": + staging.write_bytes(b"not a directory") + else: + target = tmp_path / "target" + target.mkdir() + staging.symlink_to(target, target_is_directory=True) + + with pytest.raises(ValueError, match="staging path is not a directory"): + persistence.validate_local_checkpoint(staging, '{"schema": 1}') + + +@pytest.mark.parametrize( + ("mutate", "message"), + [ + ( + lambda path: (path / "abliteration_metadata.json").write_text( + "{", + encoding="utf-8", + ), + "metadata is corrupt", + ), + ( + lambda path: (path / "abliteration_metadata.json").write_text( + '{"schema": 2}', + encoding="utf-8", + ), + "metadata does not match", + ), + ( + lambda path: (path / "config.json").write_text("[]", encoding="utf-8"), + "model config must contain a JSON object", + ), + ( + lambda path: (path / "tokenizer_config.json").write_bytes(b""), + "tokenizer config is missing or empty", + ), + ( + lambda path: (path / "model.safetensors").write_bytes(b""), + "weights is missing or empty", + ), + ], +) +def test_validate_local_checkpoint_rejects_corrupt_or_truncated_artifacts( + tmp_path, + mutate, + message, +): + _write_valid_local_checkpoint(tmp_path) + mutate(tmp_path) + + with pytest.raises(ValueError, match=message): + persistence.validate_local_checkpoint(tmp_path, '{"schema": 1}') + + +def test_validate_local_checkpoint_reports_exact_metadata_mismatch(tmp_path): + _write_valid_local_checkpoint(tmp_path, '{"schema": 2}') + + with pytest.raises(ValueError) as exc_info: + persistence.validate_local_checkpoint(tmp_path, '{"schema": 1}') + + assert str(exc_info.value) == ( + "Checkpoint metadata does not match the prepared transaction" + ) + + +@pytest.mark.parametrize( + "weight_map", + [ + {}, + {"layer": "missing.safetensors"}, + {"layer": "../outside.safetensors"}, + {"layer": ["not", "a", "path"]}, + ], +) +def test_validate_local_checkpoint_rejects_invalid_weight_indexes(tmp_path, weight_map): + _write_valid_local_checkpoint(tmp_path) + (tmp_path / "model.safetensors").unlink() + (tmp_path / "model.safetensors.index.json").write_text( + json.dumps({"weight_map": weight_map}), + encoding="utf-8", + ) + + with pytest.raises(ValueError, match="weight index|weight shard"): + persistence.validate_local_checkpoint(tmp_path, '{"schema": 1}') + + +def test_validate_local_checkpoint_reports_corrupt_weight_index_exactly(tmp_path): + _write_valid_local_checkpoint(tmp_path) + (tmp_path / "model.safetensors").unlink() + index_path = tmp_path / "model.safetensors.index.json" + index_path.write_text("{", encoding="utf-8") + + with pytest.raises(ValueError) as exc_info: + persistence.validate_local_checkpoint(tmp_path, '{"schema": 1}') + + assert str(exc_info.value) == f"Checkpoint weight index is corrupt: {index_path}" + + +def test_validate_local_checkpoint_reports_missing_weight_shard_exactly(tmp_path): + _write_valid_local_checkpoint(tmp_path) + (tmp_path / "model.safetensors").unlink() + index_path = tmp_path / "model.safetensors.index.json" + index_path.write_text( + json.dumps({"weight_map": {"layer": "missing.safetensors"}}), + encoding="utf-8", + ) + + with pytest.raises(ValueError) as exc_info: + persistence.validate_local_checkpoint(tmp_path, '{"schema": 1}') + + missing = tmp_path / "missing.safetensors" + assert str(exc_info.value) == f"Checkpoint weight shard is missing or empty: {missing}" + + +def test_validate_local_checkpoint_rejects_existing_unsafe_weight_shard(tmp_path): + _write_valid_local_checkpoint(tmp_path) + (tmp_path / "model.safetensors").unlink() + outside = tmp_path.parent / "outside.safetensors" + outside.write_bytes(b"outside") + index_path = tmp_path / "model.safetensors.index.json" + index_path.write_text( + json.dumps({"weight_map": {"layer": "../outside.safetensors"}}), + encoding="utf-8", + ) + + with pytest.raises(ValueError) as exc_info: + persistence.validate_local_checkpoint(tmp_path, '{"schema": 1}') + + assert str(exc_info.value) == ( + f"Checkpoint weight index contains an unsafe shard path: {index_path}" + ) + + +def test_validate_local_checkpoint_rejects_checkpoint_without_weights(tmp_path): + _write_valid_local_checkpoint(tmp_path) + (tmp_path / "model.safetensors").unlink() + + with pytest.raises(ValueError, match="has no model weights"): + persistence.validate_local_checkpoint(tmp_path, '{"schema": 1}') + + +def test_validate_local_checkpoint_rejects_link_backed_weights(tmp_path): + _write_valid_local_checkpoint(tmp_path) + (tmp_path / "model.safetensors").unlink() + outside = tmp_path / "outside.safetensors" + outside.write_bytes(b"outside") + (tmp_path / "model.safetensors").symlink_to(outside) + + with pytest.raises(ValueError, match="weights is missing or empty"): + persistence.validate_local_checkpoint(tmp_path, '{"schema": 1}') + + +def test_validate_local_checkpoint_rejects_link_backed_json(tmp_path): + _write_valid_local_checkpoint(tmp_path) + (tmp_path / "config.json").unlink() + outside = tmp_path / "outside.json" + outside.write_text("{}", encoding="utf-8") + (tmp_path / "config.json").symlink_to(outside) + + with pytest.raises(ValueError, match="model config is missing or empty"): + persistence.validate_local_checkpoint(tmp_path, '{"schema": 1}') + + def test_remove_checkpoint_path_handles_file_directory_symlink_and_missing(tmp_path): file_path = tmp_path / "file" file_path.write_text("data", encoding="utf-8") @@ -138,7 +363,7 @@ def test_remove_checkpoint_path_requests_race_safe_file_unlink(): path.is_file.assert_not_called() -def test_remove_checkpoint_path_requests_best_effort_directory_cleanup(monkeypatch): +def test_remove_checkpoint_path_requests_observable_directory_cleanup(monkeypatch): path = MagicMock() path.is_symlink.return_value = False path.is_file.return_value = False @@ -148,7 +373,22 @@ def test_remove_checkpoint_path_requests_best_effort_directory_cleanup(monkeypat persistence._remove_checkpoint_path(path) - remove_tree.assert_called_once_with(path, ignore_errors=True) + remove_tree.assert_called_once_with(path) + + +def test_remove_checkpoint_path_tolerates_directory_disappearing_during_cleanup( + monkeypatch, +): + path = MagicMock() + path.is_symlink.return_value = False + path.is_file.return_value = False + path.exists.return_value = True + remove_tree = MagicMock(side_effect=FileNotFoundError) + monkeypatch.setattr(persistence.shutil, "rmtree", remove_tree) + + persistence._remove_checkpoint_path(path) + + remove_tree.assert_called_once_with(path) def test_atomic_checkpoint_creates_parent_and_promotes_new_destination(tmp_path): @@ -213,3 +453,208 @@ def test_atomic_checkpoint_uses_os_replace_not_copy(tmp_path, monkeypatch): assert len(calls) == 1 assert calls[0][1] == destination assert ".staging-" in calls[0][0].name + + +def test_atomic_checkpoint_flushes_files_before_promotion(tmp_path, monkeypatch): + destination = tmp_path / "checkpoint" + events = [] + real_sync_file = persistence._sync_file + real_replace = os.replace + + def record_sync(path): + events.append(("sync", Path(path).name)) + return real_sync_file(path) + + def record_replace(source, target): + events.append(("replace", Path(source).name)) + return real_replace(source, target) + + monkeypatch.setattr(persistence, "_sync_file", record_sync) + monkeypatch.setattr(persistence.os, "replace", record_replace) + + with persistence.atomic_checkpoint_directory(destination) as staging: + (staging / "model.bin").write_bytes(b"saved") + + assert events[0] == ("sync", "model.bin") + assert events[1][0] == "replace" + + +def test_atomic_checkpoint_flushes_nested_directories(tmp_path, monkeypatch): + destination = tmp_path / "checkpoint" + synced_directories = [] + real_sync_directory = persistence._sync_directory + + def record_directory_sync(path): + synced_directories.append(Path(path).name) + return real_sync_directory(path) + + monkeypatch.setattr(persistence, "_sync_directory", record_directory_sync) + + with persistence.atomic_checkpoint_directory(destination) as staging: + nested = staging / "nested" + nested.mkdir() + (nested / "payload").write_bytes(b"saved") + + assert "nested" in synced_directories + + +def test_sync_directory_uses_platform_flags_and_closes_descriptor( + tmp_path, + monkeypatch, +): + open_file = MagicMock(return_value=17) + sync_file = MagicMock() + close_file = MagicMock() + monkeypatch.setattr(persistence, "_DIRECTORY_SYNC_FLAGS", 123) + monkeypatch.setattr(persistence.os, "open", open_file) + monkeypatch.setattr(persistence.os, "fsync", sync_file) + monkeypatch.setattr(persistence.os, "close", close_file) + + persistence._sync_directory(tmp_path) + + open_file.assert_called_once_with(tmp_path, 123) + sync_file.assert_called_once_with(17) + close_file.assert_called_once_with(17) + + +def test_sync_directory_skips_unsupported_platform(tmp_path, monkeypatch): + open_file = MagicMock() + monkeypatch.setattr(persistence, "_DIRECTORY_SYNC_FLAGS", None) + monkeypatch.setattr(persistence.os, "open", open_file) + + persistence._sync_directory(tmp_path) + + open_file.assert_not_called() + + +def test_sync_directory_closes_descriptor_after_fsync_failure(tmp_path, monkeypatch): + close_file = MagicMock() + monkeypatch.setattr(persistence, "_DIRECTORY_SYNC_FLAGS", 123) + monkeypatch.setattr(persistence.os, "open", MagicMock(return_value=17)) + monkeypatch.setattr( + persistence.os, + "fsync", + MagicMock(side_effect=OSError("simulated fsync failure")), + ) + monkeypatch.setattr(persistence.os, "close", close_file) + + with pytest.raises(OSError, match="simulated fsync failure"): + persistence._sync_directory(tmp_path) + + close_file.assert_called_once_with(17) + + +def test_sync_checkpoint_tree_requests_bottom_up_nonfollowing_walk( + tmp_path, + monkeypatch, +): + real_walk = os.walk + observed = {} + + def record_walk(path, **kwargs): + observed.update(kwargs) + return real_walk(path, **kwargs) + + monkeypatch.setattr(persistence.os, "walk", record_walk) + + persistence._sync_checkpoint_tree(tmp_path) + + assert observed == {"topdown": False, "followlinks": False} + + +@pytest.mark.parametrize("link_kind", ["file", "directory"]) +def test_atomic_checkpoint_rejects_symlinks_inside_staging(tmp_path, link_kind): + destination = tmp_path / "checkpoint" + outside = tmp_path / "outside" + if link_kind == "file": + outside.write_bytes(b"outside") + else: + outside.mkdir() + + with pytest.raises(OSError, match="not a regular file|contains a symlink"): + with persistence.atomic_checkpoint_directory(destination) as staging: + link = staging / "link" + link.symlink_to(outside, target_is_directory=link_kind == "directory") + + assert not destination.exists() + assert outside.exists() + + +def test_atomic_checkpoint_serializes_concurrent_complete_writers(tmp_path): + destination = tmp_path / "checkpoint" + destination.mkdir() + (destination / "generation").write_text("old", encoding="utf-8") + ready = Barrier(2) + + def write_generation(generation): + with persistence.atomic_checkpoint_directory(destination) as staging: + (staging / "generation").write_text(generation, encoding="utf-8") + (staging / "payload-a").write_text(generation, encoding="utf-8") + (staging / "payload-b").write_text(generation, encoding="utf-8") + ready.wait(timeout=5) + + with ThreadPoolExecutor(max_workers=2) as pool: + futures = [pool.submit(write_generation, generation) for generation in ("a", "b")] + for future in futures: + future.result(timeout=10) + + generation = (destination / "generation").read_text(encoding="utf-8") + assert generation in {"a", "b"} + assert (destination / "payload-a").read_text(encoding="utf-8") == generation + assert (destination / "payload-b").read_text(encoding="utf-8") == generation + assert list(tmp_path.glob(".checkpoint.staging-*")) == [] + assert list(tmp_path.glob(".checkpoint.backup-*")) == [] + + +def test_atomic_checkpoint_preserves_artifacts_owned_by_other_transactions(tmp_path): + destination = tmp_path / "checkpoint" + unrelated_staging = tmp_path / ".checkpoint.staging-operator-preserve" + unrelated_backup = tmp_path / ".checkpoint.backup-operator-preserve" + unrelated_staging.mkdir() + unrelated_backup.mkdir() + + with persistence.atomic_checkpoint_directory(destination) as staging: + (staging / "model.bin").write_bytes(b"saved") + + assert unrelated_staging.is_dir() + assert unrelated_backup.is_dir() + + +def test_rollback_rejects_ambiguous_destination_and_staging(tmp_path): + destination = tmp_path / "checkpoint" + staging = tmp_path / ".checkpoint.staging-owned" + backup = tmp_path / ".checkpoint.backup-owned" + destination.mkdir() + staging.mkdir() + backup.mkdir() + + with pytest.raises(RuntimeError) as exc_info: + persistence._rollback_checkpoint_commit( + destination, + staging, + backup, + had_destination=True, + ) + + assert str(exc_info.value) == ( + "Checkpoint rollback found both destination and staging; " + f"recover the previous checkpoint from {backup}" + ) + + +def test_rollback_with_no_backup_preserves_preexisting_destination(tmp_path): + destination = tmp_path / "checkpoint" + destination.mkdir() + sentinel = destination / "sentinel" + sentinel.write_text("preserve", encoding="utf-8") + staging = tmp_path / ".checkpoint.staging-missing" + + persistence._rollback_checkpoint_commit( + destination, + staging, + None, + had_destination=True, + ) + + assert sentinel.read_text(encoding="utf-8") == "preserve" + assert not staging.exists() diff --git a/tests/test_persistence_pipeline.py b/tests/test_persistence_pipeline.py index cdc29f3..b0ef1c2 100644 --- a/tests/test_persistence_pipeline.py +++ b/tests/test_persistence_pipeline.py @@ -29,6 +29,20 @@ def _pipeline(tmp_path: Path) -> AbliterationPipeline: return pipeline +def _write_valid_checkpoint( + checkpoint_dir: Path, + metadata_json: str, + payload: bytes = b"saved", +) -> None: + (checkpoint_dir / "model.safetensors").write_bytes(payload) + (checkpoint_dir / "config.json").write_text("{}", encoding="utf-8") + (checkpoint_dir / "tokenizer_config.json").write_text("{}", encoding="utf-8") + (checkpoint_dir / "abliteration_metadata.json").write_text( + metadata_json, + encoding="utf-8", + ) + + def test_rebirth_rejects_invalid_metadata_before_gathering_state(tmp_path): pipeline = _pipeline(tmp_path) destination = pipeline.output_dir @@ -89,7 +103,7 @@ def test_rebirth_ignores_non_os_disk_probe_failure_and_promotes_checkpoint( def write_checkpoint(checkpoint_dir, metadata_json, received_state_dict): assert json.loads(metadata_json) == {"schema": 1} assert received_state_dict is state_dict - (checkpoint_dir / "model.bin").write_bytes(b"saved") + _write_valid_checkpoint(checkpoint_dir, metadata_json) pipeline._write_local_checkpoint = MagicMock(side_effect=write_checkpoint) pipeline._free_gpu_memory = MagicMock() @@ -101,7 +115,46 @@ def test_rebirth_ignores_non_os_disk_probe_failure_and_promotes_checkpoint( monkeypatch.setattr(abliterate.shutil, "disk_usage", fail_probe) assert pipeline._rebirth() == pipeline.output_dir - assert (pipeline.output_dir / "model.bin").read_bytes() == b"saved" + assert (pipeline.output_dir / "model.safetensors").read_bytes() == b"saved" + pipeline._free_gpu_memory.assert_called_once_with() + pipeline._cleanup_offload_dir.assert_called_once_with() + + +def test_rebirth_rejects_truncated_checkpoint_then_retries_idempotently(tmp_path): + pipeline = _pipeline(tmp_path) + destination = pipeline.output_dir + destination.mkdir() + sentinel = destination / "sentinel" + sentinel.write_text("old", encoding="utf-8") + pipeline._build_metadata = MagicMock(return_value={"schema": 1}) + pipeline._gather_state_dict = MagicMock(return_value={"weight": torch.ones(1)}) + pipeline._free_gpu_memory = MagicMock() + pipeline._cleanup_offload_dir = MagicMock() + attempts = 0 + + def write_checkpoint(checkpoint_dir, metadata_json, _state_dict): + nonlocal attempts + attempts += 1 + _write_valid_checkpoint(checkpoint_dir, metadata_json) + if attempts == 1: + (checkpoint_dir / "abliteration_metadata.json").write_text( + "{", + encoding="utf-8", + ) + + pipeline._write_local_checkpoint = MagicMock(side_effect=write_checkpoint) + + with pytest.raises(ValueError, match="metadata is corrupt"): + pipeline._rebirth() + + assert sentinel.read_text(encoding="utf-8") == "old" + pipeline._free_gpu_memory.assert_not_called() + pipeline._cleanup_offload_dir.assert_not_called() + assert list(tmp_path.glob(".checkpoint.staging-*")) == [] + + assert pipeline._rebirth() == destination + assert (destination / "model.safetensors").read_bytes() == b"saved" + assert not sentinel.exists() pipeline._free_gpu_memory.assert_called_once_with() pipeline._cleanup_offload_dir.assert_called_once_with() @@ -118,8 +171,8 @@ def test_rebirth_pushes_only_after_local_checkpoint_promotion(tmp_path, monkeypa pipeline._free_gpu_memory = MagicMock() pipeline._cleanup_offload_dir = MagicMock() - def write_checkpoint(checkpoint_dir, _metadata_json, _state_dict): - (checkpoint_dir / "model.bin").write_bytes(b"saved") + def write_checkpoint(checkpoint_dir, metadata_json, _state_dict): + _write_valid_checkpoint(checkpoint_dir, metadata_json) pipeline._write_local_checkpoint = MagicMock(side_effect=write_checkpoint) api = MagicMock() @@ -130,7 +183,7 @@ def test_rebirth_pushes_only_after_local_checkpoint_promotion(tmp_path, monkeypa assert pipeline._rebirth() == pipeline.output_dir - assert (pipeline.output_dir / "model.bin").read_bytes() == b"saved" + assert (pipeline.output_dir / "model.safetensors").read_bytes() == b"saved" api_factory.assert_called_once_with(token="test-token") auto_name.assert_called_once_with( "test-model", @@ -159,7 +212,10 @@ def test_rebirth_uses_fallback_token_for_explicit_hub_destination( pipeline._build_metadata = MagicMock(return_value={"schema": 1}) pipeline._gather_state_dict = MagicMock(return_value={"weight": torch.ones(1)}) pipeline._write_local_checkpoint = MagicMock( - side_effect=lambda path, *_args: (path / "model.bin").write_bytes(b"saved"), + side_effect=lambda path, metadata_json, _state_dict: _write_valid_checkpoint( + path, + metadata_json, + ), ) pipeline._free_gpu_memory = MagicMock() pipeline._cleanup_offload_dir = MagicMock() diff --git a/tests/test_quality_policy.py b/tests/test_quality_policy.py index 31de0b8..9437976 100644 --- a/tests/test_quality_policy.py +++ b/tests/test_quality_policy.py @@ -45,6 +45,11 @@ def test_mutation_campaign_uses_fork_safe_native_runtime_policy(): assert '"obliteratus/analysis/numerical_contracts.py"' in mutmut_config assert '"obliteratus/analysis/whitened_svd.py"' in mutmut_config assert "required_mutation_targets" in mutmut_config + required_mutation_config = mutmut_config.split( + "required_mutation_targets = [", + maxsplit=1, + )[1].split("]", maxsplit=1)[0] + assert '"obliteratus/persistence_contracts.py"' in required_mutation_config assert '"obliteratus/reporting/report.py"' not in mutmut_config assert '"tests/test_runtime_contracts.py"' in mutmut_config assert '"tests/test_persistence_contracts.py"' in mutmut_config @@ -59,6 +64,9 @@ def test_mutation_campaign_uses_fork_safe_native_runtime_policy(): assert '"tests/test_telemetry_failure_contracts.py"' in Path( "scripts/run_repeat_gate.py", ).read_text() + repeat_config = Path("scripts/run_repeat_gate.py").read_text() + assert '"tests/test_checkpoint_atomicity.py"' in repeat_config + assert '"tests/test_persistence_pipeline.py"' in repeat_config assert mutation_env == { "BLIS_NUM_THREADS": "1", "MKL_NUM_THREADS": "1",