Nucleus-Resynthesis / runtime /src /resynthesis /training_throughput.py
Wl6adams's picture
Add portable Release 188 generation runtime
919fd68 verified
Raw
History Blame Contribute Delete
128 kB
"""Throughput scheduling and receipt-owned compact NoNE page growth.
The training helpers preserve every corpus row and every validation row. Page
growth creates transfer-initialized candidates only; it never advances an
accepted generation and never calls an initialized page trained.
"""
from __future__ import annotations
import hashlib
import json
import math
import os
import shutil
import struct
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Iterable, Mapping, cast
import torch
from resynthesis.none_paging import (
DEFAULT_GPU_PAGE_CACHE_ENTRIES,
NoNEAcceptedTrainingSaturationPacket,
NoNECompactTransferTemplate,
NoNEImmutablePageStore,
NoNEPageObjectBinding,
NoNEScaleCohortPacket,
NoNEScaleEvidencePacket,
NoNEPageWeights,
PAGE_STORE_LOCATOR_SCHEMA,
SCALED_FLOAT8_TRANSFER_STORAGE,
_default_page_store_registry_roots_boundary,
_page_store_registry_root_for_boundary,
_session_key,
digest_tensor,
file_sha256_authority_batch_boundary,
plan_none_scale_cohort,
validate_none_accepted_training_saturation_boundary,
validate_page_weights,
)
ACTIVE_FORWARD_PARAM_BUDGET_APPROX = 14_000_000
COMPACT_TRANSFER_JOURNAL_SCHEMA = (
"nnf.resynthesis.compact_transfer_page_journal.v1"
)
COMPACT_TRANSFER_SUMMARY_SCHEMA = (
"nnf.resynthesis.compact_transfer_page_summary.v1"
)
COMPACT_TRANSFER_LOCATOR_SCHEMA = (
"nnf.resynthesis.compact_transfer_page_locator.v1"
)
COMPACT_TRANSFER_REGISTRY_DIRECTORY = "compact-transfer-banks"
TEMPLATE_REFERENCE_TRANSFER_STORAGE = (
"template_reference_scaled_float8_e4m3fn_implicit_zero_optimizer_v1"
)
TRAINING_THROUGHPUT_LANE_SCHEMA = (
"nnf.resynthesis.training_throughput_lane.v1"
)
FAILURE_CATEGORY_ORDER: tuple[str, ...] = (
"stop_overrun_after_exact",
"correction_transfer_failure",
"uncertainty_disposition",
"source_join_chemistry",
"present_no_provider_checksum",
"general_knowledge_miss",
"unknown",
)
@dataclass(frozen=True)
class ThroughputLaneConfig:
"""Successor-lane training geometry, never a grading-surface cap."""
# Wide retained windows + large residency: physical pages only become live
# model weight when they stay GPU-resident long enough to be routed. The
# defaults below favor traversal of the physical bank; promotion still
# requires held-out / lineage proof and does not relabel storage as trained.
training_updates_per_validation: int = 512
training_microbatch_rows: int = 8
# Validation automatically bisects a batch on OOM, so start wide enough to
# amortize parent/RBO traversal and ledger boundaries on 96 GiB lanes.
validation_microbatch_rows: int = 64
training_emission_backward_chunk_requested: int = 2
training_emission_backward_chunk: int = 2
# These are external diagnostic bounds, not production defaults. ``None``
# proves that the complete prompt and answer sequence participate.
training_target_tokens_per_row: int | None = None
training_prompt_window_tokens: int | None = None
training_task_wall_clock_seconds: float = 45.0
# Default off: compile is a post-stability accelerator. Dynamo currently
# fights dynamic page/sequence geometry; eager traversal of the physical
# page bank is the faster path to first optimizer commits.
compile_science_stack: bool = False
# Explicit canary only. The bulk completion objective performs its
# full-vocabulary low-rank projections in float32 before the stable
# logsumexp reduction. Hopper can execute those same float32 matmuls
# through TF32 tensor cores while retaining float32 accumulation and the
# complete target/vocabulary surface. Keep the ordinary lane at
# ``highest`` until a durable rate comparison proves this useful.
tf32_training_matmul: bool = False
gpu_resident_page_cache_entries: int = DEFAULT_GPU_PAGE_CACHE_ENTRIES
def complete_sequence_throughput_lane_boundary(value: object) -> bool:
"""Validate uncapped prompt/target participation at a receipt boundary."""
if not isinstance(value, Mapping):
return False
return (
value.get("schema") == TRAINING_THROUGHPUT_LANE_SCHEMA
and value.get("completeTargetSequenceUsed") is True
and value.get("completePromptSequenceUsed") is True
and value.get("trainingTargetTokensPerRow") is None
and value.get("trainingPromptWindowTokens") is None
)
@dataclass(frozen=True)
class CompactPageStreamPacket:
"""Verified append-only capacity observed at an external I/O boundary.
``trained_t`` and ``accepted_generation_committed_t`` remain explicit so
discovery cannot promote transfer initialization into a capability claim.
"""
store_root: Path
summary_path: Path
journal_path: Path
storage_format: str
session_id_t: torch.Tensor
page_ids_t: torch.Tensor
object_sha256_t: torch.Tensor
object_bytes_t: torch.Tensor
trained_t: torch.Tensor
accepted_generation_committed_t: torch.Tensor
@dataclass(frozen=True)
class CompactPageBankBinding:
"""Verified boundary identity for an untrained compact-page bank.
Object sizes and mappings are checked for the complete bank during
discovery. Object payload hashes are checked when a model-selected cohort
is admitted so discovery does not reread the complete physical bank.
Writer-source digests are observation-only diagnostics and never form
part of the bank's authority.
"""
summary_path: Path
journal_path: Path
store_root: Path
summary_sha256_t: torch.Tensor
journal_sha256_t: torch.Tensor
session_id_t: torch.Tensor
source_checkpoint_sha256_t: torch.Tensor
writer_source_sha256s_t: torch.Tensor
page_ids_t: torch.Tensor
object_sha256s_t: torch.Tensor
object_bytes_t: torch.Tensor
page_parameter_elements_t: torch.Tensor
physical_parameter_elements_t: torch.Tensor
total_object_bytes_t: torch.Tensor
storage_format: str
template_object_path: Path | None
template_object_sha256_t: torch.Tensor
template_page_id_t: torch.Tensor
template_page_id_offset_t: torch.Tensor
physical_archive_object_bytes_t: torch.Tensor
ready_t: torch.Tensor
@dataclass(frozen=True)
class CompactPageAdmissionPacket:
"""Tensor-owned mapping from model-selected families to physical pages.
Parent-family page identities may repeat so one retained family can own
multiple distinct child pages in the same evidence-bound transaction.
"""
bank_summary_path: Path
bank_journal_path: Path
bank_store_root: Path
bank_storage_format: str
bank_template_object_path: Path | None
bank_summary_sha256_t: torch.Tensor
bank_journal_sha256_t: torch.Tensor
bank_session_id_t: torch.Tensor
bank_source_checkpoint_sha256_t: torch.Tensor
bank_writer_source_sha256s_t: torch.Tensor
selected_family_page_ids_t: torch.Tensor
selected_page_ids_t: torch.Tensor
selected_object_sha256s_t: torch.Tensor
selected_object_bytes_t: torch.Tensor
bank_page_count_t: torch.Tensor
page_parameter_elements_t: torch.Tensor
bank_physical_parameter_elements_t: torch.Tensor
bank_total_object_bytes_t: torch.Tensor
bank_template_object_sha256_t: torch.Tensor
bank_template_page_id_t: torch.Tensor
bank_template_page_id_offset_t: torch.Tensor
bank_physical_archive_object_bytes_t: torch.Tensor
selected_page_count_t: torch.Tensor
ready_t: torch.Tensor
@dataclass(frozen=True)
class DiscoveredCompactAdmissionPlan:
"""Auto-discovered physical bank bound to one model-owned scale cohort."""
bank: CompactPageBankBinding
scale_cohort: NoNEScaleCohortPacket
admission: CompactPageAdmissionPacket
def throughput_lane_config_from_env() -> ThroughputLaneConfig:
"""Read performance geometry without changing model-owned routing."""
def _pos_int(name: str, default: int) -> int:
raw = os.environ.get(name, "").strip()
if not raw:
return default
value = int(raw)
if value < 1:
raise ValueError(f"{name} must be >= 1")
return value
def _nonneg_int(name: str, default: int) -> int:
raw = os.environ.get(name, "").strip()
if not raw:
return default
value = int(raw)
if value < 0:
raise ValueError(f"{name} must be >= 0")
return value
def _optional_pos_int(name: str) -> int | None:
raw = os.environ.get(name, "").strip()
if not raw:
return None
value = int(raw)
if value < 1:
raise ValueError(f"{name} must be >= 1")
return value
def _pos_float(name: str, default: float) -> float:
raw = os.environ.get(name, "").strip()
if not raw:
return default
value = float(raw)
if value <= 0.0:
raise ValueError(f"{name} must be > 0")
return value
def _bool(name: str) -> bool:
raw = os.environ.get(name, "").strip()
if not raw:
return False
if raw in {"1", "true", "True", "yes"}:
return True
if raw in {"0", "false", "False", "no"}:
return False
raise ValueError(f"{name} must be a boolean")
compile_flag = _bool("NNF_RESYNTHESIS_COMPILE_SCIENCE")
tf32_training_matmul = _bool("NNF_RESYNTHESIS_TF32_TRAINING_MATMUL")
requested_emission_backward_chunk = _pos_int(
"NNF_RESYNTHESIS_EMISSION_BACKWARD_CHUNK",
2,
)
return ThroughputLaneConfig(
training_updates_per_validation=_pos_int(
"NNF_RESYNTHESIS_TRAINING_UPDATES_PER_VALIDATION",
512,
),
training_microbatch_rows=_pos_int(
"NNF_RESYNTHESIS_TRAINING_MICROBATCH_ROWS",
8,
),
validation_microbatch_rows=_pos_int(
"NNF_RESYNTHESIS_VALIDATION_MICROBATCH_ROWS",
64,
),
training_emission_backward_chunk_requested=(
requested_emission_backward_chunk
),
# The signed caller request owns the grouping geometry. Silently
# shrinking it creates a misleading throughput receipt and prevents a
# clean-lane proof from ever exercising the requested traversal. An
# unsafe request therefore fails at the real training boundary rather
# than being rewritten by a host-side fixed cap.
training_emission_backward_chunk=requested_emission_backward_chunk,
training_target_tokens_per_row=_optional_pos_int(
"NNF_RESYNTHESIS_TRAIN_TARGET_TOKENS_PER_ROW"
),
training_prompt_window_tokens=_optional_pos_int(
"NNF_RESYNTHESIS_TRAIN_PROMPT_WINDOW_TOKENS"
),
training_task_wall_clock_seconds=_pos_float(
"NNF_RESYNTHESIS_TRAINING_TASK_WALL_CLOCK_SEC",
45.0,
),
compile_science_stack=compile_flag,
tf32_training_matmul=tf32_training_matmul,
gpu_resident_page_cache_entries=_pos_int(
"NNF_RESYNTHESIS_GPU_PAGE_CACHE_ENTRIES",
DEFAULT_GPU_PAGE_CACHE_ENTRIES,
),
)
def configure_training_matmul_precision_boundary(
lane: ThroughputLaneConfig,
) -> tuple[str, bool]:
"""Install one launch-owned float32 matmul policy before model loading.
This is a process-global PyTorch execution setting, so it is established
exactly once at the external CLI boundary before any model or optimizer
tensors are constructed. It changes only CUDA float32 matmul execution;
target participation, loss geometry, routing, and durable cursor authority
remain unchanged.
"""
requested_precision = (
"high" if lane.tf32_training_matmul else "highest"
)
torch.set_float32_matmul_precision(requested_precision)
effective_precision = torch.get_float32_matmul_precision()
tf32_activated = bool(torch.backends.cuda.matmul.allow_tf32)
if effective_precision != requested_precision:
raise RuntimeError("training float32 matmul precision differs")
if tf32_activated != lane.tf32_training_matmul:
raise RuntimeError("training TF32 matmul activation differs")
return effective_precision, tf32_activated
def failure_category_for_row(row: dict[str, Any]) -> str:
"""Map a training-boundary row to a curriculum priority bucket."""
explicit = row.get("failureCategory")
if isinstance(explicit, str) and explicit.strip():
return explicit.strip()
axis = row.get("generalizationAxis") or row.get("generalization_axis")
group = str(
row.get("generalizationGroup") or row.get("generalization_group") or ""
)
if axis == "adaptive_source_join":
return "source_join_chemistry"
if "uncertainty" in group.lower() or "unsupported" in group.lower():
return "uncertainty_disposition"
if row.get("stopOverrunAfterExact") is True:
return "stop_overrun_after_exact"
if row.get("correctionTransferFailure") is True:
return "correction_transfer_failure"
return "unknown"
def prioritize_train_rows_by_failure_category(
rows: list[dict[str, Any]],
*,
baseline_grade_rows: Iterable[dict[str, Any]] | None = None,
) -> list[dict[str, Any]]:
"""Order all rows by observed weakness without dropping any row."""
overrun_ids: set[str] = set()
correction_ids: set[str] = set()
if baseline_grade_rows is not None:
for grade in baseline_grade_rows:
qid = grade.get("questionId") or grade.get("record_id")
if qid is None:
continue
key = str(qid)
audit = grade.get("generationAuditTrace")
if isinstance(audit, dict):
first_exact = audit.get("firstExactAnswerEmissionIndex")
after = audit.get("tokensAfterFirstExactAnswerBoundary")
if (
isinstance(first_exact, int)
and first_exact >= 0
and isinstance(after, int)
and after > 0
and not grade.get("passed")
):
overrun_ids.add(key)
if grade.get("passed") is False and grade.get("attempt") == 2:
correction_ids.add(key)
def sort_key(row: dict[str, Any]) -> tuple[int, str]:
qid = str(row.get("question_id") or row.get("record_id") or "")
category = failure_category_for_row(row)
if qid in overrun_ids:
category = "stop_overrun_after_exact"
elif qid in correction_ids:
category = "correction_transfer_failure"
try:
rank = FAILURE_CATEGORY_ORDER.index(category)
except ValueError:
rank = len(FAILURE_CATEGORY_ORDER)
return rank, qid
ordered = sorted(rows, key=sort_key)
if len(ordered) != len(rows):
raise RuntimeError("curriculum ordering dropped training rows")
return ordered
def group_train_rows_into_microbatches(
rows: list[dict[str, Any]],
*,
microbatch_rows: int,
) -> list[list[dict[str, Any]]]:
"""Pack every row into geometry-owned microbatches exactly once."""
if microbatch_rows < 1:
raise ValueError("microbatch_rows must be >= 1")
batches = [
rows[cursor : cursor + microbatch_rows]
for cursor in range(0, len(rows), microbatch_rows)
]
if sum(len(batch) for batch in batches) != len(rows):
raise RuntimeError("microbatch construction dropped training rows")
return batches
def load_baseline_grade_rows(path: str) -> list[dict[str, Any]]:
"""Load pre-update grade rows from an execution-ledger boundary."""
grades: list[dict[str, Any]] = []
with open(path, encoding="utf-8") as handle:
for line in handle:
line = line.strip()
if not line:
continue
try:
row = json.loads(line)
except json.JSONDecodeError:
continue
if row.get("schema") != "nnf.resynthesis.grade.v1":
continue
if row.get("evaluationPhase") != "pre_update_baseline":
continue
grades.append(row)
return grades
def _science_compile_recurrent_types() -> tuple[type[torch.nn.Module], ...]:
"""Return recurrent module types that never enter the compiled region."""
from resynthesis.kda_expert import CRConditionedKDAExpert
return (
torch.nn.RNN,
torch.nn.LSTM,
torch.nn.GRU,
torch.nn.RNNCell,
torch.nn.LSTMCell,
torch.nn.GRUCell,
CRConditionedKDAExpert,
)
def _science_compile_targets(
stack: torch.nn.Module,
) -> tuple[
tuple[tuple[str, torch.nn.Module], ...],
tuple[tuple[str, torch.nn.Module], ...],
]:
"""Select graph-heavy, non-recurrent kernels from the science stack."""
from resynthesis.science_layers import IntentContextPivotAttention
recurrent_types = _science_compile_recurrent_types()
excluded_recurrent = tuple(
(name, module)
for name, module in stack.named_modules()
if name and isinstance(module, recurrent_types)
)
selected: list[tuple[str, torch.nn.Module]] = []
for name, module in stack.named_modules():
# DeltaBlockAttnRes selects a model-owned depth between one and
# max_blocks. Compiling that dynamic-depth module produced shape
# recompiles and eventually a detached output on the live r152 B2
# canary. Keep it eager; only the fixed-geometry attention kernel is a
# supported compile target.
if not isinstance(module, IntentContextPivotAttention):
continue
if any(
nested is not module and isinstance(nested, recurrent_types)
for nested in module.modules()
):
continue
selected.append((name, module))
return tuple(selected), excluded_recurrent
def _science_compile_canary_forward_backward(
module: torch.nn.Module,
*,
compiled: bool,
) -> tuple[
torch.Tensor,
torch.Tensor,
tuple[torch.Tensor | None, ...],
]:
"""Run one real forward/backward without touching parameter gradients."""
from resynthesis.delta_attn_res import DeltaBlockAttnRes
from resynthesis.science_layers import IntentContextPivotAttention
# Scalar control parameters intentionally remain fp32 while the expensive
# projection weights may be bf16 on the live training lane. The canary
# input must follow the first projection that consumes it, not whichever
# direct scalar happens to be first in Module.parameters(); otherwise a
# valid mixed-precision model fails before torch.compile is even exercised.
if isinstance(module, IntentContextPivotAttention):
reference_t = module.q_proj.weight
elif isinstance(module, DeltaBlockAttnRes):
reference_t = module.query_proj.weight
else:
raise RuntimeError("science compile canary target is unsupported")
if reference_t.device.type == "meta":
raise RuntimeError("science compile canary has no materialized parameter")
trainable_parameters = tuple(
parameter
for _name, parameter in module.named_parameters()
if parameter.requires_grad
)
# Paged NoNE intentionally freezes the dense science kernels while the
# external page/RBO/Fabric pathway remains trainable through their input.
# Such a kernel still participates in backward and is a valid compile
# target: the contract to prove is its exact input gradient. Parameter
# gradients are additionally compared only when this particular runtime
# leaves local parameters trainable.
dtype = (
reference_t.dtype
if reference_t.dtype.is_floating_point
else torch.float32
)
hidden_size = int(getattr(module, "hidden_size", 0))
glyph_dim = int(getattr(module, "glyph_dim", 0))
if hidden_size < 1 or glyph_dim < 1:
raise RuntimeError("science compile canary geometry is invalid")
hidden_t = torch.ones(
1,
2,
hidden_size,
device=reference_t.device,
dtype=dtype,
requires_grad=True,
)
glyph_t = torch.ones(
1,
2,
glyph_dim,
device=reference_t.device,
dtype=dtype,
)
forward = module if compiled else module.forward
if isinstance(module, IntentContextPivotAttention):
output_t = forward(
hidden_t,
intent_glyph_context=glyph_t,
action_glyph_context=glyph_t,
relation_glyph_context=glyph_t,
)
elif isinstance(module, DeltaBlockAttnRes):
output_t = forward(
hidden_t,
delta_bank_t=hidden_t.detach().unsqueeze(0),
intent_glyph_t=glyph_t,
relation_glyph_t=glyph_t,
relation_bank_t=glyph_t.detach().unsqueeze(0),
)
if (
not isinstance(output_t, torch.Tensor)
or output_t.shape != hidden_t.shape
or not output_t.isfinite().all()
or not output_t.requires_grad
):
raise RuntimeError("science compile forward canary failed")
gradients = torch.autograd.grad(
output_t.float().square().mean(),
(hidden_t, *trainable_parameters),
retain_graph=False,
create_graph=False,
allow_unused=True,
)
gradient_t = gradients[0]
parameter_gradients = gradients[1:]
if (
gradient_t is None
or gradient_t.shape != hidden_t.shape
or not gradient_t.isfinite().all()
or (
bool(trainable_parameters)
and not any(
parameter_gradient_t is not None
for parameter_gradient_t in parameter_gradients
)
)
or any(
parameter_gradient_t is not None
and not parameter_gradient_t.isfinite().all()
for parameter_gradient_t in parameter_gradients
)
):
raise RuntimeError("science compile backward canary failed")
return (
output_t.detach(),
gradient_t.detach(),
tuple(
parameter_gradient_t.detach()
if parameter_gradient_t is not None
else None
for parameter_gradient_t in parameter_gradients
),
)
def _named_identity_stable(
expected: tuple[tuple[str, object], ...],
active: tuple[tuple[str, object], ...],
) -> bool:
return len(expected) == len(active) and all(
expected_name == active_name and expected_value is active_value
for (expected_name, expected_value), (
active_name,
active_value,
) in zip(expected, active, strict=True)
)
def _optimizer_parameter_names_snapshot(group: dict[str, Any]) -> object:
names = group.get("param_names")
return tuple(names) if isinstance(names, (list, tuple)) else names
def maybe_compile_science_stack(
model: Any,
*,
optimizer: torch.optim.Optimizer | None = None,
) -> bool:
"""Compile proven non-recurrent science kernels after optimizer restore.
``nn.Module.compile`` is lazy, so registration alone is not activation.
Every selected kernel must execute an eager and compiled forward/backward
canary with matching outputs/gradients before the receipt can call the
compiler active. GRU, RNN, and recurrent KDA modules stay eager.
"""
stack = getattr(model, "science_stack", None)
if not isinstance(stack, torch.nn.Module):
return False
if getattr(model, "_science_stack_compiled", False):
return True
model._science_stack_compiled = False
model._science_stack_compile_canary_passed = False
model._science_stack_compile_failure = None
model._science_stack_optimizer_parameter_names_stable = False
if not torch.cuda.is_available():
model._science_stack_compile_failure = "CUDA is unavailable"
return False
selected, excluded_recurrent = _science_compile_targets(stack)
model._science_stack_compile_target_count = len(selected)
model._science_stack_compile_excluded_recurrent_count = len(
excluded_recurrent
)
model._science_stack_compile_target_names = tuple(
name for name, _module in selected
)
model._science_stack_compile_excluded_recurrent_names = tuple(
name for name, _module in excluded_recurrent
)
if not selected:
model._science_stack_compile_failure = (
"no supported non-recurrent science kernels were found"
)
return False
module_identity = tuple(model.named_modules())
parameter_identity = tuple(model.named_parameters())
buffer_identity = tuple(model.named_buffers())
optimizer_group_identity = (
tuple(
(
group,
tuple(group["params"]),
_optimizer_parameter_names_snapshot(group),
)
for group in optimizer.param_groups
)
if optimizer is not None
else ()
)
optimizer_state_identity = (
tuple((parameter, state) for parameter, state in optimizer.state.items())
if optimizer is not None
else ()
)
recurrent_compile_identity = tuple(
(module, getattr(module, "_compiled_call_impl", None))
for _name, module in excluded_recurrent
)
target_compile_identity = tuple(
(module, getattr(module, "_compiled_call_impl", None))
for _name, module in selected
)
parameter_gradient_identity = tuple(
(
parameter,
parameter.grad,
parameter.grad.detach().clone()
if parameter.grad is not None
else None,
)
for _name, parameter in parameter_identity
)
delta_depth_weight_identity = tuple(
(
module,
getattr(module, "last_depth_weights_t", None),
)
for _name, module in selected
if hasattr(module, "last_depth_weights_t")
)
compile_error: Exception | None = None
try:
import torch._functorch.config as _functorch_config
_functorch_config.donated_buffer = False
for _name, target in selected:
(
eager_output_t,
eager_gradient_t,
eager_parameter_gradients,
) = (
_science_compile_canary_forward_backward(
target,
compiled=False,
)
)
target.compile(
mode="default",
fullgraph=False,
)
(
compiled_output_t,
compiled_gradient_t,
compiled_parameter_gradients,
) = (
_science_compile_canary_forward_backward(
target,
compiled=True,
)
)
torch.testing.assert_close(
compiled_output_t,
eager_output_t,
rtol=5.0e-2,
atol=5.0e-3,
)
torch.testing.assert_close(
compiled_gradient_t,
eager_gradient_t,
rtol=5.0e-2,
atol=5.0e-3,
)
if (
len(compiled_parameter_gradients)
!= len(eager_parameter_gradients)
or any(
(compiled_parameter_gradient_t is None)
!= (eager_parameter_gradient_t is None)
for (
compiled_parameter_gradient_t,
eager_parameter_gradient_t,
) in zip(
compiled_parameter_gradients,
eager_parameter_gradients,
strict=True,
)
)
):
raise RuntimeError(
"science compile parameter-gradient topology differs"
)
for (
compiled_parameter_gradient_t,
eager_parameter_gradient_t,
) in zip(
compiled_parameter_gradients,
eager_parameter_gradients,
strict=True,
):
if (
compiled_parameter_gradient_t is not None
and eager_parameter_gradient_t is not None
):
torch.testing.assert_close(
compiled_parameter_gradient_t,
eager_parameter_gradient_t,
rtol=5.0e-2,
atol=5.0e-3,
)
except Exception as error:
compile_error = error
finally:
for delta_module, prior_depth_weights_t in (
delta_depth_weight_identity
):
setattr(
delta_module,
"last_depth_weights_t",
prior_depth_weights_t,
)
active_module_identity = tuple(model.named_modules())
active_parameter_identity = tuple(model.named_parameters())
active_buffer_identity = tuple(model.named_buffers())
optimizer_identity_stable = bool(
optimizer is None
or (
len(optimizer_group_identity) == len(optimizer.param_groups)
and all(
expected_group is active_group
and len(expected_parameters) == len(active_group["params"])
and expected_parameter_names
== _optimizer_parameter_names_snapshot(active_group)
and all(
expected_parameter is active_parameter
for expected_parameter, active_parameter in zip(
expected_parameters,
active_group["params"],
strict=True,
)
)
for (
expected_group,
expected_parameters,
expected_parameter_names,
), active_group in zip(
optimizer_group_identity,
optimizer.param_groups,
strict=True,
)
)
and len(optimizer_state_identity) == len(optimizer.state)
and all(
optimizer.state.get(parameter) is state
for parameter, state in optimizer_state_identity
)
)
)
recurrent_identity_stable = all(
getattr(module, "_compiled_call_impl", None) is compiled_call
for module, compiled_call in recurrent_compile_identity
)
gradient_identity_stable = all(
parameter.grad is prior_gradient
and (
prior_gradient_value is None
or (
parameter.grad is not None
and torch.equal(parameter.grad, prior_gradient_value)
)
)
for parameter, prior_gradient, prior_gradient_value in (
parameter_gradient_identity
)
)
identity_stable = bool(
getattr(model, "science_stack", None) is stack
and _named_identity_stable(module_identity, active_module_identity)
and _named_identity_stable(parameter_identity, active_parameter_identity)
and _named_identity_stable(buffer_identity, active_buffer_identity)
and optimizer_identity_stable
and recurrent_identity_stable
and gradient_identity_stable
)
if not identity_stable:
for target, compiled_call in target_compile_identity:
target._compiled_call_impl = compiled_call
model._science_stack_compile_failure = (
"science-stack compilation changed model or optimizer identity"
)
raise RuntimeError(
"science-stack compilation changed model, parameter, buffer, "
"gradient, or optimizer identity"
) from compile_error
if compile_error is not None:
for target, compiled_call in target_compile_identity:
target._compiled_call_impl = compiled_call
model._science_stack_compile_failure = (
f"{type(compile_error).__name__}: {compile_error}"
)
return False
model._science_stack_optimizer_parameter_names_stable = bool(
optimizer is not None and optimizer_identity_stable
)
model._science_stack_compile_canary_passed = True
model._science_stack_compiled = True
return True
def estimate_page_storage_bytes(
*,
page_parameter_elements: int,
page_count: int,
bytes_per_element: int = 2,
) -> int:
"""Return the physical model-weight storage envelope for a page cohort."""
if page_parameter_elements < 1 or page_count < 0 or bytes_per_element < 1:
raise ValueError("page storage geometry is malformed")
return page_parameter_elements * page_count * bytes_per_element
def bulk_transfer_init_pages(
*,
source_gate_t: torch.Tensor,
source_up_t: torch.Tensor,
source_down_t: torch.Tensor,
source_glyph_down_t: torch.Tensor,
source_glyph_up_t: torch.Tensor,
objects_root: Path,
start_page_id: int,
count: int,
router_size: int = 8,
batch_size: int = 100,
) -> list[tuple[int, str]]:
"""Persist transfer-initialized pages without making a trained claim.
This compatibility boundary restores the original bulk-catalog API. New
growth uses :func:`write_compact_transfer_pages`; both surfaces retain the
page ID in the immutable payload and store objects below ``sha256/``.
"""
from safetensors.torch import save_file
if start_page_id < 0 or count < 0 or router_size < 1 or batch_size < 1:
raise ValueError("bulk transfer page geometry is malformed")
if source_gate_t.shape != source_up_t.shape:
raise ValueError("bulk transfer gate/up geometry differs")
shard_root = Path(objects_root).expanduser().resolve() / "sha256"
shard_root.mkdir(parents=True, exist_ok=True)
gate_t = source_gate_t.detach().cpu().contiguous()
up_t = source_up_t.detach().cpu().contiguous()
down_t = source_down_t.detach().cpu().contiguous()
glyph_down_t = source_glyph_down_t.detach().cpu().contiguous()
glyph_up_t = source_glyph_up_t.detach().cpu().contiguous()
flat_parameter_count = sum(
tensor.numel()
for tensor in (gate_t, up_t, down_t, glyph_down_t, glyph_up_t)
) + 1 + 3 * router_size
results: list[tuple[int, str]] = []
for page_id in range(start_page_id, start_page_id + count):
payload = {
"format_revision_t": torch.full((1,), 2, dtype=torch.long),
"page_ids_t": torch.full((1,), page_id, dtype=torch.long),
"ffn_mode_t": torch.zeros(1, 1, dtype=gate_t.dtype),
"gate_t": gate_t.unsqueeze(0),
"up_t": up_t.unsqueeze(0),
"down_t": down_t.unsqueeze(0),
"glyph_down_t": glyph_down_t.unsqueeze(0),
"glyph_up_t": glyph_up_t.unsqueeze(0),
"translation_gate_t": torch.zeros(1, 1, dtype=gate_t.dtype),
"outcome_memory_t": torch.zeros(1, router_size, dtype=gate_t.dtype),
"repair_memory_t": torch.zeros(1, router_size, dtype=gate_t.dtype),
"transfer_memory_t": torch.zeros(1, router_size, dtype=gate_t.dtype),
"optimizer_mean_t": torch.zeros(1, flat_parameter_count),
"optimizer_square_t": torch.zeros(1, flat_parameter_count),
"step_t": torch.zeros(1, dtype=torch.long),
}
temporary = shard_root / f".bulk_page.{os.getpid()}.{page_id}.tmp"
temporary.unlink(missing_ok=True)
save_file(payload, str(temporary))
with temporary.open("rb") as handle:
os.fsync(handle.fileno())
object_sha256 = _file_sha256(temporary)
object_path = shard_root / f"{object_sha256}.safetensors"
if object_path.is_file():
if _file_sha256(object_path) != object_sha256:
raise RuntimeError("existing bulk page object differs")
temporary.unlink()
else:
os.replace(temporary, object_path)
results.append((page_id, object_sha256))
return results
def page_count_for_physical_target(
*,
target_parameter_elements: int,
current_parameter_elements: int,
page_parameter_elements: int,
) -> int:
"""Derive needed physical pages from observed model and page geometry."""
if target_parameter_elements < 1:
raise ValueError("physical parameter target must be positive")
if current_parameter_elements < 0:
raise ValueError("current physical parameters cannot be negative")
if page_parameter_elements < 1:
raise ValueError("page parameter geometry must be positive")
remaining = max(0, target_parameter_elements - current_parameter_elements)
return math.ceil(remaining / page_parameter_elements)
def _atomic_json(path: Path, payload: Mapping[str, Any]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
temporary = path.with_suffix(path.suffix + f".{os.getpid()}.tmp")
temporary.write_text(
json.dumps(payload, sort_keys=True, indent=2) + "\n",
encoding="utf-8",
)
with temporary.open("rb") as handle:
os.fsync(handle.fileno())
os.replace(temporary, path)
def _file_sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(8 * 1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def _fsync_directory(path: Path) -> None:
descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
try:
os.fsync(descriptor)
finally:
os.close(descriptor)
def _digest_hex(digest_t: torch.Tensor) -> str:
values = digest_t.detach().cpu().to(dtype=torch.uint8).reshape(-1).tolist()
return bytes(values).hex()
def compact_transfer_weights_from_source(
*,
page_id: int,
source_gate_t: torch.Tensor,
source_up_t: torch.Tensor,
source_down_t: torch.Tensor,
source_glyph_down_t: torch.Tensor,
source_glyph_up_t: torch.Tensor,
router_size: int,
) -> NoNEPageWeights:
"""Build one zero-residual, untrained page without optimizer moments."""
if page_id < 0 or router_size < 1:
raise ValueError("page identity and router geometry must be nonnegative")
dtype = source_gate_t.dtype
weights = NoNEPageWeights(
page_ids_t=torch.tensor([page_id], dtype=torch.long),
ffn_mode_t=torch.zeros(1, 1, dtype=dtype),
gate_t=source_gate_t.detach().cpu().unsqueeze(0).contiguous(),
up_t=source_up_t.detach().cpu().unsqueeze(0).contiguous(),
down_t=source_down_t.detach().cpu().unsqueeze(0).contiguous(),
glyph_down_t=(
source_glyph_down_t.detach().cpu().unsqueeze(0).contiguous()
),
glyph_up_t=source_glyph_up_t.detach().cpu().unsqueeze(0).contiguous(),
translation_gate_t=torch.zeros(1, 1, dtype=dtype),
outcome_memory_t=torch.zeros(1, router_size, dtype=dtype),
repair_memory_t=torch.zeros(1, router_size, dtype=dtype),
transfer_memory_t=torch.zeros(1, router_size, dtype=dtype),
)
validate_page_weights(weights)
return weights
def _read_compact_journal(
path: Path,
*,
source_sha256: str,
session_id: list[int],
) -> dict[int, dict[str, Any]]:
if not path.is_file():
return {}
records: dict[int, dict[str, Any]] = {}
with path.open(encoding="utf-8") as handle:
for line_number, line in enumerate(handle, start=1):
line = line.strip()
if not line:
continue
row = json.loads(line)
page_id = row.get("pageId")
if (
row.get("schema") != COMPACT_TRANSFER_JOURNAL_SCHEMA
or row.get("sourceCheckpointSha256") != source_sha256
or row.get("sessionId") != session_id
or row.get("trained") is not False
or row.get("acceptedGenerationCommitted") is not False
or not isinstance(page_id, int)
or isinstance(page_id, bool)
or page_id in records
):
raise RuntimeError(
f"compact page journal differs at line {line_number}"
)
records[page_id] = row
return records
def _binding_from_record(record: Mapping[str, Any]) -> NoNEPageObjectBinding:
page_id = record.get("pageId")
sha256 = record.get("objectSha256")
object_bytes = record.get("objectBytes")
if (
not isinstance(page_id, int)
or isinstance(page_id, bool)
or not isinstance(sha256, str)
or len(sha256) != 64
or not isinstance(object_bytes, int)
or isinstance(object_bytes, bool)
or object_bytes < 1
):
raise RuntimeError("compact page journal object identity is malformed")
return NoNEPageObjectBinding(
page_id_t=torch.tensor(page_id, dtype=torch.long),
object_sha256_t=digest_tensor(sha256),
object_bytes_t=torch.tensor(object_bytes, dtype=torch.long),
)
def _validated_sha256_bytes(value: object, *, field: str) -> bytes:
if not isinstance(value, str) or len(value) != 64:
raise RuntimeError(f"{field} is not a SHA-256 digest")
try:
raw = bytes.fromhex(value)
except ValueError as exc:
raise RuntimeError(f"{field} is not a SHA-256 digest") from exc
if len(raw) != 32 or value != value.lower():
raise RuntimeError(f"{field} is not a canonical SHA-256 digest")
return raw
def _diagnostic_sha256_bytes(value: object) -> bytes | None:
"""Decode an optional diagnostic digest without granting it authority."""
if (
not isinstance(value, str)
or len(value) != 64
or value != value.lower()
or any(character not in "0123456789abcdef" for character in value)
):
return None
return bytes.fromhex(value)
def _positive_int(value: object, *, field: str) -> int:
if not isinstance(value, int) or isinstance(value, bool) or value < 1:
raise RuntimeError(f"{field} must be a positive integer")
return value
def _nonnegative_int(value: object, *, field: str) -> int:
if not isinstance(value, int) or isinstance(value, bool) or value < 0:
raise RuntimeError(f"{field} must be a nonnegative integer")
return value
def _template_page_id_offset_from_payload_boundary(payload: bytearray) -> int:
"""Return the sole signed page-identity slot from one safetensors object."""
if len(payload) < 8:
raise RuntimeError("compact template object is truncated")
header_bytes = int(struct.unpack_from("<Q", payload, 0)[0])
header_end = 8 + header_bytes
if header_end > len(payload):
raise RuntimeError("compact template header escapes payload")
try:
header = json.loads(payload[8:header_end])
except json.JSONDecodeError as exc:
raise RuntimeError("compact template header is malformed") from exc
page_record = header.get("page_ids_t") if isinstance(header, dict) else None
offsets = page_record.get("data_offsets") if isinstance(page_record, dict) else None
if (
not isinstance(offsets, list)
or len(offsets) != 2
or not all(isinstance(value, int) for value in offsets)
or offsets[1] - offsets[0] != 8
):
raise RuntimeError("compact template page identity offset differs")
offset_values = cast(list[int], offsets)
page_id_offset = header_end + int(offset_values[0])
if page_id_offset < header_end or page_id_offset + 8 > len(payload):
raise RuntimeError("compact template page identity escaped payload")
return page_id_offset
def _template_reference_payload_boundary(
*,
template_path: Path,
template_sha256: str,
template_bytes: int,
template_page_id: int,
page_id_offset: int,
) -> NoNECompactTransferTemplate:
"""Load one immutable transfer template and verify its identity slot.
Template-reference banks retain the original byte-for-byte transfer object
once. Each virtual page is reconstructed only by changing this isolated
signed 64-bit identity slot, then checked against its journaled digest.
"""
if (
not template_path.is_file()
or template_path.stat().st_size != template_bytes
or _file_sha256(template_path) != template_sha256
):
raise RuntimeError("compact template object identity differs")
payload = bytearray(template_path.read_bytes())
if len(payload) != template_bytes:
raise RuntimeError("compact template object is truncated")
observed_offset = _template_page_id_offset_from_payload_boundary(payload)
if (
observed_offset != page_id_offset
or struct.unpack_from("<q", payload, page_id_offset)[0]
!= template_page_id
):
raise RuntimeError("compact template page identity differs")
return NoNECompactTransferTemplate(
payload=payload,
page_id_offset=page_id_offset,
)
def _template_reference_fields_boundary(
*,
payload: Mapping[str, Any],
records: Mapping[int, Mapping[str, Any]],
objects_root: Path,
) -> tuple[Path, bytes, int, int, int]:
"""Validate the single retained object that backs virtual page identities."""
template_digest = _validated_sha256_bytes(
payload.get("templateObjectSha256"),
field="templateObjectSha256",
)
template_bytes = _positive_int(
payload.get("templateObjectBytes"),
field="templateObjectBytes",
)
template_page_id = _nonnegative_int(
payload.get("templatePageId"),
field="templatePageId",
)
page_id_offset = _nonnegative_int(
payload.get("templatePageIdOffset"),
field="templatePageIdOffset",
)
archive_bytes = _positive_int(
payload.get("physicalArchiveObjectBytes"),
field="physicalArchiveObjectBytes",
)
row = records.get(template_page_id)
if (
row is None
or row.get("objectSha256") != template_digest.hex()
or row.get("objectBytes") != template_bytes
or archive_bytes != template_bytes
):
raise RuntimeError("compact template reference journal identity differs")
template_path = objects_root / f"{template_digest.hex()}.safetensors"
_template_reference_payload_boundary(
template_path=template_path,
template_sha256=template_digest.hex(),
template_bytes=template_bytes,
template_page_id=template_page_id,
page_id_offset=page_id_offset,
)
return (
template_path,
template_digest,
template_page_id,
page_id_offset,
archive_bytes,
)
def discover_compact_transfer_page_bank(
summary_path: Path,
) -> CompactPageBankBinding:
"""Discover and validate a complete untrained compact-page bank.
The summary and journal are content-addressed. A conventional bank keeps
one object per page; a template-reference bank keeps one immutable object
plus every expected page digest. Both defer full page hashing to the
selected-cohort materialization boundary.
"""
resolved_summary = summary_path.expanduser().resolve()
if not resolved_summary.is_file():
raise FileNotFoundError(resolved_summary)
payload = json.loads(resolved_summary.read_text(encoding="utf-8"))
if not isinstance(payload, dict):
raise RuntimeError("compact page summary is not an object")
requested_count = _positive_int(
payload.get("requestedPageCount"),
field="requestedPageCount",
)
start_page_id = _nonnegative_int(
payload.get("startPageId"),
field="startPageId",
)
page_parameter_elements = _positive_int(
payload.get("pageParameterElements"),
field="pageParameterElements",
)
total_object_bytes = _positive_int(
payload.get("objectBytes"),
field="objectBytes",
)
session_values = payload.get("sessionId")
writer_values = payload.get("writerSourceSha256s", [])
journal_value = payload.get("journalPath")
store_value = payload.get("storeRoot")
storage_format = payload.get("storageFormat")
if (
payload.get("schema") != COMPACT_TRANSFER_SUMMARY_SCHEMA
or payload.get("passed") is not True
or storage_format
not in {
SCALED_FLOAT8_TRANSFER_STORAGE,
TEMPLATE_REFERENCE_TRANSFER_STORAGE,
}
or payload.get("completePageMappingRetained") is not True
or payload.get("acceptedGenerationCommitted") is not False
or payload.get("promotionRequiresTrainingAndHeldoutProof") is not True
or payload.get("trainedPageCount") != 0
or payload.get("journaledPageCount") != requested_count
or payload.get("transferInitializedPageCount") != requested_count
or payload.get("physicalParameterElementsInitialized")
!= requested_count * page_parameter_elements
or not isinstance(journal_value, str)
or not journal_value
or not isinstance(store_value, str)
or not store_value
or not isinstance(session_values, list)
or len(session_values) != 4
or any(
not isinstance(value, int) or isinstance(value, bool)
for value in session_values
)
):
raise RuntimeError("compact page summary authority differs")
source_digest = _validated_sha256_bytes(
payload.get("sourceCheckpointSha256"),
field="sourceCheckpointSha256",
)
writer_digests = tuple(
dict.fromkeys(
digest
for value in writer_values
for digest in (_diagnostic_sha256_bytes(value),)
if digest is not None
)
) if isinstance(writer_values, list) else ()
recorded_journal = Path(journal_value).expanduser()
recorded_store = Path(store_value).expanduser()
resolved_store = (
recorded_store.resolve()
if recorded_store.is_dir()
else resolved_summary.parent.resolve()
)
resolved_journal = (
recorded_journal.resolve()
if recorded_journal.is_file()
else (resolved_store / recorded_journal.name).resolve()
)
if not resolved_journal.is_file() or not resolved_store.is_dir():
raise RuntimeError("compact page bank storage is incomplete")
journal_sha256 = _file_sha256(resolved_journal)
expected_journal_digest = _validated_sha256_bytes(
payload.get("journalSha256"),
field="journalSha256",
)
if journal_sha256 != expected_journal_digest.hex():
raise RuntimeError("compact page journal SHA-256 differs")
records = _read_compact_journal(
resolved_journal,
source_sha256=source_digest.hex(),
session_id=session_values,
)
requested_ids = tuple(range(start_page_id, start_page_id + requested_count))
if tuple(sorted(records)) != requested_ids:
raise RuntimeError("compact page journal mapping is incomplete")
object_digests = bytearray()
object_sizes: list[int] = []
observed_object_digests: set[bytes] = set()
objects_root = resolved_store / "objects" / "sha256"
if not objects_root.is_dir():
raise RuntimeError("compact page bank object root is absent")
template_path: Path | None = None
template_digest_t = torch.zeros(0, dtype=torch.uint8)
template_page_id_t = torch.zeros(0, dtype=torch.long)
template_page_id_offset_t = torch.zeros(0, dtype=torch.long)
physical_archive_object_bytes = total_object_bytes
if storage_format == TEMPLATE_REFERENCE_TRANSFER_STORAGE:
(
template_path,
template_digest,
template_page_id,
template_page_id_offset,
physical_archive_object_bytes,
) = _template_reference_fields_boundary(
payload=payload,
records=records,
objects_root=objects_root,
)
template_digest_t = torch.tensor(list(template_digest), dtype=torch.uint8)
template_page_id_t = torch.tensor(template_page_id, dtype=torch.long)
template_page_id_offset_t = torch.tensor(
template_page_id_offset,
dtype=torch.long,
)
for page_id in requested_ids:
row = records[page_id]
object_digest = _validated_sha256_bytes(
row.get("objectSha256"),
field=f"page[{page_id}].objectSha256",
)
object_bytes = _positive_int(
row.get("objectBytes"),
field=f"page[{page_id}].objectBytes",
)
object_path = objects_root / f"{object_digest.hex()}.safetensors"
if (
row.get("storageFormat") != SCALED_FLOAT8_TRANSFER_STORAGE
or object_digest in observed_object_digests
or (
storage_format == SCALED_FLOAT8_TRANSFER_STORAGE
and (
not object_path.is_file()
or object_path.stat().st_size != object_bytes
)
)
or (
storage_format == TEMPLATE_REFERENCE_TRANSFER_STORAGE
and template_path is not None
and object_bytes != template_path.stat().st_size
)
):
raise RuntimeError(
f"compact page object mapping differs for page {page_id}"
)
observed_object_digests.add(object_digest)
object_digests.extend(object_digest)
object_sizes.append(object_bytes)
if sum(object_sizes) != total_object_bytes:
raise RuntimeError("compact page bank object-byte total differs")
object_sha256s_t = torch.frombuffer(
object_digests,
dtype=torch.uint8,
).clone().reshape(requested_count, 32)
writer_sha256s_t = (
torch.tensor(
[list(value) for value in writer_digests],
dtype=torch.uint8,
)
if writer_digests
else torch.zeros((0, 32), dtype=torch.uint8)
)
return CompactPageBankBinding(
summary_path=resolved_summary,
journal_path=resolved_journal,
store_root=resolved_store,
summary_sha256_t=digest_tensor(_file_sha256(resolved_summary)),
journal_sha256_t=digest_tensor(journal_sha256),
session_id_t=torch.tensor(session_values, dtype=torch.long),
source_checkpoint_sha256_t=torch.tensor(
list(source_digest),
dtype=torch.uint8,
),
writer_source_sha256s_t=writer_sha256s_t,
page_ids_t=torch.arange(
start_page_id,
start_page_id + requested_count,
dtype=torch.long,
),
object_sha256s_t=object_sha256s_t,
object_bytes_t=torch.tensor(object_sizes, dtype=torch.long),
page_parameter_elements_t=torch.tensor(
page_parameter_elements,
dtype=torch.long,
),
physical_parameter_elements_t=torch.tensor(
requested_count * page_parameter_elements,
dtype=torch.long,
),
total_object_bytes_t=torch.tensor(total_object_bytes, dtype=torch.long),
storage_format=str(storage_format),
template_object_path=template_path,
template_object_sha256_t=template_digest_t,
template_page_id_t=template_page_id_t,
template_page_id_offset_t=template_page_id_offset_t,
physical_archive_object_bytes_t=torch.tensor(
physical_archive_object_bytes,
dtype=torch.long,
),
ready_t=torch.ones((), dtype=torch.bool),
)
def compact_transfer_page_bank_to_template_reference(
*,
source_summary_path: Path,
output_root: Path,
) -> dict[str, Any]:
"""Archive a fully verified untrained bank as one template plus identities.
Every source payload is rehashed before publication. The resulting bank
preserves the complete page-to-digest journal and can later hydrate only
a model-selected cohort into ordinary immutable page objects. It neither
moves an accepted generation nor makes a trained-capability claim.
"""
source_bank = discover_compact_transfer_page_bank(source_summary_path)
if source_bank.storage_format != SCALED_FLOAT8_TRANSFER_STORAGE:
raise RuntimeError("template-reference compaction requires full objects")
resolved_output = output_root.expanduser().resolve()
resolved_output.parent.mkdir(parents=True, exist_ok=True)
if resolved_output.exists():
raise FileExistsError(resolved_output)
temporary_root = resolved_output.parent / (
f".{resolved_output.name}.template_reference_pending"
)
if temporary_root.exists():
raise FileExistsError(temporary_root)
source_summary = json.loads(
source_bank.summary_path.read_text(encoding="utf-8")
)
if not isinstance(source_summary, dict):
raise RuntimeError("compact source summary is not an object")
session_values = source_bank.session_id_t.detach().cpu().long().tolist()
source_checkpoint_sha256 = bytes(
source_bank.source_checkpoint_sha256_t.detach().cpu().tolist()
).hex()
records = _read_compact_journal(
source_bank.journal_path,
source_sha256=source_checkpoint_sha256,
session_id=session_values,
)
page_ids = source_bank.page_ids_t.detach().cpu().long().tolist()
if not page_ids or tuple(page_ids) != tuple(sorted(records)):
raise RuntimeError("compact source page mapping differs")
template_page_id = page_ids[0]
template_row = records[template_page_id]
template_sha256 = str(template_row["objectSha256"])
template_bytes = int(template_row["objectBytes"])
template_source = (
source_bank.store_root
/ "objects"
/ "sha256"
/ f"{template_sha256}.safetensors"
)
if _file_sha256(template_source) != template_sha256:
raise RuntimeError("compact source template hash differs")
template_payload = bytearray(template_source.read_bytes())
page_id_offset = _template_page_id_offset_from_payload_boundary(template_payload)
if struct.unpack_from("<q", template_payload, page_id_offset)[0] != template_page_id:
raise RuntimeError("compact source template page identity differs")
# Verify all old payloads and all virtual reconstructions before retaining
# only the template. This is intentionally exhaustive: storage release
# has a higher bar than candidate-capacity discovery.
for page_id in page_ids:
row = records[page_id]
expected_sha256 = str(row["objectSha256"])
expected_bytes = int(row["objectBytes"])
source_object = (
source_bank.store_root
/ "objects"
/ "sha256"
/ f"{expected_sha256}.safetensors"
)
if (
source_object.stat().st_size != expected_bytes
or _file_sha256(source_object) != expected_sha256
):
raise RuntimeError("compact source page hash differs")
struct.pack_into("<q", template_payload, page_id_offset, page_id)
if hashlib.sha256(template_payload).hexdigest() != expected_sha256:
raise RuntimeError("compact virtual page reconstruction differs")
struct.pack_into(
"<q",
template_payload,
page_id_offset,
template_page_id,
)
if hashlib.sha256(template_payload).hexdigest() != template_sha256:
raise RuntimeError("compact template restoration differs")
temporary_objects = temporary_root / "objects" / "sha256"
temporary_objects.mkdir(parents=True, exist_ok=False)
template_destination = temporary_objects / template_source.name
if template_source.stat().st_dev == temporary_objects.stat().st_dev:
os.link(template_source, template_destination)
else:
shutil.copyfile(template_source, template_destination)
with template_destination.open("rb") as handle:
os.fsync(handle.fileno())
temporary_journal = temporary_root / "pages.jsonl"
shutil.copyfile(source_bank.journal_path, temporary_journal)
with temporary_journal.open("rb") as handle:
os.fsync(handle.fileno())
_fsync_directory(temporary_objects)
_fsync_directory(temporary_journal.parent)
summary_payload: dict[str, Any] = {
"schema": COMPACT_TRANSFER_SUMMARY_SCHEMA,
"passed": True,
"storageFormat": TEMPLATE_REFERENCE_TRANSFER_STORAGE,
"storeRoot": str(resolved_output),
"sessionId": session_values,
"sourceCheckpointSha256": source_checkpoint_sha256,
"startPageId": int(source_summary["startPageId"]),
"requestedPageCount": len(page_ids),
"journaledPageCount": len(page_ids),
"journalPath": str(resolved_output / "pages.jsonl"),
"journalSha256": _file_sha256(temporary_journal),
"pageParameterElements": int(source_bank.page_parameter_elements_t),
"physicalParameterElementsInitialized": int(
source_bank.physical_parameter_elements_t
),
"objectBytes": int(source_bank.total_object_bytes_t),
"physicalArchiveObjectBytes": template_bytes,
"templateObjectSha256": template_sha256,
"templateObjectBytes": template_bytes,
"templatePageId": template_page_id,
"templatePageIdOffset": page_id_offset,
"sourceSummaryPath": str(source_bank.summary_path),
"sourceSummarySha256": _file_sha256(source_bank.summary_path),
"completePageMappingRetained": True,
"trainedPageCount": 0,
"transferInitializedPageCount": len(page_ids),
"acceptedGenerationCommitted": False,
"promotionRequiresTrainingAndHeldoutProof": True,
}
_atomic_json(temporary_root / "summary.json", summary_payload)
os.replace(temporary_root, resolved_output)
_fsync_directory(resolved_output.parent)
cold_bank = discover_compact_transfer_page_bank(resolved_output / "summary.json")
if (
cold_bank.storage_format != TEMPLATE_REFERENCE_TRANSFER_STORAGE
or not torch.equal(cold_bank.page_ids_t, source_bank.page_ids_t)
or not torch.equal(cold_bank.object_sha256s_t, source_bank.object_sha256s_t)
or int(cold_bank.physical_archive_object_bytes_t) != template_bytes
):
raise RuntimeError("compact template-reference cold proof differs")
publish_compact_transfer_page_bank_locator_boundary(
resolved_output / "summary.json"
)
receipt: dict[str, Any] = {
"schema": "nnf.resynthesis.compact_transfer_template_reference.v1",
"passed": True,
"sourceSummaryPath": str(source_bank.summary_path),
"sourceSummarySha256": _file_sha256(source_bank.summary_path),
"outputSummaryPath": str(resolved_output / "summary.json"),
"outputSummarySha256": _file_sha256(resolved_output / "summary.json"),
"sourcePageCount": len(page_ids),
"trainedPageCount": 0,
"acceptedGenerationCommitted": False,
"logicalObjectBytes": int(source_bank.total_object_bytes_t),
"physicalArchiveObjectBytes": template_bytes,
"sourcePayloadHashesExhaustivelyVerified": True,
"virtualPayloadHashesExhaustivelyVerified": True,
"templateObjectSha256": template_sha256,
"templatePageId": template_page_id,
"promotionRequiresTrainingAndHeldoutProof": True,
}
_atomic_json(resolved_output / "template_reference_receipt.json", receipt)
return receipt
def hydrate_compact_page_admission_objects_boundary(
packet: CompactPageAdmissionPacket,
) -> NoNEImmutablePageStore:
"""Expose selected virtual objects as ordinary immutable page files.
The model owns cohort selection before this boundary. Hydration only
recreates those receipt-bound page identities, validates their exact hashes,
and gives the existing paged runtime its normal object-store interface.
"""
bank = discover_compact_transfer_page_bank(packet.bank_summary_path)
if (
bank.summary_path != packet.bank_summary_path.expanduser().resolve()
or bank.journal_path != packet.bank_journal_path.expanduser().resolve()
or bank.store_root != packet.bank_store_root.expanduser().resolve()
or bank.storage_format != packet.bank_storage_format
or not torch.equal(bank.summary_sha256_t, packet.bank_summary_sha256_t)
or not torch.equal(bank.journal_sha256_t, packet.bank_journal_sha256_t)
or not torch.equal(bank.session_id_t, packet.bank_session_id_t)
or not torch.equal(
bank.source_checkpoint_sha256_t,
packet.bank_source_checkpoint_sha256_t,
)
):
raise RuntimeError("compact page admission bank identity differs")
selected_page_ids_t = packet.selected_page_ids_t.detach().cpu().long().reshape(-1)
selected_hashes_t = packet.selected_object_sha256s_t.detach().cpu().to(
dtype=torch.uint8
)
selected_bytes_t = packet.selected_object_bytes_t.detach().cpu().long().reshape(-1)
if (
selected_hashes_t.shape != (selected_page_ids_t.numel(), 32)
or selected_bytes_t.shape != selected_page_ids_t.shape
):
raise RuntimeError("compact page admission object geometry differs")
if bank.storage_format == SCALED_FLOAT8_TRANSFER_STORAGE:
source_store = NoNEImmutablePageStore(bank.store_root)
source_store.begin_session(bank.session_id_t)
return source_store
if bank.storage_format != TEMPLATE_REFERENCE_TRANSFER_STORAGE:
raise RuntimeError("compact page admission storage format is unsupported")
if (
bank.template_object_path is None
or not torch.equal(
bank.template_object_sha256_t,
packet.bank_template_object_sha256_t.detach().cpu().to(
dtype=torch.uint8
),
)
or not torch.equal(
bank.template_page_id_t,
packet.bank_template_page_id_t.detach().cpu().long(),
)
or not torch.equal(
bank.template_page_id_offset_t,
packet.bank_template_page_id_offset_t.detach().cpu().long(),
)
or not torch.equal(
bank.physical_archive_object_bytes_t,
packet.bank_physical_archive_object_bytes_t.detach().cpu().long(),
)
):
raise RuntimeError("compact template admission identity differs")
hydration_root = bank.store_root / ".template_reference_hydrated"
hydration_store = NoNEImmutablePageStore(
hydration_root,
advertise_locator=False,
)
hydration_store.begin_session(bank.session_id_t)
template = _template_reference_payload_boundary(
template_path=bank.template_object_path,
template_sha256=bytes(bank.template_object_sha256_t.tolist()).hex(),
template_bytes=int(bank.physical_archive_object_bytes_t),
template_page_id=int(bank.template_page_id_t),
page_id_offset=int(bank.template_page_id_offset_t),
)
bindings: list[NoNEPageObjectBinding] = []
for index, page_id in enumerate(selected_page_ids_t.tolist()):
binding = hydration_store.write_compact_transfer_template_page_boundary(
template,
page_id=int(page_id),
)
if (
not torch.equal(
binding.object_sha256_t,
selected_hashes_t[index],
)
or not torch.equal(binding.object_bytes_t, selected_bytes_t[index])
):
raise RuntimeError("compact template hydrated object differs")
bindings.append(binding)
if not bindings:
raise RuntimeError("compact template admission selected no pages")
hydration_store.commit_compact_transfer_chunk_boundary(tuple(bindings))
return hydration_store
def compose_compact_transfer_page_banks(
*,
summary_paths: tuple[Path, ...],
output_root: Path,
) -> dict[str, Any]:
"""Compose contiguous immutable banks without copying page payloads.
This is an external storage-authority operation. It hardlinks already
verified content-addressed objects on one filesystem, preserves every
source journal row, and publishes one movable compact-bank locator only
after the aggregate summary cold-validates. Composition never marks a
page trained and never advances an accepted graph generation.
"""
if len(summary_paths) < 2:
raise ValueError("compact page composition requires at least two banks")
resolved_output = output_root.expanduser().resolve()
resolved_output.parent.mkdir(parents=True, exist_ok=True)
if resolved_output.exists():
raise FileExistsError(resolved_output)
banks = tuple(
sorted(
(
discover_compact_transfer_page_bank(path)
for path in summary_paths
),
key=lambda bank: int(bank.page_ids_t[0]),
)
)
reference = banks[0]
if reference.storage_format != SCALED_FLOAT8_TRANSFER_STORAGE or any(
bank.storage_format != SCALED_FLOAT8_TRANSFER_STORAGE for bank in banks
):
raise RuntimeError(
"compact template-reference banks are independently discoverable "
"and cannot be hardlink-composed"
)
for bank in banks[1:]:
if (
not torch.equal(bank.session_id_t, reference.session_id_t)
or not torch.equal(
bank.source_checkpoint_sha256_t,
reference.source_checkpoint_sha256_t,
)
or not torch.equal(
bank.page_parameter_elements_t,
reference.page_parameter_elements_t,
)
):
raise RuntimeError("compact page banks do not share one authority")
page_ids_t = torch.cat(tuple(bank.page_ids_t for bank in banks), dim=0)
first_page_id = int(page_ids_t[0])
expected_page_ids_t = torch.arange(
first_page_id,
first_page_id + page_ids_t.numel(),
dtype=torch.long,
)
if not torch.equal(page_ids_t, expected_page_ids_t):
raise RuntimeError("compact page banks are not one contiguous interval")
temporary_root = resolved_output.with_name(
f".{resolved_output.name}.compose.{os.getpid()}.tmp"
)
if temporary_root.exists():
raise FileExistsError(temporary_root)
temporary_root.mkdir()
objects_root = temporary_root / "objects" / "sha256"
objects_root.mkdir(parents=True)
journal_path = temporary_root / "pages.jsonl"
linked_digests: set[str] = set()
renamed = False
try:
destination_device = temporary_root.stat().st_dev
with journal_path.open("wb") as output_journal:
for bank in banks:
with bank.journal_path.open("rb") as source_journal:
shutil.copyfileobj(
source_journal,
output_journal,
length=8 * 1024 * 1024,
)
for object_sha256_t in bank.object_sha256s_t:
object_sha256 = bytes(
object_sha256_t.detach()
.cpu()
.to(dtype=torch.uint8)
.tolist()
).hex()
if object_sha256 in linked_digests:
raise RuntimeError(
"compact page banks repeat one page object"
)
source_object = (
bank.store_root
/ "objects"
/ "sha256"
/ f"{object_sha256}.safetensors"
)
if source_object.stat().st_dev != destination_device:
raise RuntimeError(
"compact page banks require one hardlink filesystem"
)
os.link(
source_object,
objects_root / source_object.name,
)
linked_digests.add(object_sha256)
output_journal.flush()
os.fsync(output_journal.fileno())
requested_page_count = page_ids_t.numel()
page_parameter_elements = int(reference.page_parameter_elements_t)
source_checkpoint_sha256 = bytes(
reference.source_checkpoint_sha256_t.detach()
.cpu()
.to(dtype=torch.uint8)
.tolist()
).hex()
session_id = reference.session_id_t.detach().cpu().long().tolist()
records = _read_compact_journal(
journal_path,
source_sha256=source_checkpoint_sha256,
session_id=session_id,
)
if tuple(sorted(records)) != tuple(expected_page_ids_t.tolist()):
raise RuntimeError("composed compact page journal is incomplete")
final_journal_path = resolved_output / journal_path.name
final_summary_path = resolved_output / "summary.json"
journal_sha256 = _file_sha256(journal_path)
summary: dict[str, Any] = {
"schema": COMPACT_TRANSFER_SUMMARY_SCHEMA,
"startPageId": first_page_id,
"requestedPageCount": requested_page_count,
"pageParameterElements": page_parameter_elements,
"physicalParameterElementsInitialized": (
requested_page_count * page_parameter_elements
),
"objectBytes": sum(
int(bank.total_object_bytes_t) for bank in banks
),
"journalPath": str(final_journal_path),
"journalSha256": journal_sha256,
"journaledPageCount": requested_page_count,
"storeRoot": str(resolved_output),
"sessionId": session_id,
"sourceCheckpointSha256": source_checkpoint_sha256,
"storageFormat": SCALED_FLOAT8_TRANSFER_STORAGE,
"passed": True,
"trainedPageCount": 0,
"transferInitializedPageCount": requested_page_count,
"completePageMappingRetained": True,
"acceptedGenerationCommitted": False,
"promotionRequiresTrainingAndHeldoutProof": True,
}
_atomic_json(temporary_root / final_summary_path.name, summary)
_fsync_directory(objects_root)
os.replace(temporary_root, resolved_output)
renamed = True
_fsync_directory(resolved_output.parent)
cold_bank = discover_compact_transfer_page_bank(final_summary_path)
if (
not torch.equal(cold_bank.page_ids_t, expected_page_ids_t)
or not torch.equal(cold_bank.session_id_t, reference.session_id_t)
or int(cold_bank.physical_parameter_elements_t)
!= requested_page_count * page_parameter_elements
):
raise RuntimeError("composed compact page bank cold proof differs")
publish_compact_transfer_page_bank_locator_boundary(
final_summary_path
)
receipt: dict[str, Any] = {
"schema": "nnf.resynthesis.compact_transfer_page_composition.v1",
"passed": True,
"sourceBanks": [
{
"summaryPath": str(bank.summary_path),
"summarySha256": _digest_hex(bank.summary_sha256_t),
"startPageId": int(bank.page_ids_t[0]),
"pageCount": bank.page_ids_t.numel(),
}
for bank in banks
],
"outputSummaryPath": str(final_summary_path),
"outputSummarySha256": _file_sha256(final_summary_path),
"outputJournalPath": str(final_journal_path),
"outputJournalSha256": journal_sha256,
"physicalGraphLayerCount": requested_page_count,
"physicalParameterElementsInitialized": (
requested_page_count * page_parameter_elements
),
"objectBytesReusedByHardlink": int(
cold_bank.total_object_bytes_t
),
"payloadBytesCopied": 0,
"trainedPageCount": 0,
"acceptedGenerationCommitted": False,
"coldDiscoveryPassed": True,
}
_atomic_json(resolved_output / "composition_receipt.json", receipt)
return receipt
except Exception:
if not renamed and temporary_root.exists():
shutil.rmtree(temporary_root)
raise
def plan_compact_page_bank_admission(
bank: CompactPageBankBinding,
scale_cohort: NoNEScaleCohortPacket,
*,
admitted_page_ids_t: torch.Tensor,
) -> CompactPageAdmissionPacket:
"""Bind available physical pages to a model-owned functional cohort."""
family_page_ids_t = (
scale_cohort.selected_family_page_ids_t.detach().cpu().long().reshape(-1)
)
if (
bank.ready_t.numel() != 1
or not bool(bank.ready_t)
or scale_cohort.ready_t.numel() != 1
or not bool(scale_cohort.ready_t)
or scale_cohort.selected_page_count_t.numel() != 1
or int(scale_cohort.selected_page_count_t) != family_page_ids_t.numel()
or family_page_ids_t.numel() < 1
or bank.page_ids_t.ndim != 1
or bank.object_sha256s_t.shape != (bank.page_ids_t.numel(), 32)
or bank.object_bytes_t.shape != bank.page_ids_t.shape
or admitted_page_ids_t.dtype != torch.long
):
raise RuntimeError("compact page admission tensor authority differs")
admitted_t = admitted_page_ids_t.detach().cpu().reshape(-1)
if torch.unique(admitted_t).numel() != admitted_t.numel():
raise RuntimeError("compact page admission history contains duplicates")
available_indexes_t = (~torch.isin(bank.page_ids_t, admitted_t)).nonzero(
as_tuple=False
).reshape(-1)
selected_count = family_page_ids_t.numel()
if available_indexes_t.numel() < selected_count:
raise RuntimeError("compact page bank has insufficient unadmitted capacity")
selected_indexes_t = available_indexes_t[:selected_count]
selected_page_ids_t = bank.page_ids_t.index_select(0, selected_indexes_t)
selected_sha256s_t = bank.object_sha256s_t.index_select(
0,
selected_indexes_t,
)
selected_bytes_t = bank.object_bytes_t.index_select(0, selected_indexes_t)
# Discovery seals the complete bank mapping once. Admission must still
# recheck the selected physical inputs before they cross from storage into
# a model-owned training transaction: that is a bounded cohort read, not a
# full-bank rescan. Template-reference banks have one physical object, so
# validate that exact object once before virtual selected pages are derived.
if bank.storage_format == SCALED_FLOAT8_TRANSFER_STORAGE:
objects_root = bank.store_root / "objects" / "sha256"
verification_rows = tuple(
(
int(selected_page_ids_t[index]),
bytes(selected_sha256s_t[index].tolist()).hex(),
int(selected_bytes_t[index]),
)
for index in range(selected_page_ids_t.numel())
)
object_authorities: list[tuple[Path, str]] = []
for page_id, expected_sha256, expected_bytes in verification_rows:
object_path = objects_root / f"{expected_sha256}.safetensors"
if (
not object_path.is_file()
or object_path.stat().st_size != expected_bytes
):
raise RuntimeError("compact page object differs at admission")
object_authorities.append((object_path, expected_sha256))
verified_sha256s = file_sha256_authority_batch_boundary(
tuple(object_authorities),
identity_cache_root=(
bank.store_root / "compact_admission_sha256_identity_cache"
),
)
if verified_sha256s != tuple(
expected_sha256
for _page_id, expected_sha256, _expected_bytes in verification_rows
):
raise RuntimeError("compact page object differs at admission")
elif bank.storage_format == TEMPLATE_REFERENCE_TRANSFER_STORAGE:
if bank.template_object_path is None:
raise RuntimeError("compact template admission object is absent")
_template_reference_payload_boundary(
template_path=bank.template_object_path,
template_sha256=bytes(bank.template_object_sha256_t.tolist()).hex(),
template_bytes=int(bank.physical_archive_object_bytes_t),
template_page_id=int(bank.template_page_id_t),
page_id_offset=int(bank.template_page_id_offset_t),
)
else:
raise RuntimeError("compact page admission storage format is unsupported")
return CompactPageAdmissionPacket(
bank_summary_path=bank.summary_path,
bank_journal_path=bank.journal_path,
bank_store_root=bank.store_root,
bank_storage_format=bank.storage_format,
bank_template_object_path=bank.template_object_path,
bank_summary_sha256_t=bank.summary_sha256_t.clone(),
bank_journal_sha256_t=bank.journal_sha256_t.clone(),
bank_session_id_t=bank.session_id_t.clone(),
bank_source_checkpoint_sha256_t=(
bank.source_checkpoint_sha256_t.clone()
),
bank_writer_source_sha256s_t=(
bank.writer_source_sha256s_t.clone()
),
selected_family_page_ids_t=family_page_ids_t,
selected_page_ids_t=selected_page_ids_t.clone(),
selected_object_sha256s_t=selected_sha256s_t.clone(),
selected_object_bytes_t=selected_bytes_t.clone(),
bank_page_count_t=torch.tensor(bank.page_ids_t.numel(), dtype=torch.long),
page_parameter_elements_t=bank.page_parameter_elements_t.clone(),
bank_physical_parameter_elements_t=(
bank.physical_parameter_elements_t.clone()
),
bank_total_object_bytes_t=bank.total_object_bytes_t.clone(),
bank_template_object_sha256_t=(
bank.template_object_sha256_t.clone()
),
bank_template_page_id_t=bank.template_page_id_t.clone(),
bank_template_page_id_offset_t=(
bank.template_page_id_offset_t.clone()
),
bank_physical_archive_object_bytes_t=(
bank.physical_archive_object_bytes_t.clone()
),
selected_page_count_t=torch.tensor(selected_count, dtype=torch.long),
ready_t=torch.ones((), dtype=torch.bool),
)
def _compact_summary_candidates(store_root: Path) -> tuple[Path, ...]:
"""Find compact summaries below an externally supplied store root."""
resolved_root = store_root.expanduser().resolve()
if resolved_root.is_file():
return (resolved_root,)
candidates = {
resolved_root / "summary.json",
*resolved_root.glob("*.summary.json"),
}
return tuple(sorted((path for path in candidates if path.is_file()), key=str))
def _compact_transfer_registry_roots_boundary(
*,
registry_roots: tuple[Path, ...] | None,
) -> tuple[Path, ...]:
"""Map mount-local page registries to transfer-bank-only registries.
A transfer-initialized bank has no accepted generation and must never be
advertised through ``page-stores``. Its sibling registry remains
filesystem-discoverable across mounts while keeping generation, route, and
acceptance authority outside this metadata boundary.
"""
page_roots = (
_default_page_store_registry_roots_boundary()
if registry_roots is None
else tuple(root.expanduser().resolve() for root in registry_roots)
)
return tuple(
sorted(
{
root.parent / COMPACT_TRANSFER_REGISTRY_DIRECTORY
for root in page_roots
},
key=str,
)
)
def publish_compact_transfer_page_bank_locator_boundary(
summary_path: Path,
*,
registry_roots: tuple[Path, ...] | None = None,
) -> tuple[Path, ...]:
"""Publish one sealed untrained bank without claiming a page generation.
The locator identifies immutable compact-bank metadata only. It cannot
select routes, mutate an accepted pointer, or make an initialization a
trained capability. Discovery revalidates the complete summary/journal
contract before admitting a model-selected subset.
"""
bank = discover_compact_transfer_page_bank(summary_path)
session_id_t = bank.session_id_t.detach().cpu().long().clone()
session_key = _session_key(session_id_t)
summary_sha256 = _digest_hex(bank.summary_sha256_t)
journal_sha256 = _digest_hex(bank.journal_sha256_t)
payload: dict[str, Any] = {
"schema": COMPACT_TRANSFER_LOCATOR_SCHEMA,
"sessionKey": session_key,
"sessionId": session_id_t.tolist(),
"root": str(bank.store_root),
"summaryPath": str(bank.summary_path),
"summarySha256": summary_sha256,
"journalPath": str(bank.journal_path),
"journalSha256": journal_sha256,
"storageFormat": bank.storage_format,
"completePageMappingRetained": True,
"trainedPageCount": 0,
"acceptedGenerationCommitted": False,
"routingAuthority": False,
"acceptedPointerMutationAuthority": False,
}
locator_id = hashlib.sha256(
json.dumps(
{
"root": payload["root"],
"summarySha256": summary_sha256,
},
sort_keys=True,
separators=(",", ":"),
).encode("utf-8")
).hexdigest()
roots = {
bank.store_root
/ ".nnf-resynthesis"
/ COMPACT_TRANSFER_REGISTRY_DIRECTORY,
_page_store_registry_root_for_boundary(bank.store_root).parent
/ COMPACT_TRANSFER_REGISTRY_DIRECTORY,
}
if registry_roots is not None:
roots.update(
root.parent / COMPACT_TRANSFER_REGISTRY_DIRECTORY
for root in registry_roots
)
written: list[Path] = []
for root in sorted(roots, key=str):
path = root / session_key / f"{locator_id}.json"
_atomic_json(path, payload)
written.append(path)
return tuple(written)
def _compact_transfer_locator_candidates_boundary(
*,
session_id_t: torch.Tensor,
registry_roots: tuple[Path, ...] | None,
) -> tuple[tuple[Path, Path], ...]:
"""Resolve only sealed dedicated compact-transfer locator records."""
session = session_id_t.detach().cpu().long().reshape(-1)
session_key = _session_key(session)
expected_session = session.tolist()
candidates: dict[Path, Path] = {}
for registry_root in _compact_transfer_registry_roots_boundary(
registry_roots=registry_roots
):
session_root = registry_root / session_key
try:
session_root_available = session_root.is_dir()
locator_paths = (
sorted(session_root.glob("*.json"), key=str)
if session_root_available
else ()
)
except PermissionError:
# Registry roots are discovery hints. An optional kernel/BPF
# registry may be mounted but intentionally unreadable to this
# process; skip that root and retain strict validation for every
# locator that is actually discovered.
continue
for locator_path in locator_paths:
payload = json.loads(locator_path.read_text(encoding="utf-8"))
if not isinstance(payload, dict):
raise RuntimeError("compact transfer locator is not an object")
root_value = payload.get("root")
summary_value = payload.get("summaryPath")
summary_sha256 = payload.get("summarySha256")
journal_value = payload.get("journalPath")
journal_sha256 = payload.get("journalSha256")
if (
payload.get("schema") != COMPACT_TRANSFER_LOCATOR_SCHEMA
or payload.get("sessionKey") != session_key
or payload.get("sessionId") != expected_session
or not isinstance(root_value, str)
or not isinstance(summary_value, str)
or not isinstance(summary_sha256, str)
or len(summary_sha256) != 64
or not isinstance(journal_value, str)
or not isinstance(journal_sha256, str)
or len(journal_sha256) != 64
or payload.get("completePageMappingRetained") is not True
or payload.get("trainedPageCount") != 0
or payload.get("acceptedGenerationCommitted") is not False
or payload.get("routingAuthority") is not False
or payload.get("acceptedPointerMutationAuthority") is not False
):
raise RuntimeError("compact transfer locator authority differs")
store_root = Path(root_value).expanduser().resolve()
summary_path = Path(summary_value).expanduser().resolve()
journal_path = Path(journal_value).expanduser().resolve()
if (
not store_root.is_dir()
or summary_path.parent != store_root
or not summary_path.is_file()
or not journal_path.is_file()
or _file_sha256(summary_path) != summary_sha256
or _file_sha256(journal_path) != journal_sha256
):
raise RuntimeError("compact transfer locator payload differs")
previous = candidates.get(summary_path)
if previous is not None and previous != store_root:
raise RuntimeError("compact transfer locators disagree on a bank")
candidates[summary_path] = store_root
return tuple((root, summary) for summary, root in sorted(candidates.items()))
def _legacy_compact_transfer_locator_candidates_boundary(
*,
session_id_t: torch.Tensor,
registry_roots: tuple[Path, ...] | None,
) -> tuple[tuple[Path, Path], ...]:
"""Read legacy transfer-bank metadata without granting page-store status.
Older writers published transfer candidates in ``page-stores`` before an
accepted pointer existed. Their roots are valid only if a compact summary
independently passes the current immutable bank validation below; stale
page-store records with no compact summary contribute no capacity.
"""
session = session_id_t.detach().cpu().long().reshape(-1)
session_key = _session_key(session)
expected_session = session.tolist()
page_roots = (
_default_page_store_registry_roots_boundary()
if registry_roots is None
else tuple(root.expanduser().resolve() for root in registry_roots)
)
candidates: dict[Path, Path] = {}
for registry_root in page_roots:
session_root = registry_root / session_key
try:
session_root_exists = session_root.is_dir()
except OSError:
continue
if not session_root_exists:
continue
try:
locator_paths = sorted(session_root.glob("*.json"), key=str)
except OSError:
continue
for locator_path in locator_paths:
try:
payload = json.loads(locator_path.read_text(encoding="utf-8"))
except OSError:
continue
if not isinstance(payload, dict):
continue
if payload.get("schema") != PAGE_STORE_LOCATOR_SCHEMA:
continue
root_value = payload.get("root")
if (
payload.get("sessionKey") != session_key
or payload.get("sessionId") != expected_session
or not isinstance(root_value, str)
):
raise RuntimeError("legacy compact locator authority differs")
store_root = Path(root_value).expanduser().resolve()
try:
store_root_exists = store_root.is_dir()
except OSError:
continue
if not store_root_exists:
continue
try:
summary_paths = _compact_summary_candidates(store_root)
except OSError:
continue
for summary_path in summary_paths:
candidates.setdefault(summary_path, store_root)
return tuple((root, summary) for summary, root in sorted(candidates.items()))
def _compact_stream_from_summary_boundary(
*,
store_root: Path,
session_id_t: torch.Tensor,
summary_path: Path,
) -> CompactPageStreamPacket | None:
summary = json.loads(summary_path.read_text(encoding="utf-8"))
if not isinstance(summary, dict):
raise RuntimeError("compact page summary is not an object")
if summary.get("schema") != COMPACT_TRANSFER_SUMMARY_SCHEMA:
return None
# A disjoint compact-bank builder publishes progress in its summary. It
# is discoverable metadata, but it is not admission authority until the
# complete page mapping has been sealed.
if (
summary.get("passed") is not True
or summary.get("completePageMappingRetained") is not True
):
return None
session_id = session_id_t.detach().cpu().long().reshape(-1).tolist()
journal_value = summary.get("journalPath")
journal_sha256 = summary.get("journalSha256")
source_sha256 = summary.get("sourceCheckpointSha256")
journaled_page_count = summary.get("journaledPageCount")
transfer_initialized_page_count = summary.get(
"transferInitializedPageCount"
)
requested_page_count = summary.get("requestedPageCount")
start_page_id = summary.get("startPageId")
if (
summary.get("sessionId") != session_id
or summary.get("storageFormat")
not in {
SCALED_FLOAT8_TRANSFER_STORAGE,
TEMPLATE_REFERENCE_TRANSFER_STORAGE,
}
or summary.get("trainedPageCount") != 0
or summary.get("acceptedGenerationCommitted") is not False
or not isinstance(journal_value, str)
or not isinstance(journal_sha256, str)
or len(journal_sha256) != 64
or not isinstance(source_sha256, str)
or len(source_sha256) != 64
or not isinstance(journaled_page_count, int)
or isinstance(journaled_page_count, bool)
or journaled_page_count < 0
or not isinstance(transfer_initialized_page_count, int)
or isinstance(transfer_initialized_page_count, bool)
or not isinstance(requested_page_count, int)
or isinstance(requested_page_count, bool)
or requested_page_count < 1
or journaled_page_count != requested_page_count
or transfer_initialized_page_count != requested_page_count
or not isinstance(start_page_id, int)
or isinstance(start_page_id, bool)
or start_page_id < 0
):
raise RuntimeError("compact page summary authority differs")
recorded_journal = Path(journal_value).expanduser()
journal_path = (
recorded_journal.resolve()
if recorded_journal.is_file()
else (store_root / recorded_journal.name).resolve()
)
if not journal_path.is_file():
raise RuntimeError("compact page journal is absent during discovery")
if _file_sha256(journal_path) != journal_sha256:
refreshed = json.loads(summary_path.read_text(encoding="utf-8"))
if (
isinstance(refreshed, dict)
and refreshed.get("journalSha256") != journal_sha256
):
return _compact_stream_from_summary_boundary(
store_root=store_root,
session_id_t=session_id_t,
summary_path=summary_path,
)
raise RuntimeError("compact page journal changed during discovery")
records = _read_compact_journal(
journal_path,
source_sha256=source_sha256,
session_id=session_id,
)
ordered_ids = sorted(records)
if (
len(ordered_ids) != journaled_page_count
or ordered_ids
!= list(range(start_page_id, start_page_id + journaled_page_count))
):
raise RuntimeError("compact page journal mapping is incomplete")
if not (store_root / "objects/sha256").is_dir():
raise RuntimeError("discovered compact page object root is absent")
page_ids_t = torch.tensor(ordered_ids, dtype=torch.long)
object_sha256_t = (
torch.stack(
tuple(
digest_tensor(str(records[page_id]["objectSha256"]))
for page_id in ordered_ids
),
dim=0,
)
if ordered_ids
else torch.zeros(0, 32, dtype=torch.uint8)
)
object_bytes_t = torch.tensor(
[int(records[page_id]["objectBytes"]) for page_id in ordered_ids],
dtype=torch.long,
)
return CompactPageStreamPacket(
store_root=store_root,
summary_path=summary_path.resolve(),
journal_path=journal_path,
storage_format=str(summary["storageFormat"]),
session_id_t=session_id_t.detach().cpu().long().clone(),
page_ids_t=page_ids_t,
object_sha256_t=object_sha256_t,
object_bytes_t=object_bytes_t,
trained_t=torch.zeros(len(ordered_ids), dtype=torch.bool),
accepted_generation_committed_t=torch.zeros(
len(ordered_ids),
dtype=torch.bool,
),
)
def discover_compact_page_streams_boundary(
*,
session_id_t: torch.Tensor,
anchor_roots: tuple[Path, ...] = (),
registry_roots: tuple[Path, ...] | None = None,
) -> tuple[CompactPageStreamPacket, ...]:
"""Auto-discover every stable compact stream for one NoNE session."""
candidates: dict[Path, tuple[Path, bool]] = {}
for store_root, summary_path in _compact_transfer_locator_candidates_boundary(
session_id_t=session_id_t,
registry_roots=registry_roots,
):
candidates[summary_path] = (store_root, True)
for store_root, summary_path in _legacy_compact_transfer_locator_candidates_boundary(
session_id_t=session_id_t,
registry_roots=registry_roots,
):
candidates.setdefault(summary_path, (store_root, False))
for anchor_root in anchor_roots:
resolved_anchor = anchor_root.expanduser().resolve()
for summary_path in _compact_summary_candidates(resolved_anchor):
candidates.setdefault(summary_path, (resolved_anchor, False))
streams: list[CompactPageStreamPacket] = []
observed_pages: dict[int, bytes] = {}
for summary_path, (store_root, dedicated_locator) in sorted(
candidates.items(),
key=lambda entry: str(entry[0]),
):
stream = _compact_stream_from_summary_boundary(
store_root=store_root,
session_id_t=session_id_t,
summary_path=summary_path,
)
if stream is None:
if dedicated_locator:
raise RuntimeError("compact transfer locator summary differs")
continue
for index, page_id in enumerate(stream.page_ids_t.tolist()):
digest = bytes(stream.object_sha256_t[index].tolist())
previous = observed_pages.get(page_id)
if previous is not None and previous != digest:
raise RuntimeError(
"compact page discovery found conflicting page identities"
)
observed_pages[page_id] = digest
streams.append(stream)
return tuple(
sorted(
streams,
key=lambda stream: (str(stream.store_root), str(stream.summary_path)),
)
)
def none_scale_evidence_from_record_boundary(
record: Mapping[str, Any],
) -> NoNEScaleEvidencePacket:
"""Reconstruct and revalidate persisted model-owned scale evidence."""
family_ids = record.get("familyRootPageIds")
retained = record.get("retainedFamilyMask")
gap = record.get("unresolvedGapPressure")
route = record.get("routePressure")
distinct = record.get("distinctGradientMask")
updates = record.get("freshGradientUpdateCounts")
gradient_norms = record.get("freshGradientNorms")
parameter_deltas = record.get("freshParameterDeltaNorms")
signatures = record.get("freshGradientSignatures")
if (
record.get("schema") != "nnf.resynthesis.none_scale_evidence.v1"
or record.get("modelOwned") is not True
or record.get("targetFree") is not True
or record.get("storageAuthority") is not False
or record.get("graphAdmissionAuthority") is not False
or record.get("trainingClaimed") is not False
or type(record.get("evidenceReady")) is not bool
or not isinstance(family_ids, list)
or not family_ids
or any(
not isinstance(value, int) or isinstance(value, bool) or value < 0
for value in family_ids
)
or len(set(family_ids)) != len(family_ids)
):
raise RuntimeError("NoNE scale evidence authority differs")
family_count = len(family_ids)
vector_values = (
retained,
gap,
route,
distinct,
updates,
gradient_norms,
parameter_deltas,
)
if any(
not isinstance(value, list) or len(value) != family_count
for value in vector_values
):
raise RuntimeError("NoNE scale evidence vector geometry differs")
retained_values = cast(list[Any], retained)
gap_values = cast(list[Any], gap)
route_values = cast(list[Any], route)
distinct_values = cast(list[Any], distinct)
update_values = cast(list[Any], updates)
gradient_norm_values = cast(list[Any], gradient_norms)
parameter_delta_values = cast(list[Any], parameter_deltas)
if (
any(type(value) is not bool for value in retained_values)
or any(type(value) is not bool for value in distinct_values)
or any(
not isinstance(value, int) or isinstance(value, bool) or value < 0
for value in update_values
)
or any(
not isinstance(value, (int, float))
or isinstance(value, bool)
or not math.isfinite(float(value))
or float(value) < 0.0
for values in (
gap_values,
route_values,
gradient_norm_values,
parameter_delta_values,
)
for value in values
)
or not isinstance(signatures, list)
or len(signatures) != family_count
or not signatures
or not isinstance(signatures[0], list)
or not signatures[0]
):
raise RuntimeError("NoNE scale evidence values differ")
signature_values = signatures
signature_width = len(signature_values[0])
if any(
not isinstance(row, list)
or len(row) != signature_width
or any(
not isinstance(value, (int, float))
or isinstance(value, bool)
or not math.isfinite(float(value))
for value in row
)
for row in signature_values
):
raise RuntimeError("NoNE scale evidence signature geometry differs")
packet = NoNEScaleEvidencePacket(
family_page_ids_t=torch.tensor(family_ids, dtype=torch.long),
retained_family_mask_t=torch.tensor(retained_values, dtype=torch.bool),
unresolved_gap_pressure_t=torch.tensor(gap_values, dtype=torch.float32),
route_pressure_t=torch.tensor(route_values, dtype=torch.float32),
distinct_gradient_mask_t=torch.tensor(
distinct_values,
dtype=torch.bool,
),
fresh_gradient_update_count_t=torch.tensor(
update_values,
dtype=torch.long,
),
fresh_gradient_norm_t=torch.tensor(
gradient_norm_values,
dtype=torch.float32,
),
fresh_parameter_delta_norm_t=torch.tensor(
parameter_delta_values,
dtype=torch.float32,
),
fresh_gradient_signature_t=torch.tensor(
signature_values,
dtype=torch.float32,
),
evidence_ready_t=torch.tensor(
bool(record.get("evidenceReady")),
dtype=torch.bool,
),
)
eligible_t = _qualified_scale_evidence_mask(packet)
eligible_count = record.get("eligibleFamilyCount")
if (
not isinstance(eligible_count, int)
or isinstance(eligible_count, bool)
or eligible_count != int(eligible_t.long().sum())
or record.get("evidenceReady") != bool(eligible_t.any())
):
raise RuntimeError("NoNE scale evidence readiness differs")
return packet
def _qualified_scale_evidence_mask(
packet: NoNEScaleEvidencePacket,
) -> torch.Tensor:
"""Require fresh routing, gradients, and parameter change for page growth."""
family_count = packet.family_page_ids_t.reshape(-1).shape[0]
vector_shapes = (
packet.retained_family_mask_t.reshape(-1).shape,
packet.unresolved_gap_pressure_t.reshape(-1).shape,
packet.route_pressure_t.reshape(-1).shape,
packet.distinct_gradient_mask_t.reshape(-1).shape,
packet.fresh_gradient_update_count_t.reshape(-1).shape,
packet.fresh_gradient_norm_t.reshape(-1).shape,
packet.fresh_parameter_delta_norm_t.reshape(-1).shape,
)
if any(shape != (family_count,) for shape in vector_shapes):
raise RuntimeError("NoNE scale evidence vector geometry differs")
gap_t = packet.unresolved_gap_pressure_t.reshape(-1)
route_t = packet.route_pressure_t.reshape(-1)
gradient_norm_t = packet.fresh_gradient_norm_t.reshape(-1)
parameter_delta_t = packet.fresh_parameter_delta_norm_t.reshape(-1)
finite_t = (
torch.isfinite(gap_t)
& torch.isfinite(route_t)
& torch.isfinite(gradient_norm_t)
& torch.isfinite(parameter_delta_t)
)
return (
packet.retained_family_mask_t.reshape(-1).bool()
& packet.distinct_gradient_mask_t.reshape(-1).bool()
& finite_t
& gap_t.gt(0)
& route_t.gt(0)
& packet.fresh_gradient_update_count_t.reshape(-1).gt(0)
& gradient_norm_t.gt(0)
& parameter_delta_t.gt(0)
)
def _template_reference_selected_equivalent_bank_boundary(
*,
streams: tuple[CompactPageStreamPacket, ...],
admitted_page_ids_t: torch.Tensor,
source_bank: CompactPageBankBinding,
selected_page_ids_t: torch.Tensor,
selected_object_sha256s_t: torch.Tensor,
selected_object_bytes_t: torch.Tensor,
) -> CompactPageBankBinding:
"""Prefer an exact template representation for an already selected cohort.
A shorter conventional bank can rank before a wider template-reference bank
even when their next selected pages are byte-identical. Storage extent is
not model authority, so preserve the source cohort exactly and substitute
only a template bank with the same selected IDs, object digests, sizes,
geometry, session, source checkpoint, and writer provenance.
"""
admitted_t = admitted_page_ids_t.detach().cpu().long().reshape(-1)
selected_ids = selected_page_ids_t.detach().cpu().long().reshape(-1)
selected_sha256s = selected_object_sha256s_t.detach().cpu().to(
dtype=torch.uint8
)
selected_bytes = selected_object_bytes_t.detach().cpu().long().reshape(-1)
selected_count = selected_ids.numel()
if (
selected_count < 1
or selected_sha256s.shape != (selected_count, 32)
or selected_bytes.numel() != selected_count
):
raise RuntimeError("compact selected cohort geometry differs")
for stream in streams:
if stream.storage_format != TEMPLATE_REFERENCE_TRANSFER_STORAGE:
continue
available_indexes_t = (~torch.isin(stream.page_ids_t, admitted_t)).nonzero(
as_tuple=False
).reshape(-1)
if available_indexes_t.numel() < selected_count:
continue
indexes_t = available_indexes_t[:selected_count]
if (
not torch.equal(
stream.page_ids_t.index_select(0, indexes_t),
selected_ids,
)
or not torch.equal(
stream.object_sha256_t.index_select(0, indexes_t),
selected_sha256s,
)
or not torch.equal(
stream.object_bytes_t.index_select(0, indexes_t),
selected_bytes,
)
):
continue
template_bank = discover_compact_transfer_page_bank(stream.summary_path)
template_available_indexes_t = (
~torch.isin(template_bank.page_ids_t, admitted_t)
).nonzero(as_tuple=False).reshape(-1)
template_indexes_t = template_available_indexes_t[:selected_count]
if (
template_indexes_t.numel() != selected_count
or not torch.equal(
template_bank.page_ids_t.index_select(0, template_indexes_t),
selected_ids,
)
or not torch.equal(
template_bank.object_sha256s_t.index_select(0, template_indexes_t),
selected_sha256s,
)
or not torch.equal(
template_bank.object_bytes_t.index_select(0, template_indexes_t),
selected_bytes,
)
or not torch.equal(
template_bank.session_id_t,
source_bank.session_id_t,
)
or not torch.equal(
template_bank.source_checkpoint_sha256_t,
source_bank.source_checkpoint_sha256_t,
)
or not torch.equal(
template_bank.page_parameter_elements_t,
source_bank.page_parameter_elements_t,
)
):
continue
return template_bank
return source_bank
def plan_discovered_compact_page_admission_boundary(
*,
session_id_t: torch.Tensor,
scale_evidence: NoNEScaleEvidencePacket,
accepted_training_saturation: NoNEAcceptedTrainingSaturationPacket,
admitted_page_ids_t: torch.Tensor,
replica_free_bytes_t: torch.Tensor,
reserve_bytes_t: torch.Tensor,
current_physical_parameter_elements_t: torch.Tensor,
capacity_envelope_parameter_elements_t: torch.Tensor,
maximum_page_count_t: torch.Tensor | None = None,
objective_compatible_family_mask_t: torch.Tensor | None = None,
anchor_roots: tuple[Path, ...] = (),
registry_roots: tuple[Path, ...] | None = None,
) -> DiscoveredCompactAdmissionPlan:
"""Bind retained model evidence to auto-discovered physical capacity.
Discovery resolves a movable session-owned store and chooses the earliest
unadmitted physical identities. It does not mutate a catalog or accepted
pointer; the returned packet must still pass the immutable migration.
"""
if session_id_t.reshape(-1).shape != (4,):
raise ValueError("NoNE compact admission session geometry differs")
saturation_ready_t = (
validate_none_accepted_training_saturation_boundary(
accepted_training_saturation
)
)
if (
saturation_ready_t.numel() != 1
or not bool(saturation_ready_t)
or accepted_training_saturation.remaining_unproven_page_ids_t.numel()
!= 0
):
raise RuntimeError(
"NoNE compact admission requires a saturated accepted page bank"
)
maximum_count_t: torch.Tensor | None = None
if maximum_page_count_t is not None:
maximum_count_t = maximum_page_count_t.reshape(()).long()
if maximum_count_t.numel() != 1 or not bool(maximum_count_t.gt(0)):
raise ValueError("NoNE compact admission maximum must be positive")
qualified_family_mask_t = _qualified_scale_evidence_mask(scale_evidence)
if (
scale_evidence.evidence_ready_t.numel() != 1
or not bool(scale_evidence.evidence_ready_t)
or not bool(qualified_family_mask_t.any())
):
raise RuntimeError("NoNE compact admission has no retained scale evidence")
admitted_t = admitted_page_ids_t.detach().cpu().long().reshape(-1)
if torch.unique(admitted_t).numel() != admitted_t.numel():
raise RuntimeError("NoNE compact admission history contains duplicates")
if not torch.equal(
torch.sort(admitted_t).values,
accepted_training_saturation.training_eligible_page_ids_t.detach()
.cpu()
.long(),
):
raise RuntimeError(
"NoNE compact admission saturation page identity differs"
)
streams = discover_compact_page_streams_boundary(
session_id_t=session_id_t,
anchor_roots=anchor_roots,
registry_roots=registry_roots,
)
candidates: list[
tuple[
tuple[int, int, int, int],
str,
tuple[tuple[int, bytes, int], ...],
CompactPageStreamPacket,
]
] = []
for stream in streams:
available_mask_t = ~torch.isin(stream.page_ids_t, admitted_t)
available_indexes_t = available_mask_t.nonzero(
as_tuple=False
).reshape(-1)
available_count = int(available_indexes_t.numel())
if available_count > 0:
total_count = int(stream.page_ids_t.numel())
available_ids_t = stream.page_ids_t.index_select(
0,
available_indexes_t,
)
fully_unadmitted = available_count == total_count
identity = tuple(
(
int(stream.page_ids_t[index]),
bytes(stream.object_sha256_t[index].tolist()),
int(stream.object_bytes_t[index]),
)
for index in available_indexes_t.tolist()
)
candidates.append(
(
(
int(available_ids_t.amin()),
0 if fully_unadmitted else 1,
total_count,
available_count,
),
str(stream.summary_path),
identity,
stream,
)
)
selected_candidates: list[
tuple[
tuple[int, int, int, int],
str,
tuple[tuple[int, bytes, int], ...],
CompactPageStreamPacket,
]
] = []
for rank in sorted({candidate[0] for candidate in candidates}):
ranked = sorted(
(candidate for candidate in candidates if candidate[0] == rank),
# Equal-rank candidates must expose the exact same virtual page
# identities below. Prefer the template-reference representation
# in that case: it revalidates one immutable template plus every
# selected virtual identity instead of rereading every equivalent
# scaled-float8 object before the same admission transaction.
key=lambda candidate: (
0
if candidate[3].storage_format
== TEMPLATE_REFERENCE_TRANSFER_STORAGE
else 1,
candidate[1],
),
)
identities = {candidate[2] for candidate in ranked}
if len(identities) != 1:
raise RuntimeError(
"compact page discovery found conflicting equal-rank banks"
)
selected_candidates.append(ranked[0])
for _rank, _summary_name, _identity, stream in selected_candidates:
bank = discover_compact_transfer_page_bank(stream.summary_path)
if (
not torch.equal(bank.session_id_t, session_id_t.detach().cpu().long())
or not torch.equal(bank.page_ids_t, stream.page_ids_t)
or not torch.equal(bank.object_sha256s_t, stream.object_sha256_t)
or not torch.equal(bank.object_bytes_t, stream.object_bytes_t)
):
raise RuntimeError("auto-discovered compact bank identity differs")
available_mask_t = ~torch.isin(bank.page_ids_t, admitted_t)
available_count_t = available_mask_t.long().sum()
if not bool(available_count_t.gt(0)):
continue
measured_page_bytes_t = bank.object_bytes_t.masked_select(
available_mask_t
).amax()
free_bytes_t = replica_free_bytes_t.reshape(-1).long()
reserve_t = reserve_bytes_t.reshape(()).long()
bank_capacity_bytes_t = available_count_t * measured_page_bytes_t
effective_free_bytes_t = torch.minimum(
(free_bytes_t - reserve_t).clamp_min(0),
bank_capacity_bytes_t.expand_as(free_bytes_t),
) + reserve_t
effective_envelope_t = capacity_envelope_parameter_elements_t
if maximum_count_t is not None:
effective_envelope_t = torch.minimum(
capacity_envelope_parameter_elements_t.reshape(()).long(),
current_physical_parameter_elements_t.reshape(()).long()
+ maximum_count_t * bank.page_parameter_elements_t,
)
cohort = plan_none_scale_cohort(
family_page_ids_t=scale_evidence.family_page_ids_t,
retained_family_mask_t=qualified_family_mask_t,
unresolved_gap_pressure_t=(
scale_evidence.unresolved_gap_pressure_t
),
route_pressure_t=scale_evidence.route_pressure_t,
distinct_gradient_mask_t=(
scale_evidence.distinct_gradient_mask_t
),
replica_free_bytes_t=effective_free_bytes_t,
reserve_bytes_t=reserve_t,
measured_compact_page_bytes_t=measured_page_bytes_t,
page_parameter_elements_t=bank.page_parameter_elements_t,
current_physical_parameter_elements_t=(
current_physical_parameter_elements_t
),
capacity_envelope_parameter_elements_t=(
effective_envelope_t
),
objective_compatible_family_mask_t=(
objective_compatible_family_mask_t
),
)
if not bool(cohort.ready_t):
continue
selected_count = int(cohort.selected_page_count_t)
selected_indexes_t = available_mask_t.nonzero(
as_tuple=False
).reshape(-1)[:selected_count]
bank = _template_reference_selected_equivalent_bank_boundary(
streams=streams,
admitted_page_ids_t=admitted_t,
source_bank=bank,
selected_page_ids_t=bank.page_ids_t.index_select(
0,
selected_indexes_t,
),
selected_object_sha256s_t=bank.object_sha256s_t.index_select(
0,
selected_indexes_t,
),
selected_object_bytes_t=bank.object_bytes_t.index_select(
0,
selected_indexes_t,
),
)
admission = plan_compact_page_bank_admission(
bank,
cohort,
admitted_page_ids_t=admitted_t,
)
return DiscoveredCompactAdmissionPlan(
bank=bank,
scale_cohort=cohort,
admission=admission,
)
raise RuntimeError(
"NoNE compact admission found no storage-safe unadmitted page cohort"
)
def compact_training_eligible_page_ids_boundary(
*,
streams: tuple[CompactPageStreamPacket, ...],
catalog_page_ids_t: torch.Tensor,
catalog_untrained_page_ids_t: torch.Tensor,
) -> torch.Tensor:
"""Intersect discovered capacity with explicit untrained catalog authority."""
catalog_ids = catalog_page_ids_t.detach().cpu().long().reshape(-1)
untrained_ids = catalog_untrained_page_ids_t.detach().cpu().long().reshape(-1)
if (
torch.unique(catalog_ids).numel() != catalog_ids.numel()
or torch.unique(untrained_ids).numel() != untrained_ids.numel()
or not untrained_ids.unsqueeze(1).eq(catalog_ids.unsqueeze(0)).any(dim=1).all()
):
raise RuntimeError("compact training catalog authority differs")
discovered_ids = (
torch.cat(tuple(stream.page_ids_t for stream in streams), dim=0)
if streams
else torch.zeros(0, dtype=torch.long)
)
if discovered_ids.numel() == 0:
return discovered_ids
discovered_ids = torch.unique(discovered_ids, sorted=True)
return discovered_ids[
discovered_ids.unsqueeze(1).eq(untrained_ids.unsqueeze(0)).any(dim=1)
]
def _append_journal_rows(
path: Path,
rows: tuple[Mapping[str, Any], ...],
) -> None:
if not rows:
raise ValueError("compact page journal chunk is empty")
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("a", encoding="utf-8") as handle:
for row in rows:
handle.write(
json.dumps(row, sort_keys=True, separators=(",", ":")) + "\n"
)
handle.flush()
os.fsync(handle.fileno())
def _write_compact_summary(
*,
summary_path: Path,
journal_path: Path,
records: Mapping[int, Mapping[str, Any]],
store_root: Path,
session_id: list[int],
source_checkpoint_sha256: str,
start_page_id: int,
requested_page_count: int,
page_parameter_elements: int,
) -> dict[str, Any]:
ordered = [records[page_id] for page_id in sorted(records)]
payload: dict[str, Any] = {
"schema": COMPACT_TRANSFER_SUMMARY_SCHEMA,
"passed": len(ordered) == requested_page_count,
"storageFormat": SCALED_FLOAT8_TRANSFER_STORAGE,
"storeRoot": str(store_root),
"sessionId": session_id,
"sourceCheckpointSha256": source_checkpoint_sha256,
"startPageId": start_page_id,
"requestedPageCount": requested_page_count,
"journaledPageCount": len(ordered),
"journalPath": str(journal_path),
"journalSha256": _file_sha256(journal_path),
"pageParameterElements": page_parameter_elements,
"physicalParameterElementsInitialized": (
len(ordered) * page_parameter_elements
),
"objectBytes": sum(int(row["objectBytes"]) for row in ordered),
"completePageMappingRetained": True,
"trainedPageCount": 0,
"transferInitializedPageCount": len(ordered),
"acceptedGenerationCommitted": False,
"promotionRequiresTrainingAndHeldoutProof": True,
}
_atomic_json(summary_path, payload)
return payload
def write_compact_transfer_pages(
*,
source_gate_t: torch.Tensor,
source_up_t: torch.Tensor,
source_down_t: torch.Tensor,
source_glyph_down_t: torch.Tensor,
source_glyph_up_t: torch.Tensor,
source_checkpoint_sha256: str,
store_root: Path,
session_id_t: torch.Tensor,
journal_path: Path,
summary_path: Path,
start_page_id: int,
page_count: int,
router_size: int,
writer_source_sha256: str | None = None,
receipt_every: int = 100,
registry_roots: tuple[Path, ...] | None = None,
) -> dict[str, Any]:
"""Create and verify every compact candidate with crash-safe full mapping."""
if len(source_checkpoint_sha256) != 64:
raise ValueError("source checkpoint SHA-256 is malformed")
writer_source_diagnostic = (
writer_source_sha256
if _diagnostic_sha256_bytes(writer_source_sha256) is not None
else None
)
if start_page_id < 0 or page_count < 1 or receipt_every < 1:
raise ValueError("compact page range and receipt cadence must be positive")
session_id = session_id_t.detach().cpu().long().reshape(-1).tolist()
if not session_id:
raise ValueError("compact page growth requires a session identity")
store = NoNEImmutablePageStore(store_root)
store.begin_session(session_id_t)
template_weights = compact_transfer_weights_from_source(
page_id=0,
source_gate_t=source_gate_t,
source_up_t=source_up_t,
source_down_t=source_down_t,
source_glyph_down_t=source_glyph_down_t,
source_glyph_up_t=source_glyph_up_t,
router_size=router_size,
)
transfer_template = store.build_compact_transfer_template_boundary(
template_weights
)
records = _read_compact_journal(
journal_path,
source_sha256=source_checkpoint_sha256,
session_id=session_id,
)
requested_ids = range(start_page_id, start_page_id + page_count)
unexpected = set(records).difference(requested_ids)
if unexpected:
raise RuntimeError("compact page journal contains out-of-range pages")
page_parameter_elements = sum(
int(tensor.numel())
for tensor in (
source_gate_t,
source_up_t,
source_down_t,
source_glyph_down_t,
source_glyph_up_t,
)
) + 1 + 3 * router_size
pending_bindings: list[NoNEPageObjectBinding] = []
pending_records: list[dict[str, Any]] = []
def commit_pending_chunk() -> None:
if not pending_bindings:
return
store.commit_compact_transfer_chunk_boundary(tuple(pending_bindings))
_append_journal_rows(journal_path, tuple(pending_records))
for record in pending_records:
records[int(record["pageId"])] = record
pending_bindings.clear()
pending_records.clear()
_write_compact_summary(
summary_path=summary_path,
journal_path=journal_path,
records=records,
store_root=store.root,
session_id=session_id,
source_checkpoint_sha256=source_checkpoint_sha256,
start_page_id=start_page_id,
requested_page_count=page_count,
page_parameter_elements=page_parameter_elements,
)
for page_id in requested_ids:
existing = records.get(page_id)
if existing is not None:
store.verify_compact_transfer_page_object_boundary(
_binding_from_record(existing)
)
else:
binding = store.write_compact_transfer_template_page_boundary(
transfer_template,
page_id=page_id,
)
record: dict[str, Any] = {
"schema": COMPACT_TRANSFER_JOURNAL_SCHEMA,
"pageId": page_id,
"objectSha256": _digest_hex(binding.object_sha256_t),
"objectBytes": int(binding.object_bytes_t.detach().cpu()),
"storageFormat": SCALED_FLOAT8_TRANSFER_STORAGE,
"sourceCheckpointSha256": source_checkpoint_sha256,
"sessionId": session_id,
"trained": False,
"acceptedGenerationCommitted": False,
}
pending_bindings.append(binding)
pending_records.append(record)
if len(pending_bindings) >= receipt_every:
commit_pending_chunk()
commit_pending_chunk()
summary = _write_compact_summary(
summary_path=summary_path,
journal_path=journal_path,
records=records,
store_root=store.root,
session_id=session_id,
source_checkpoint_sha256=source_checkpoint_sha256,
start_page_id=start_page_id,
requested_page_count=page_count,
page_parameter_elements=page_parameter_elements,
)
publish_compact_transfer_page_bank_locator_boundary(
summary_path,
registry_roots=registry_roots,
)
if writer_source_diagnostic is None:
return summary
return {
**summary,
"writerSourceSha256Diagnostic": writer_source_diagnostic,
"writerSourceSha256DiagnosticOnly": True,
}