Files
OBLITERATUS/obliteratus/checkpoint_inspection.py

1128 lines
40 KiB
Python

"""Bounded, offline, structure-only checkpoint inventory and classification."""
from __future__ import annotations
import json
import os
import stat
from dataclasses import dataclass
from hashlib import sha256
from pathlib import Path, PurePosixPath
from typing import Any
from obliteratus.checkpoint_capabilities import AdapterRegistry
from obliteratus.checkpoint_errors import CheckpointContractError
_INT64_MAX = (1 << 63) - 1
_VERIFIER = "obliteratus.safe-structure-v1"
_DTYPE_BYTES = {
"BOOL": 1,
"U8": 1,
"I8": 1,
"F8_E4M3": 1,
"F8_E5M2": 1,
"I16": 2,
"U16": 2,
"F16": 2,
"BF16": 2,
"I32": 4,
"U32": 4,
"F32": 4,
"C64": 8,
"I64": 8,
"U64": 8,
"F64": 8,
"C128": 16,
}
@dataclass(frozen=True)
class InspectionLimits:
"""Explicit bounds for untrusted local structural inspection."""
max_files: int = 100_000
max_directories: int = 10_000
max_total_bytes: int = 1 << 40
max_json_bytes: int = 8 << 20
max_safetensors_header_bytes: int = 64 << 20
max_tensors: int = 2_000_000
hash_chunk_bytes: int = 1 << 20
def __post_init__(self) -> None:
for name, value in vars(self).items():
if type(value) is not int or value <= 0:
raise ValueError(f"{name} must be a positive integer")
@dataclass(frozen=True)
class _Fingerprint:
device: int
inode: int
mode: int
size: int
modified_ns: int
@classmethod
def from_stat(cls, value: os.stat_result) -> _Fingerprint:
return cls(
device=value.st_dev,
inode=value.st_ino,
mode=value.st_mode,
size=value.st_size,
modified_ns=value.st_mtime_ns,
)
@dataclass(frozen=True)
class _ObservedFile:
file_id: str
relative_path: str
path: Path
fingerprint: _Fingerprint
sha256: str
role: str
@dataclass(frozen=True)
class _ObservedDirectory:
relative_path: str
path: Path
fingerprint: _Fingerprint
@dataclass(frozen=True)
class _HeaderSummary:
tensor_names: tuple[str, ...]
tensor_count: int
logical_bytes: int
class _DuplicateJsonKey(ValueError):
pass
@dataclass(frozen=True)
class CheckpointInspection:
"""A strict descriptor produced without invoking any checkpoint reader."""
descriptor_id: str
primary_format: str
support_decision: str
_descriptor: dict[str, Any]
def to_dict(self) -> dict[str, Any]:
"""Return a detached JSON-compatible descriptor."""
return json.loads(json.dumps(self._descriptor, sort_keys=True, allow_nan=False))
def to_json(self) -> str:
"""Return deterministic JSON suitable for CLI and retained evidence."""
return json.dumps(
self._descriptor,
indent=2,
sort_keys=True,
ensure_ascii=True,
allow_nan=False,
) + "\n"
def _boundary(detail: str, *references: str) -> CheckpointContractError:
return CheckpointContractError(
"DCI_SOURCE_BOUNDARY_VIOLATION",
detail=detail,
affected_refs=references,
)
def _changed(*references: str) -> CheckpointContractError:
return CheckpointContractError(
"DCI_SOURCE_CHANGED",
detail="source_changed",
affected_refs=references,
)
def _resource(detail: str, *references: str) -> CheckpointContractError:
return CheckpointContractError(
"DCI_RESOURCE_LIMIT",
detail=detail,
affected_refs=references,
)
def _validation(detail: str, *references: str) -> CheckpointContractError:
return CheckpointContractError(
"DCI_VALIDATION_FAILED",
detail=detail,
affected_refs=references,
)
def _safe_relative(path: Path, root: Path, *, root_is_file: bool) -> str:
relative = path.name if root_is_file else path.relative_to(root).as_posix()
pure = PurePosixPath(relative)
if (
not relative
or pure.is_absolute()
or any(part in {"", ".", ".."} for part in pure.parts)
or len(relative) > 4096
):
raise _boundary("path_traversal")
return relative
def _role(relative_path: str) -> str:
name = PurePosixPath(relative_path).name
if name.endswith(".safetensors"):
return "tensor_payload"
if name.endswith(".index.json"):
return "weight_index"
if name in {"config.json", "adapter_config.json", "fsdp_metadata.json"}:
return "configuration"
if name in {"metadata.json", ".metadata", "universal_checkpoint_info.json"}:
return "producer_metadata"
if name.endswith((".bin", ".pt", ".pth", ".distcp")):
return "unsafe_serialization_or_shard"
return "other"
def _open_flags() -> int:
flags = os.O_RDONLY | getattr(os, "O_BINARY", 0)
flags |= getattr(os, "O_NOFOLLOW", 0)
return flags
def _matches(descriptor: int, expected: _Fingerprint) -> bool:
return _Fingerprint.from_stat(os.fstat(descriptor)) == expected
def _open_observed(path: Path, reference: str) -> int:
try:
return os.open(path, _open_flags())
except OSError as error:
raise _changed(reference) from error
def _hash_regular_file(
path: Path,
expected: _Fingerprint,
limits: InspectionLimits,
) -> str:
"""Hash one already-bounded regular file while detecting replacement or mutation."""
descriptor = _open_observed(path, path.name)
digest = sha256()
try:
if not _matches(descriptor, expected):
raise _changed(path.name)
while True:
chunk = os.read(descriptor, limits.hash_chunk_bytes)
if not chunk:
break
digest.update(chunk)
if not _matches(descriptor, expected):
raise _changed(path.name)
finally:
os.close(descriptor)
try:
if _Fingerprint.from_stat(path.lstat()) != expected:
raise _changed(path.name)
except FileNotFoundError as error:
raise _changed(path.name) from error
return f"sha256:{digest.hexdigest()}"
def _collect_paths(
source: Path,
limits: InspectionLimits,
) -> tuple[
list[tuple[Path, str, _Fingerprint]],
tuple[_ObservedDirectory, ...],
_Fingerprint,
]:
try:
root_stat = source.lstat()
except FileNotFoundError as error:
raise _boundary("source_missing") from error
if stat.S_ISLNK(root_stat.st_mode):
raise _boundary("source_symlink")
try:
if source.absolute() != source.resolve(strict=True):
raise _boundary("source_symlink")
except FileNotFoundError as error:
raise _changed("source") from error
root_fingerprint = _Fingerprint.from_stat(root_stat)
root_is_file = stat.S_ISREG(root_stat.st_mode)
if root_is_file:
if root_stat.st_size < 0 or root_stat.st_size > _INT64_MAX:
raise _resource("file_size")
if root_stat.st_size > limits.max_total_bytes:
raise _resource("max_total_bytes")
return [(source, source.name, root_fingerprint)], (), root_fingerprint
if not stat.S_ISDIR(root_stat.st_mode):
raise _boundary("source_special_file")
pending = [(source, ".", root_fingerprint)]
directory_count = 0
directories: list[_ObservedDirectory] = []
result: list[tuple[Path, str, _Fingerprint]] = []
total_bytes = 0
while pending:
directory, relative_directory, expected_directory = pending.pop()
directory_count += 1
if directory_count > limits.max_directories:
raise _resource("max_directories")
try:
current_directory = _Fingerprint.from_stat(directory.lstat())
if current_directory != expected_directory or not stat.S_ISDIR(
current_directory.mode
):
raise _changed(relative_directory)
with os.scandir(directory) as iterator:
entries = sorted(iterator, key=lambda entry: entry.name)
except OSError as error:
raise _boundary("source_unreadable") from error
directories.append(
_ObservedDirectory(relative_directory, directory, expected_directory)
)
for entry in entries:
path = Path(entry.path)
try:
observed = entry.stat(follow_symlinks=False)
except FileNotFoundError as error:
raise _changed(entry.name) from error
mode = observed.st_mode
relative = _safe_relative(path, source, root_is_file=False)
if stat.S_ISLNK(mode):
raise _boundary("source_symlink", relative)
if stat.S_ISDIR(mode):
pending.append((path, relative, _Fingerprint.from_stat(observed)))
continue
if not stat.S_ISREG(mode):
raise _boundary("source_special_file", relative)
if len(result) >= limits.max_files:
raise _resource("max_files")
if observed.st_size < 0 or observed.st_size > _INT64_MAX:
raise _resource("file_size")
if total_bytes > limits.max_total_bytes - observed.st_size:
raise _resource("max_total_bytes")
total_bytes += observed.st_size
result.append((path, relative, _Fingerprint.from_stat(observed)))
try:
if _Fingerprint.from_stat(directory.lstat()) != expected_directory:
raise _changed(relative_directory)
except FileNotFoundError as error:
raise _changed(relative_directory) from error
return (
sorted(result, key=lambda item: item[1]),
tuple(sorted(directories, key=lambda item: item.relative_path)),
root_fingerprint,
)
def _inventory(
source: Path,
limits: InspectionLimits,
) -> tuple[tuple[_ObservedFile, ...], tuple[_ObservedDirectory, ...], _Fingerprint]:
paths, directories, root_fingerprint = _collect_paths(source, limits)
observed_files = []
for index, (path, relative, fingerprint) in enumerate(paths, start=1):
digest = _hash_regular_file(path, fingerprint, limits)
observed_files.append(
_ObservedFile(
file_id=f"file-{index:06d}",
relative_path=relative,
path=path,
fingerprint=fingerprint,
sha256=digest,
role=_role(relative),
)
)
return tuple(observed_files), directories, root_fingerprint
def _read_bounded(file: _ObservedFile, maximum: int, limit_name: str) -> bytes:
if file.fingerprint.size > maximum:
raise _resource(limit_name, file.file_id)
descriptor = _open_observed(file.path, file.file_id)
try:
if not _matches(descriptor, file.fingerprint):
raise _changed(file.file_id)
chunks: list[bytes] = []
remaining = file.fingerprint.size
while remaining:
chunk = os.read(descriptor, min(remaining, 1 << 20))
if not chunk:
raise _changed(file.file_id)
chunks.append(chunk)
remaining -= len(chunk)
if os.read(descriptor, 1) or not _matches(descriptor, file.fingerprint):
raise _changed(file.file_id)
return b"".join(chunks)
finally:
os.close(descriptor)
def _read_json(file: _ObservedFile, limits: InspectionLimits) -> dict[str, Any]:
payload = _read_bounded(file, limits.max_json_bytes, "max_json_bytes")
try:
value = json.loads(
payload.decode("utf-8"),
object_pairs_hook=_reject_duplicate_json_keys,
)
except (UnicodeError, json.JSONDecodeError, _DuplicateJsonKey) as error:
raise _validation("json_invalid", file.file_id) from error
if not isinstance(value, dict):
raise _validation("json_object_required", file.file_id)
return value
def _reject_duplicate_json_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
result: dict[str, Any] = {}
for key, value in pairs:
if key in result:
raise _DuplicateJsonKey(key)
result[key] = value
return result
def _checked_tensor_bytes(dtype: str, shape: object, reference: str) -> int:
if dtype not in _DTYPE_BYTES or not isinstance(shape, list) or len(shape) > 32:
raise _validation("safetensors_header_invalid", reference)
elements = 1
for dimension in shape:
if type(dimension) is not int or dimension < 0 or dimension > _INT64_MAX:
raise _validation("safetensors_header_invalid", reference)
if dimension and elements > _INT64_MAX // dimension:
raise _validation("integer_overflow", reference)
elements *= dimension
width = _DTYPE_BYTES[dtype]
if elements and elements > _INT64_MAX // width:
raise _validation("integer_overflow", reference)
return elements * width
def _safetensors_header(file: _ObservedFile, limits: InspectionLimits) -> _HeaderSummary:
if file.fingerprint.size < 8:
raise _validation("safetensors_truncated", file.file_id)
descriptor = _open_observed(file.path, file.file_id)
try:
if not _matches(descriptor, file.fingerprint):
raise _changed(file.file_id)
length_bytes = os.read(descriptor, 8)
if len(length_bytes) != 8:
raise _validation("safetensors_truncated", file.file_id)
header_length = int.from_bytes(length_bytes, "little")
if header_length > limits.max_safetensors_header_bytes:
raise _resource("max_safetensors_header_bytes", file.file_id)
if header_length > file.fingerprint.size - 8:
raise _validation("safetensors_truncated", file.file_id)
header_bytes = b""
while len(header_bytes) < header_length:
chunk = os.read(descriptor, header_length - len(header_bytes))
if not chunk:
raise _validation("safetensors_truncated", file.file_id)
header_bytes += chunk
if not _matches(descriptor, file.fingerprint):
raise _changed(file.file_id)
finally:
os.close(descriptor)
try:
header = json.loads(
header_bytes.decode("utf-8"),
object_pairs_hook=_reject_duplicate_json_keys,
)
except (UnicodeError, json.JSONDecodeError, _DuplicateJsonKey) as error:
raise _validation("safetensors_header_invalid", file.file_id) from error
if not isinstance(header, dict):
raise _validation("safetensors_header_invalid", file.file_id)
entries = [(name, value) for name, value in header.items() if name != "__metadata__"]
if len(entries) > limits.max_tensors:
raise _resource("max_tensors", file.file_id)
names: list[str] = []
ranges: list[tuple[int, int]] = []
logical_bytes = 0
payload_bytes = file.fingerprint.size - 8 - header_length
for name, value in entries:
if not isinstance(name, str) or not name or not isinstance(value, dict):
raise _validation("safetensors_header_invalid", file.file_id)
offsets = value.get("data_offsets")
if (
not isinstance(offsets, list)
or len(offsets) != 2
or any(type(item) is not int or item < 0 for item in offsets)
):
raise _validation("safetensors_header_invalid", file.file_id)
begin, end = offsets
dtype = value.get("dtype")
if not isinstance(dtype, str):
raise _validation("safetensors_header_invalid", file.file_id)
expected_bytes = _checked_tensor_bytes(dtype, value.get("shape"), file.file_id)
if begin > end or end > payload_bytes or end - begin != expected_bytes:
raise _validation("safetensors_range_invalid", file.file_id)
names.append(name)
ranges.append((begin, end))
logical_bytes += expected_bytes
if logical_bytes > _INT64_MAX:
raise _validation("integer_overflow", file.file_id)
ranges.sort()
if any(left_end > right_start for (_, left_end), (right_start, _) in zip(ranges, ranges[1:])):
raise _validation("safetensors_range_overlap", file.file_id)
cursor = 0
for begin, end in ranges:
if begin != cursor:
raise _validation("safetensors_range_gap", file.file_id)
cursor = end
if cursor != payload_bytes:
raise _validation("safetensors_range_gap", file.file_id)
return _HeaderSummary(tuple(sorted(names)), len(names), logical_bytes)
def _revalidate(
source: Path,
root_fingerprint: _Fingerprint,
files: tuple[_ObservedFile, ...],
directories: tuple[_ObservedDirectory, ...],
limits: InspectionLimits,
) -> None:
try:
if _Fingerprint.from_stat(source.lstat()) != root_fingerprint:
raise _changed("source")
for file in files:
if _Fingerprint.from_stat(file.path.lstat()) != file.fingerprint:
raise _changed(file.file_id)
if _hash_regular_file(file.path, file.fingerprint, limits) != file.sha256:
raise _changed(file.file_id)
for directory in directories:
if _Fingerprint.from_stat(directory.path.lstat()) != directory.fingerprint:
raise _changed(directory.relative_path)
except FileNotFoundError as error:
raise _changed("source") from error
paths, observed_directories, observed_root = _collect_paths(source, limits)
expected_paths = tuple(
(file.relative_path, file.fingerprint) for file in files
)
actual_paths = tuple((relative, fingerprint) for _, relative, fingerprint in paths)
if observed_root != root_fingerprint or actual_paths != expected_paths:
raise _changed("source")
if tuple(
(directory.relative_path, directory.fingerprint)
for directory in observed_directories
) != tuple(
(directory.relative_path, directory.fingerprint) for directory in directories
):
raise _changed("source")
def _evidence(
evidence_id: str,
*,
subject: str,
kind: str,
file: _ObservedFile | None,
confidence: str,
location: str,
) -> dict[str, Any]:
return {
"evidence_id": evidence_id,
"subject": subject,
"kind": kind,
"file_ref": file.file_id if file is not None else None,
"location": location,
"confidence": confidence,
"verifier": _VERIFIER,
}
def _producer(name: str | None, evidence_refs: list[str]) -> dict[str, Any]:
return {
"name": name,
"version": None,
"format_version": None,
"evidence_refs": evidence_refs,
}
def _component(
component_id: str,
kind: str,
checkpoint_format: str,
producer_name: str | None,
scopes: list[str],
files: list[_ObservedFile],
evidence_refs: list[str],
) -> dict[str, Any]:
return {
"component_id": component_id,
"kind": kind,
"format": checkpoint_format,
"producer": _producer(producer_name, evidence_refs),
"state_scopes": scopes,
"topology_ref": None,
"inventory_refs": sorted(file.file_id for file in files),
"tensor_fragment_refs": [],
}
def _file_name_map(files: tuple[_ObservedFile, ...]) -> dict[str, _ObservedFile]:
result: dict[str, _ObservedFile] = {}
for item in files:
name = PurePosixPath(item.relative_path).name
if name in result:
raise _validation("duplicate_basename", result[name].file_id, item.file_id)
result[name] = item
return result
def _format_components(
files: tuple[_ObservedFile, ...],
limits: InspectionLimits,
) -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[CheckpointContractError], int, int]:
by_name = _file_name_map(files)
names = set(by_name)
evidence: list[dict[str, Any]] = []
errors: list[CheckpointContractError] = []
components: list[dict[str, Any]] = []
logical_bytes = 0
tensor_count = 0
hf_files: list[_ObservedFile] = []
hf_index = by_name.get("model.safetensors.index.json")
if hf_index is not None:
hf_files.append(hf_index)
if "model.safetensors" in names:
errors.append(
_validation(
"hf_layout_collision",
hf_index.file_id,
by_name["model.safetensors"].file_id,
)
)
try:
index = _read_json(hf_index, limits)
weight_map = index.get("weight_map")
if not isinstance(weight_map, dict) or not weight_map or len(weight_map) > limits.max_tensors:
raise _validation("hf_weight_map_invalid", hf_index.file_id)
referenced_names: set[str] = set()
for tensor_name, shard_name in weight_map.items():
if (
not isinstance(tensor_name, str)
or not tensor_name
or not isinstance(shard_name, str)
or PurePosixPath(shard_name).name != shard_name
):
raise _validation("hf_weight_map_invalid", hf_index.file_id)
referenced_names.add(shard_name)
shard_candidates = {
name for name in names if name.startswith("model-") and name.endswith(".safetensors")
}
if referenced_names != shard_candidates or not referenced_names <= names:
raise _validation("hf_weight_map_shard_mismatch", hf_index.file_id)
header_names: dict[str, set[str]] = {}
for shard_name in sorted(referenced_names):
shard = by_name[shard_name]
hf_files.append(shard)
summary = _safetensors_header(shard, limits)
header_names[shard_name] = set(summary.tensor_names)
tensor_count += summary.tensor_count
logical_bytes += summary.logical_bytes
if tensor_count > limits.max_tensors:
raise _resource("max_tensors", hf_index.file_id)
mapped_names = {
shard_name: {
tensor_name
for tensor_name, mapped_shard in weight_map.items()
if mapped_shard == shard_name
}
for shard_name in referenced_names
}
if any(
header_names[shard_name] != mapped_names[shard_name]
for shard_name in referenced_names
):
raise _validation("hf_weight_map_tensor_mismatch", hf_index.file_id)
except CheckpointContractError as error:
if error.code == "DCI_RESOURCE_LIMIT":
raise
errors.append(error)
elif "model.safetensors" in names:
model_file = by_name["model.safetensors"]
hf_files.append(model_file)
try:
summary = _safetensors_header(model_file, limits)
tensor_count += summary.tensor_count
logical_bytes += summary.logical_bytes
if tensor_count > limits.max_tensors:
raise _resource("max_tensors", model_file.file_id)
except CheckpointContractError as error:
if error.code == "DCI_RESOURCE_LIMIT":
raise
errors.append(error)
if hf_files:
evidence_id = "evidence-hf-safetensors"
evidence.append(
_evidence(
evidence_id,
subject="/components/hf_safetensors",
kind="header",
file=hf_files[0],
confidence="verified" if not errors else "inferred",
location=hf_files[0].relative_path,
)
)
components.append(
_component(
"component-model",
"model",
"hf_safetensors",
"Hugging Face",
["model_weights"],
hf_files,
[evidence_id],
)
)
peft_files: list[_ObservedFile] = []
if "adapter_model.safetensors" in names and "adapter_config.json" in names:
peft_files = [by_name["adapter_model.safetensors"], by_name["adapter_config.json"]]
try:
summary = _safetensors_header(peft_files[0], limits)
_read_json(peft_files[1], limits)
tensor_count += summary.tensor_count
logical_bytes += summary.logical_bytes
if tensor_count > limits.max_tensors:
raise _resource("max_tensors", peft_files[0].file_id)
except CheckpointContractError as error:
if error.code == "DCI_RESOURCE_LIMIT":
raise
errors.append(error)
evidence_id = "evidence-peft-safetensors"
evidence.append(
_evidence(
evidence_id,
subject="/components/peft_adapter",
kind="header",
file=peft_files[0],
confidence="verified" if not errors else "inferred",
location=peft_files[0].relative_path,
)
)
components.append(
_component(
"component-peft-adapter",
"peft_adapter",
"peft_safetensors",
"PEFT",
["adapter_weights"],
peft_files,
[evidence_id],
)
)
legacy_names = sorted(
name for name in names if name == "pytorch_model.bin" or name == "pytorch_model.bin.index.json"
)
if legacy_names:
selected = [by_name[name] for name in legacy_names]
evidence_id = "evidence-hf-pickle-filename"
evidence.append(
_evidence(
evidence_id,
subject="/components/hf_pytorch_pickle",
kind="filename",
file=selected[0],
confidence="inferred",
location=selected[0].relative_path,
)
)
components.append(
_component(
"component-model-pickle",
"model",
"hf_pytorch_pickle",
"Hugging Face",
["model_weights"],
selected,
[evidence_id],
)
)
fsdp_file = by_name.get("fsdp_metadata.json")
fsdp = False
if fsdp_file is not None:
try:
fsdp_record = _read_json(fsdp_file, limits)
fsdp = fsdp_record.get("state_dict_type") in {
"SHARDED_STATE_DICT",
"LOCAL_STATE_DICT",
"FULL_STATE_DICT",
}
except CheckpointContractError as error:
if error.code == "DCI_RESOURCE_LIMIT":
raise
errors.append(error)
dcp_files = [item for item in files if PurePosixPath(item.relative_path).name == ".metadata" or item.relative_path.endswith(".distcp")]
if fsdp:
assert fsdp_file is not None
selected = sorted({*dcp_files, fsdp_file}, key=lambda item: item.relative_path)
format_name, component_id = "fsdp_state_dict", "component-fsdp"
producer_name = "PyTorch FSDP"
elif dcp_files:
selected = dcp_files
format_name, component_id = "pytorch_dcp", "component-pytorch-dcp"
producer_name = "PyTorch"
else:
selected = []
format_name = component_id = producer_name = ""
if selected:
evidence_id = f"evidence-{format_name}"
evidence.append(
_evidence(
evidence_id,
subject=f"/components/{format_name}",
kind="filename" if not fsdp else "json",
file=selected[0],
confidence="declared" if fsdp else "inferred",
location=selected[0].relative_path,
)
)
components.append(
_component(
component_id,
"unknown",
format_name,
producer_name,
["unknown_state"],
selected,
[evidence_id],
)
)
metadata_file = by_name.get("metadata.json")
megatron = False
if metadata_file is not None:
try:
metadata = _read_json(metadata_file, limits)
megatron = metadata.get("sharded_backend") == "torch_dist"
except CheckpointContractError as error:
if error.code == "DCI_RESOURCE_LIMIT":
raise
errors.append(error)
if megatron:
assert metadata_file is not None
selected = [metadata_file] + [by_name[name] for name in sorted(names) if name == "common.pt"]
evidence_id = "evidence-megatron-torch-dist"
evidence.append(
_evidence(
evidence_id,
subject="/components/megatron_torch_dist",
kind="json",
file=metadata_file,
confidence="declared",
location="metadata.json#/sharded_backend",
)
)
components.append(
_component(
"component-megatron",
"unknown",
"megatron_torch_dist",
"Megatron-Core",
["unknown_state"],
selected,
[evidence_id],
)
)
universal = by_name.get("universal_checkpoint_info.json")
zero_files = [item for item in files if "zero_pp_rank_" in PurePosixPath(item.relative_path).name]
if universal is not None:
try:
_read_json(universal, limits)
except CheckpointContractError as error:
if error.code == "DCI_RESOURCE_LIMIT":
raise
errors.append(error)
selected = [universal, *zero_files]
ds_format = "deepspeed_universal"
elif zero_files:
selected = zero_files
ds_format = "deepspeed_zero"
else:
selected = []
ds_format = ""
if selected:
evidence_id = f"evidence-{ds_format}"
evidence.append(
_evidence(
evidence_id,
subject=f"/components/{ds_format}",
kind="json" if universal is not None else "filename",
file=selected[0],
confidence="declared" if universal is not None else "inferred",
location=selected[0].relative_path,
)
)
components.append(
_component(
"component-deepspeed",
"unknown",
ds_format,
"DeepSpeed",
["unknown_state"],
selected,
[evidence_id],
)
)
return components, evidence, errors, tensor_count, logical_bytes
def _primary(components: list[dict[str, Any]], errors: list[CheckpointContractError]) -> tuple[str, str]:
formats = [component["format"] for component in components]
if not formats:
return "unknown", "unknown"
if any(error.detail.endswith("layout_collision") for error in errors):
return "ambiguous", "inferred"
if len(formats) > 1:
return "ambiguous", "inferred"
if errors:
return formats[0], "inferred"
if formats[0] in {"hf_safetensors", "peft_safetensors"}:
return formats[0], "verified"
if formats[0] in {"fsdp_state_dict", "megatron_torch_dist", "deepspeed_universal"}:
return formats[0], "declared"
return formats[0], "inferred"
def _classification_blocker(primary_format: str) -> CheckpointContractError:
return CheckpointContractError(
"DCI_UNSUPPORTED_FORMAT_OR_VERSION",
detail="ambiguous_layout" if primary_format == "ambiguous" else "unknown_layout",
affected_refs=("source",),
)
def _trust_blocker(components: list[dict[str, Any]]) -> CheckpointContractError:
return CheckpointContractError(
"DCI_TRUST_POLICY_REQUIRED",
detail="payload_or_vendor_metadata_required",
affected_refs=(component["component_id"] for component in components),
)
def _descriptor_digest(value: object) -> str:
payload = json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False).encode()
return sha256(payload).hexdigest()
def _inspect_checkpoint(
source: Path | str,
*,
limits: InspectionLimits | None = None,
adapter_registry: AdapterRegistry | None = None,
) -> CheckpointInspection:
"""Inspect one local source using only inert filenames, JSON, and safetensors headers."""
active_limits = limits or InspectionLimits()
active_registry = adapter_registry or AdapterRegistry()
source_path = Path(source)
files, directories, root_fingerprint = _inventory(source_path, active_limits)
components, evidence, probe_errors, tensor_count, logical_bytes = _format_components(
files,
active_limits,
)
primary_format, confidence = _primary(components, probe_errors)
if not components:
evidence.append(
_evidence(
"evidence-unknown-layout",
subject="/primary_format",
kind="derived",
file=None,
confidence="unknown",
location="bounded inventory contains no recognized structural signature",
)
)
components = [
_component(
"component-unknown",
"unknown",
"unknown",
None,
["unknown_state"],
list(files),
["evidence-unknown-layout"],
)
]
formats = {component["format"] for component in components}
trusted_formats = {
"hf_pytorch_pickle",
"pytorch_dcp",
"fsdp_state_dict",
"megatron_torch_dist",
"megatron_torch_dcp",
"megatron_fsdp_dtensor",
"deepspeed_zero",
"deepspeed_universal",
}
trust_required = bool(formats & trusted_formats)
blockers = list(probe_errors)
if primary_format in {"unknown", "ambiguous"}:
blockers.append(_classification_blocker(primary_format))
elif trust_required:
blockers.append(_trust_blocker(components))
if primary_format == "ambiguous":
adapter_resolution = {
"status": "ambiguous",
"adapter_id": None,
"adapter_version": None,
"capability_digest": None,
"reason": "Ambiguous structure cannot select a capability.",
}
elif trust_required:
resolved_adapter = active_registry.resolve(primary_format)
adapter_resolution = resolved_adapter.to_dict()
if adapter_resolution["status"] == "ambiguous":
blockers.append(_classification_blocker("ambiguous"))
elif resolved_adapter.dependency_unavailable:
blockers.append(
CheckpointContractError(
"DCI_TRUST_RUNTIME_UNAVAILABLE",
detail="optional_dependency_missing_or_incompatible",
affected_refs=(str(adapter_resolution["adapter_id"]),),
)
)
else:
adapter_resolution = {
"status": "not_required" if len(formats) == 1 else "unsupported",
"adapter_id": None,
"adapter_version": None,
"capability_digest": None,
"reason": (
"Canonical safetensors needs no producer adapter."
if len(formats) == 1
else "No producer capability applies to this structure."
),
}
if blockers:
support_decision = (
"trusted_inspection_required"
if (
trust_required
and len(formats) == 1
and adapter_resolution["status"] != "ambiguous"
)
else "blocked"
)
else:
support_decision = "canonical_hf_ready"
scopes = sorted({scope for component in components for scope in component["state_scopes"]})
state_classification = (
"weights_only"
if set(scopes) <= {"model_weights", "adapter_weights"} and "unknown_state" not in scopes
else "unknown"
)
unsafe_findings = []
if formats & trusted_formats:
unsafe_findings.append("unsafe or opaque serialization requires a separately approved reader")
inventory_records = [
{
"file_id": file.file_id,
"relative_path": file.relative_path,
"role": file.role,
"size_bytes": file.fingerprint.size,
"sha256": file.sha256,
"regular_file": True,
"observation_id": f"observation-{_descriptor_digest([file.relative_path, file.fingerprint.size, file.sha256])[:24]}",
}
for file in files
]
total_source_bytes = sum(file.fingerprint.size for file in files)
inventory_core = {
"files": inventory_records,
"total_bytes": total_source_bytes,
"observation_complete": True,
}
inventory_id = f"inventory-{_descriptor_digest(inventory_core)[:32]}"
_revalidate(source_path, root_fingerprint, files, directories, active_limits)
unique_evidence = {item["evidence_id"]: item for item in evidence}
producer_names = {component["producer"]["name"] for component in components}
producer_name = next(iter(producer_names)) if len(producer_names) == 1 else None
producer_evidence = sorted(unique_evidence)
descriptor: dict[str, Any] = {
"schema_id": "obliteratus.checkpoint-descriptor",
"schema_version": "1.0.0",
"descriptor_id": "pending",
"primary_format": primary_format,
"classification_confidence": confidence,
"components": sorted(components, key=lambda item: item["component_id"]),
"producer": _producer(producer_name, producer_evidence),
"evidence": [unique_evidence[key] for key in sorted(unique_evidence)],
"source_inventory": {"inventory_id": inventory_id, **inventory_core},
"safety": {
"inspection_level": "safe_structure",
"trust_required": trust_required,
"inventory_revalidated": True,
"unsafe_serialization_findings": unsafe_findings,
"violations": [error.detail for error in probe_errors],
},
"state": {
"observed_scopes": scopes,
"classification": state_classification,
},
"topologies": [],
"tensor_fragments": [],
"adapter_resolution": adapter_resolution,
"conversion_plan": {
"eligible": False,
"target_format": "hf_safetensors",
"state_scope": "weights_only",
"dropped_scopes": sorted(scope for scope in scopes if scope not in {"model_weights", "adapter_weights"}),
},
"resource_estimate": {
"source_bytes": total_source_bytes,
"logical_bytes": logical_bytes,
"output_bytes": logical_bytes,
"temporary_bytes": 0,
"peak_ram_bytes": min(
total_source_bytes,
max(active_limits.max_json_bytes, active_limits.max_safetensors_header_bytes),
),
"peak_vram_bytes": 0,
"file_count": len(files),
"tensor_count": tensor_count,
"shard_count": sum(1 for file in files if file.relative_path.endswith((".safetensors", ".distcp", ".bin", ".pt"))),
"assumptions": [
"Structure-only estimates do not authorize payload loading or conversion."
],
"confidence": "verified" if tensor_count and not probe_errors else "unknown",
"admission": "unknown",
},
"support_decision": support_decision,
"blockers": [error.to_blocker() for error in blockers],
}
descriptor_id = f"descriptor-{_descriptor_digest({key: value for key, value in descriptor.items() if key != 'descriptor_id'})[:32]}"
descriptor["descriptor_id"] = descriptor_id
return CheckpointInspection(
descriptor_id=descriptor_id,
primary_format=primary_format,
support_decision=support_decision,
_descriptor=descriptor,
)
def inspect_checkpoint(
source: Path | str,
*,
limits: InspectionLimits | None = None,
adapter_registry: AdapterRegistry | None = None,
) -> CheckpointInspection:
"""Inspect one local source and expose only stable fail-closed errors."""
try:
return _inspect_checkpoint(
source,
limits=limits,
adapter_registry=adapter_registry,
)
except CheckpointContractError:
raise
except OSError as error:
raise _changed("source") from error