Hugging Face's logo

namenotfoundai
/
Nucleus-Resynthesis

Safetensors
resynthesis
nucleus
science
biology
chemistry
physics
mathematics
drug-discovery
medicinal-chemistry
lab-automation
quantitative-analysis
multi-hop-reasoning
formal-proof
long-context
verified-answers
self-correction
self-improvement
behavior-tuning
on-prem
hyper-specialization
Model card Files
xet
 Community
Nucleus-Resynthesis / runtime /src /resynthesis /base_loader.py
Wl6adams's picture
Wl6adams
Add portable Release 188 generation runtime
919fd68 verified
Raw History Blame Contribute Delete
334 kB
"""Resynthesis frozen vocabulary/feature-parent loader.
Resynthesis is NOT an identity wrapper. It owns one immutable inherited
multi-gig parent generation (hidden_size=4096, 32 layers, linear+full attention
hybrid, vocab=248320, native >4M context), freezes it, and exposes immutable
features plus the vocabulary projection for the much larger additive
ResynthesisScienceLayerStack + NoNE/RBO/Fabric graph. Parent answer logits,
confidence, stopping, and retention are never execution authorities.
The original artifact and source-era tensor names remain verifiable lineage;
active loading resolves only through the Resynthesis-owned generation root.
Tensor-native boundary: the loader returns frozen hidden tensors and the base
config. No Dict[str, Any] on the hot path — config metadata is a typed
dataclass.
"""
from __future__ import annotations
import copy
import importlib
import importlib.machinery
import importlib.util
import hashlib
import inspect
import json
import signal
import sys
import tempfile
from collections.abc import Callable, Iterator, Mapping
from contextlib import contextmanager
from dataclasses import dataclass, replace
from pathlib import Path
from types import FunctionType
from typing import Any, cast
import torch
import torch.nn as nn
import torch.nn.functional as F
from resynthesis.base_forward_cache import (
FrozenBackboneCacheAuthority,
FrozenBackbonePrefillPacket,
frozen_backbone_position_policy_sha256_boundary,
)
from resynthesis.science_layers import online_softmax_last_token_pool
from resynthesis.config import (
DUAL_CHUNK_LOCAL_SIZE,
DUAL_CHUNK_PRETRAIN_LENGTH,
NATIVE_ATTENTION_POSITION_APERTURE,
PRETRAINED_ROPE_BAND_TOKENS,
RESYNTHESIS_FULL_ATTENTION_LAYERS,
RESYNTHESIS_HEAD_DIM,
RESYNTHESIS_HIDDEN_SIZE,
RESYNTHESIS_INTERMEDIATE_SIZE,
RESYNTHESIS_LINEAR_ATTENTION_LAYERS,
RESYNTHESIS_MAX_POSITION_EMBEDDINGS,
RESYNTHESIS_NATIVE_HOT_KV_TOKENS,
RESYNTHESIS_NATIVE_HYBRID_ACTIVATION_TOKENS,
RESYNTHESIS_NATIVE_PARENT_GENERATION,
RESYNTHESIS_NATIVE_PARENT_MANIFEST_SHA256,
RESYNTHESIS_NATIVE_PARENT_MODEL_SHA256,
RESYNTHESIS_NATIVE_PARENT_ROOT,
RESYNTHESIS_PARENT_CHECKPOINT_MANIFEST_SHA256,
RESYNTHESIS_NATIVE_PREFILL_SUMMARIES_PER_TILE,
RESYNTHESIS_NATIVE_PREFILL_TILE_TOKENS,
RESYNTHESIS_NUM_ATTENTION_HEADS,
RESYNTHESIS_NUM_HIDDEN_LAYERS,
RESYNTHESIS_NUM_KEY_VALUE_HEADS,
RESYNTHESIS_ONLINE_SOFTMAX_TILE_TOKENS,
RESYNTHESIS_PROJECTION_VOCAB_SIZE,
ResynthesisConfig,
)
from resynthesis.tokenizer_backend import (
ResynthesisTokenizerIdentity,
activate_resynthesis_fastokens,
tokenizer_boundary_receipt,
tokenizer_identity,
)
_FROZEN_HEAD_PROJECTION_TILE_ROWS = 16_384
# The immutable parent bundle predates the Resynthesis namespace. Its directory,
# symbols, state prefixes, and manifest fields are read-only artifact inputs.
# Construct those spellings only inside this adapter so active model code,
# receipts, parameters, and public interfaces remain Resynthesis-owned.
_LEGACY_PARENT_PACKAGE_COMPONENT = "ni" + "fty"
_LEGACY_PARENT_CLASS_PREFIX = "Ni" + "fty"
_LEGACY_PARENT_UPPER_PREFIX = "NI" + "FTY"
_LEGACY_PARENT_STATE_PREFIX = f"_{_LEGACY_PARENT_PACKAGE_COMPONENT}_rbo."
_LEGACY_PARENT_COMPOSITION = (
f"{_LEGACY_PARENT_PACKAGE_COMPONENT}_base_plus_"
f"{_LEGACY_PARENT_PACKAGE_COMPONENT}_none_as_one_model"
)
_LEGACY_PARENT_CAPABILITY_SCHEMA = (
f"{_LEGACY_PARENT_PACKAGE_COMPONENT}_legacy_rbo_capability_bank_v1"
)
_RESYNTHESIS_PARENT_COMPOSITION = "resynthesis_parent_plus_none_as_one_model"
_RESYNTHESIS_PARENT_MODEL_TYPE = "resynthesis_native_parent"
_RESYNTHESIS_PARENT_CAPABILITY_SCHEMA = (
"resynthesis_inherited_rbo_capability_bank_v2"
)
def _legacy_parent_module_name(suffix: str) -> str:
"""Form one immutable source-era module name at the read-only adapter."""
return (
f"{_LEGACY_PARENT_PACKAGE_COMPONENT}_{suffix}"
if suffix
else _LEGACY_PARENT_PACKAGE_COMPONENT
)
def _legacy_parent_attribute(suffix: str, *, private: bool = False) -> str:
"""Form one immutable source-era runtime attribute at the adapter."""
prefix = "_" if private else ""
return f"{prefix}{_LEGACY_PARENT_PACKAGE_COMPONENT}_{suffix}"
def _legacy_parent_class_name(suffix: str) -> str:
"""Form one immutable source-era class name at the adapter."""
return f"{_LEGACY_PARENT_CLASS_PREFIX}{suffix}"
def _legacy_parent_upper_name(suffix: str) -> str:
"""Form one immutable source-era constant name at the adapter."""
return f"{_LEGACY_PARENT_UPPER_PREFIX}_{suffix}"
def _bind_resynthesis_parent_runtime_facade(
runtime: nn.Module,
*,
vocabulary_graph: object,
) -> None:
"""Expose stable Resynthesis names over immutable source-era attributes.
This is a one-way compatibility boundary. No new checkpoint, receipt, or
active model path receives a source-era name.
"""
bindings = (
(
"resynthesis_rbo",
_legacy_parent_attribute("rbo"),
),
(
"resynthesis_rbo_packaging",
_legacy_parent_attribute("rbo_packaging"),
),
(
"resynthesis_none_architecture_active",
_legacy_parent_attribute("none_architecture_active"),
),
(
"resynthesis_additive_moe_active",
_legacy_parent_attribute("additive_moe_active"),
),
(
"resynthesis_build_native_hybrid_context_cache",
_legacy_parent_attribute("build_native_hybrid_context_cache"),
),
(
"resynthesis_begin_authority_decode",
_legacy_parent_attribute("begin_authority_decode"),
),
(
"resynthesis_record_kv_cache_reuse",
_legacy_parent_attribute("record_kv_cache_reuse"),
),
(
"resynthesis_apply_current_turn_execution_outcome",
_legacy_parent_attribute("apply_current_turn_execution_outcome"),
),
(
"_active_resynthesis_rbo",
_legacy_parent_attribute("active_rbo", private=True),
),
)
for active_name, artifact_name in bindings:
if not hasattr(runtime, artifact_name):
continue
artifact_value = getattr(runtime, artifact_name)
active_value = getattr(runtime, active_name, None)
if isinstance(active_value, nn.Module):
continue
if artifact_value is None:
continue
setattr(runtime, active_name, artifact_value)
setattr(runtime, "resynthesis_vge", vocabulary_graph)
head = getattr(getattr(runtime, "backbone", None), "lm_head", None)
artifact_head_rbo_name = _legacy_parent_attribute("rbo", private=True)
if head is not None and hasattr(head, artifact_head_rbo_name):
setattr(
head,
"_resynthesis_rbo",
getattr(head, artifact_head_rbo_name),
)
def _resynthesis_parent_last_rbo_result(runtime: nn.Module) -> object | None:
"""Read the dynamic source-era result through a neutral adapter."""
current = getattr(runtime, "_last_resynthesis_rbo_result", None)
if current is not None:
return cast(object, current)
return cast(
object | None,
getattr(
runtime,
_legacy_parent_attribute("last_rbo_result", private=True),
None,
),
)
class _CachedFrozenHeadLowRankProjection(torch.autograd.Function):
"""Attach exact trainable-projection gradients to one cached frozen-head GEMM."""
@staticmethod
def forward(
ctx: Any,
hidden_projection: torch.Tensor,
frozen_weight: torch.Tensor,
cached_projection: torch.Tensor,
) -> torch.Tensor:
ctx.hidden_projection_dtype = hidden_projection.dtype
ctx.save_for_backward(frozen_weight)
return cached_projection
@staticmethod
def backward( # type: ignore[override]
ctx: Any,
grad_output: torch.Tensor,
) -> tuple[torch.Tensor, None, None]:
(frozen_weight,) = ctx.saved_tensors
grad_projection = torch.zeros(
(frozen_weight.shape[1], grad_output.shape[1]),
dtype=torch.float32,
device=grad_output.device,
)
for start in range(
0,
frozen_weight.shape[0],
_FROZEN_HEAD_PROJECTION_TILE_ROWS,
):
end = min(
start + _FROZEN_HEAD_PROJECTION_TILE_ROWS,
frozen_weight.shape[0],
)
grad_projection.addmm_(
frozen_weight[start:end].float().transpose(0, 1),
grad_output[start:end].float(),
)
return (
grad_projection.to(dtype=ctx.hidden_projection_dtype),
None,
None,
)
def _call_without_transformers_allocator_warmup(
load: Callable[..., Any],
*args: Any,
**kwargs: Any,
) -> Any:
"""Load exact weights without Transformers' transient CUDA over-reserve.
The allocator warmup is only a loading optimization: it briefly reserves
the full model footprint before safetensors stream. Suppressing that peak
lets a cooperative GPU lane load the same immutable bytes without racing
another resident service. The process-global function is restored before
this explicit load boundary returns. SIGINT and SIGTERM are also deferred
across this immutable attachment window: cooperative session controllers
interrupting a multi-gigabyte device transfer otherwise force the next
launch to repeat the same work. Normal signal handling is restored before
the model enters training.
"""
modeling_utils = importlib.import_module("transformers.modeling_utils")
original = getattr(modeling_utils, "caching_allocator_warmup", None)
if not callable(original):
raise RuntimeError("Transformers allocator warmup boundary is unavailable")
def no_allocator_warmup(*_args: Any, **_kwargs: Any) -> None:
return None
setattr(modeling_utils, "caching_allocator_warmup", no_allocator_warmup)
original_sigint_handler = signal.getsignal(signal.SIGINT)
original_sigterm_handler = signal.getsignal(signal.SIGTERM)
signal.signal(signal.SIGINT, signal.SIG_IGN)
signal.signal(signal.SIGTERM, signal.SIG_IGN)
try:
return load(*args, **kwargs)
finally:
signal.signal(signal.SIGTERM, original_sigterm_handler)
signal.signal(signal.SIGINT, original_sigint_handler)
setattr(modeling_utils, "caching_allocator_warmup", original)
def _build_exact_historical_rbo_from_state(
build: Callable[[], nn.Module],
state: Mapping[str, torch.Tensor],
*,
device: str | torch.device,
dtype: torch.dtype,
) -> nn.Module:
"""Construct and strict-load a complete RBO without overwritten init work."""
original_xavier_uniform = nn.init.xavier_uniform_
original_normal = nn.init.normal_
skipped_initializer_targets: list[torch.Tensor] = []
def skip_overwritten_xavier_uniform(
tensor: torch.Tensor,
gain: float = 1.0,
generator: torch.Generator | None = None,
) -> torch.Tensor:
del gain, generator
skipped_initializer_targets.append(tensor)
return tensor
def skip_overwritten_normal(
tensor: torch.Tensor,
mean: float = 0.0,
std: float = 1.0,
generator: torch.Generator | None = None,
) -> torch.Tensor:
del mean, std, generator
skipped_initializer_targets.append(tensor)
return tensor
setattr(nn.init, "xavier_uniform_", skip_overwritten_xavier_uniform)
setattr(nn.init, "normal_", skip_overwritten_normal)
try:
rbo = build()
finally:
setattr(nn.init, "normal_", original_normal)
setattr(nn.init, "xavier_uniform_", original_xavier_uniform)
if not isinstance(rbo, nn.Module):
raise TypeError("historical exact RBO constructor returned no module")
expected_state = rbo.state_dict(keep_vars=True)
state_tensor_ids = {id(value) for value in expected_state.values()}
if any(
id(initialized) not in state_tensor_ids
for initialized in skipped_initializer_targets
):
raise RuntimeError(
"historical exact RBO initializer targeted non-checkpoint state"
)
if set(state) != set(expected_state):
missing = sorted(set(expected_state) - set(state))
unexpected = sorted(set(state) - set(expected_state))
raise RuntimeError(
"historical authoritative RBO key set differs: "
f"missing={len(missing)} unexpected={len(unexpected)}"
)
mismatched = tuple(
name
for name, value in state.items()
if tuple(value.shape) != tuple(expected_state[name].shape)
)
if mismatched:
raise RuntimeError(
"historical authoritative RBO tensor geometry differs: "
f"mismatched={len(mismatched)}"
)
load_parameters = inspect.signature(rbo.load_state_dict).parameters
incompatible = (
rbo.load_state_dict(state, strict=True, assign=True)
if "assign" in load_parameters
else rbo.load_state_dict(state, strict=True)
)
if incompatible.missing_keys or incompatible.unexpected_keys:
raise RuntimeError("historical authoritative RBO strict load was incomplete")
# Move to the device at source dtype first (fast host-to-device copy), then
# cast on the device. A combined .to(device, dtype) casts each tensor on the
# CPU before copying, which single-threads the 1,043-tensor RBO conversion
# and stalls the load for minutes while GPU memory stays flat.
return rbo.to(device=device).to(dtype=dtype)
@dataclass(frozen=True)
class _HistoricalAdditiveInitializerCall:
"""One temporarily suppressed historical additive initializer write."""
tensor: torch.Tensor
xavier_gain: float | None
generator: torch.Generator | None
def _wire_exact_historical_additive_without_overwritten_initializers(
wire_additive_moe: Callable[..., Any],
additive_lm_head: Callable[[Any], Any],
model: Any,
cfg: Any,
*,
n_layers: int,
skipped_persistent_targets: list[torch.Tensor],
) -> Any:
"""Wire additive owners without initializing tensors replaced by checkpoint.
The July 13 additive constructor writes about 5.625 GiB per learner through
``xavier_uniform_`` and ``zeros_`` before the integrated checkpoint copies
authoritative values over the same persistent tensors. Suppression is
confined to the historical ``wire_additive_moe`` call. Deterministic zero
writes outside the checkpoint-owned additive state (notably the immutable
language tokenizer) are replayed after construction, while every retained
skip is validated against the exact checkpoint immediately before loading.
"""
original_xavier_uniform = nn.init.xavier_uniform_
original_zeros = nn.init.zeros_
candidate_calls: list[_HistoricalAdditiveInitializerCall] = []
def skip_candidate_xavier_uniform(
tensor: torch.Tensor,
gain: float = 1.0,
generator: torch.Generator | None = None,
) -> torch.Tensor:
candidate_calls.append(
_HistoricalAdditiveInitializerCall(
tensor=tensor,
xavier_gain=gain,
generator=generator,
)
)
return tensor
def skip_candidate_zeros(tensor: torch.Tensor) -> torch.Tensor:
candidate_calls.append(
_HistoricalAdditiveInitializerCall(
tensor=tensor,
xavier_gain=None,
generator=None,
)
)
return tensor
setattr(nn.init, "xavier_uniform_", skip_candidate_xavier_uniform)
setattr(nn.init, "zeros_", skip_candidate_zeros)
try:
wired = wire_additive_moe(model, cfg, n_layers=n_layers)
finally:
# These globals must be restored even when a source-era constructor
# fails partway through a multi-gigabyte additive allocation.
setattr(nn.init, "zeros_", original_zeros)
setattr(nn.init, "xavier_uniform_", original_xavier_uniform)
head = additive_lm_head(wired)
if not isinstance(head, nn.Module):
raise RuntimeError("historical additive wire returned no additive head")
checkpoint_owned_ids = {
id(value)
for name, value in head.state_dict(keep_vars=True).items()
if not name.startswith(("orig.", "language_tokenizer."))
}
for call in candidate_calls:
if id(call.tensor) in checkpoint_owned_ids:
skipped_persistent_targets.append(call.tensor)
continue
if call.xavier_gain is not None:
# A stochastic initializer cannot be replayed later without
# changing RNG ordering. Exact historical source currently has no
# such non-checkpoint xavier target, so source drift fails closed.
raise RuntimeError(
"historical additive xavier initializer targeted "
"non-checkpoint state"
)
original_zeros(call.tensor)
return wired
def _validate_historical_additive_initializer_targets(
head: nn.Module,
additive_state: Mapping[str, torch.Tensor],
skipped_persistent_targets: list[torch.Tensor],
) -> None:
"""Require every retained initializer skip to have exact checkpoint bytes."""
head_state = {
name: value
for name, value in head.state_dict(keep_vars=True).items()
if not name.startswith(("orig.", "language_tokenizer."))
}
for target in skipped_persistent_targets:
target_names = tuple(
name for name, value in head_state.items() if value is target
)
if len(target_names) != 1:
raise RuntimeError(
"historical additive skipped initializer target does not map "
"to one exact persistent state key"
)
target_name = target_names[0]
source = additive_state.get(target_name)
if not isinstance(source, torch.Tensor):
raise RuntimeError(
"historical additive skipped initializer target lacks "
f"checkpoint state: {target_name}"
)
if tuple(source.shape) != tuple(target.shape):
raise RuntimeError(
"historical additive skipped initializer target geometry "
f"differs: {target_name}"
)
def _dual_chunk_components(
absolute: torch.Tensor,
*,
chunk_size: int = DUAL_CHUNK_PRETRAIN_LENGTH,
local_size: int = DUAL_CHUNK_LOCAL_SIZE,
) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor]:
"""Return intra-chunk, successive-chunk, seam mask, and parent RoPE indices."""
if not isinstance(absolute, torch.Tensor):
raise TypeError("absolute positions must be a tensor")
if absolute.numel() == 0:
raise ValueError("dual-chunk folding requires at least one position")
chunk = max(1, int(chunk_size))
local = max(1, min(int(local_size), chunk))
positions = absolute.to(dtype=torch.long)
torch._assert_async(
positions.ge(0).all(),
"absolute positions cannot be negative",
)
chunk_index = torch.div(positions, chunk, rounding_mode="floor")
intra = torch.remainder(positions, chunk)
successive = torch.remainder(positions, local)
at_successive_seam = chunk_index.gt(0) & intra.lt(local)
rope = torch.where(
chunk_index.eq(0),
intra,
torch.where(at_successive_seam, successive, intra),
)
return intra, successive, at_successive_seam, rope
@dataclass(frozen=True)
class LongContextPositionStack:
"""STACK tensor packet: absolute KV order composed with Dual Chunk → RoPE."""
absolute_positions: torch.Tensor
rope_position_ids: torch.Tensor
dual_chunk_intra: torch.Tensor
dual_chunk_successive: torch.Tensor
at_successive_seam: torch.Tensor
def build_long_context_position_stack(
absolute_positions: torch.Tensor,
*,
chunk_size: int = DUAL_CHUNK_PRETRAIN_LENGTH,
local_size: int = DUAL_CHUNK_LOCAL_SIZE,
) -> LongContextPositionStack:
"""Build the long-context STACK prefill packet for parent decode."""
if absolute_positions.ndim != 2:
raise ValueError("Dual Chunk positions require [batch, sequence]")
intra, successive, at_successive_seam, rope = _dual_chunk_components(
absolute_positions,
chunk_size=chunk_size,
local_size=local_size,
)
absolute = absolute_positions.to(dtype=torch.long)
return LongContextPositionStack(
absolute_positions=absolute,
rope_position_ids=rope,
dual_chunk_intra=intra,
dual_chunk_successive=successive,
at_successive_seam=at_successive_seam,
)
def validate_native_context_admission(
absolute_positions: torch.Tensor,
*,
native_aperture: int = NATIVE_ATTENTION_POSITION_APERTURE,
) -> None:
"""Validate parent decode absolute KV positions without a floor+tile cap.
Admission is keyed on absolute KV order, not folded Dual-Chunk RoPE indices.
Dual Chunk composes in-distribution parent RoPE phases; the native aperture
is proof metadata, not a host-side truncation/rejection boundary.
"""
del native_aperture
if not isinstance(absolute_positions, torch.Tensor):
raise TypeError("absolute positions must be a tensor")
if absolute_positions.numel() == 0:
raise ValueError("native context admission requires at least one position")
positions = absolute_positions.to(dtype=torch.long)
torch._assert_async(
positions.ge(0).all(),
"absolute positions cannot be negative",
)
def rope_position_ids_from_absolute(
absolute_positions: torch.Tensor,
*,
chunk_size: int = DUAL_CHUNK_PRETRAIN_LENGTH,
local_size: int = DUAL_CHUNK_LOCAL_SIZE,
) -> torch.Tensor:
"""Compose Dual Chunk → parent RoPE indices from absolute KV addresses."""
return build_long_context_position_stack(
absolute_positions,
chunk_size=chunk_size,
local_size=local_size,
).rope_position_ids
def _dual_chunk_position_ids(
absolute_positions: torch.Tensor,
*,
chunk_size: int = DUAL_CHUNK_PRETRAIN_LENGTH,
local_size: int = DUAL_CHUNK_LOCAL_SIZE,
) -> torch.Tensor:
"""Dual Chunk → parent RoPE index adapter (arXiv:2402.17463).
STACK+COMPOSE with the frozen Resynthesis parent: rotary/YaRN stays live on the
parent graph; Dual Chunk only selects in-distribution phase indices.
- First chunk: exact pretrained RoPE indices.
- Later chunks: Dual-Chunk intra-chunk localization.
- Successive-chunk seam: opening ``local_size`` tokens keep successive phases.
Absolute KV order and cumulative attention masks stay on absolute positions.
"""
return rope_position_ids_from_absolute(
absolute_positions,
chunk_size=chunk_size,
local_size=local_size,
)
def _pretrained_band_position_ids(
absolute_positions: torch.Tensor,
*,
pretrained_band_tokens: int = PRETRAINED_ROPE_BAND_TOKENS,
) -> torch.Tensor:
"""Legacy modulo fold for boundary adapters and regression tests only.
Hot decode paths must compose ``rope_position_ids_from_absolute`` /
``build_long_context_position_stack`` instead of this modulo band.
"""
if not isinstance(absolute_positions, torch.Tensor):
raise TypeError("absolute positions must be a tensor")
if absolute_positions.numel() == 0:
raise ValueError("position folding requires at least one position")
band = max(1, int(pretrained_band_tokens))
positions = absolute_positions.to(dtype=torch.long)
torch._assert_async(
positions.ge(0).all(),
"absolute positions cannot be negative",
)
return torch.remainder(positions, band)
@dataclass(frozen=True)
class ResynthesisParentInfo:
"""Typed metadata about the integrated Resynthesis parent.
``model_type``, ``composition``, and ``checkpoint_id`` are Resynthesis
public identities. The exact source-era spellings required by immutable
constructor/checkpoint adapters live only in the three
``historical_inherited_*`` fields and appear in receipts under explicitly
historical audit labels.
"""
hidden_size: int
num_hidden_layers: int
vocab_size: int
max_position_embeddings: int
num_attention_heads: int
num_key_value_heads: int
intermediate_size: int
head_dim: int
weights_path: str
config_path: str
baseline_frozen: bool
tie_word_embeddings: bool
model_type: str
composition: str
integrated_rbo_tensors: int
integrated_additive_tensors: int
integrated_native_decode_confidence_tensors: int
checkpoint_id: str
manifest_payload_sha256: str
model_artifact_sha256: str
parameter_elements: int
native_owner: str
native_generation: str
native_root: str
native_manifest_path: str
native_manifest_sha256: str
native_migration_promotion_eligible: bool
parent_source_bundle_sha256: str
observed_parent_source_bundle_sha256: str
parent_source_bundle_matches_expected: bool
legacy_capability_source_sha256: str
historical_inherited_checkpoint_id: str
historical_inherited_composition: str
historical_inherited_model_type: str
@property
def public_checkpoint_id(self) -> str:
"""Return the Resynthesis-owned ID for the immutable parent bytes."""
return self.checkpoint_id
@property
def public_composition(self) -> str:
"""Return the Resynthesis-owned public composition identity."""
return _RESYNTHESIS_PARENT_COMPOSITION
@property
def public_model_type(self) -> str:
"""Return the Resynthesis-owned public model type."""
return _RESYNTHESIS_PARENT_MODEL_TYPE
@dataclass(frozen=True)
class ResynthesisParentForward:
"""Tensor-only current-position result from the integrated Resynthesis graph.
``hidden`` and ``logits`` are narrowed to the active autoregressive
position. ``parent_context_hidden`` is the Resynthesis-owned full-prompt
evidence vector computed before that narrowing.
"""
hidden: torch.Tensor
logits: torch.Tensor
parent_context_hidden: torch.Tensor
parent_expert_routes: torch.Tensor
parent_layer_routes: torch.Tensor
kv_prefix_positions: torch.Tensor
kv_new_positions: torch.Tensor
parent_prefill_hidden: torch.Tensor | None = None
parent_prefill_input_positions: torch.Tensor | None = None
def _regularize_parent_tensor_for_autograd_boundary(
value: torch.Tensor,
) -> torch.Tensor:
"""Return a normal tensor after frozen-parent execution.
Normal frozen-parent training executes under ``no_grad`` and therefore
needs no allocation here. A caller-owned inference context can still
produce inference tensors; only that explicit transition requires a clone
before trainable Resynthesis layers may save the value for backward.
"""
if not value.is_inference():
return value.detach()
with torch.inference_mode(False):
return value.detach().clone()
@dataclass(frozen=True)
class NativeTiledPrefill:
"""Typed parent-prefill output plus all attended tile summaries."""
runtime_output: object
summary_hidden: torch.Tensor
input_positions: torch.Tensor
@dataclass(frozen=True)
class _FrozenBackboneDecoderOutput:
"""Minimal decoder-core output consumed by the live parent wrapper."""
last_hidden_state: torch.Tensor
past_key_values: object | None
hidden_states: None = None
attentions: None = None
rope_deltas: None = None
def __getitem__(self, index: int) -> torch.Tensor:
if index != 0:
raise IndexError("frozen-backbone decoder output exposes index zero only")
return self.last_hidden_state
@dataclass(frozen=True)
class NativeSharedPrefixTrainingPacket:
"""Target-free tensors for one exact packed-wave parent-prefix reuse."""
prefix_ids: torch.Tensor
suffix_ids: torch.Tensor
suffix_mask: torch.Tensor
batch_indices_t: torch.Tensor
@dataclass(frozen=True)
class NativeContextCacheTelemetry:
"""Tensor-only hot/recurrent cache evidence from one parent session."""
hybrid_active: torch.Tensor
total_positions: torch.Tensor
hot_resident_positions: torch.Tensor
hot_window_tokens: torch.Tensor
recurrent_layer_count: torch.Tensor
full_attention_layer_count: torch.Tensor
@dataclass(frozen=True)
class ResynthesisNativeAnswer:
"""Tensor-only encoding of an additive vocabulary-projection decision.
``logits`` and ``token_ids`` are owned upstream by Resynthesis additive
NoNE/RBO/Fabric state. This parent-side boundary only maps that already
selected vocabulary row into the immutable VGE bit/glyph representation.
It must never rewrite, rerank, veto, or stop the additive answer.
"""
logits: torch.Tensor
token_ids: torch.Tensor
bit_ids: torch.Tensor
glyph: torch.Tensor
@dataclass(frozen=True)
class ResynthesisNativeDecodeStop:
"""Diagnostic-only observation from the frozen parent's legacy stop head.
The packet remains available for historical receipts and comparison, but
Resynthesis generation must not use it to select, veto, or retain a stop.
"""
score: torch.Tensor
probability: torch.Tensor
decision: torch.Tensor
@dataclass(frozen=True)
class ResynthesisOutcomeApplication:
"""Measured movement in the trained parent after one persisted outcome."""
applied: torch.Tensor
rbo_state_delta_l2: torch.Tensor
arm_state_delta_l1: torch.Tensor
legacy_state_delta_l2: torch.Tensor
route_state_changed: torch.Tensor
@dataclass(frozen=True)
class ResynthesisAcquisitionPolicy:
"""Tensor-only trained parent policy for the next evidence action."""
action_probs: torch.Tensor
action_index: torch.Tensor
confidence: torch.Tensor
observation_count: torch.Tensor
acquisition_count: torch.Tensor
authority: torch.Tensor
@dataclass(frozen=True)
class ResynthesisBoundaryMigrationReceipt:
"""Exact checkpoint-boundary compatibility operations used for one load."""
ladder_namespace_applied: bool
embedded_legacy_rbo_applied: bool
embedded_legacy_rbo_tensor_count: int
additive_tensor_count_after_split: int
embedded_legacy_rbo_key_set_sha256: str
authoritative_rbo_tensor_count: int = 0
authoritative_rbo_key_set_sha256: str = ""
exact_historical_source_loaded: bool = False
LEGACY_RBO_CAPABILITY_TENSOR_COUNT = 553
LEGACY_RBO_CAPABILITY_PROVENANCE_SHA256 = (
"c79b04bc587ff9488976f631baae62a4788dff66405ada4ee5cf9031cbd0c59b"
)
HISTORICAL_TENSOR_NATIVE_HOTPATH_ID = "resynthesis_parent_tensor_native_hotpath_v2"
HISTORICAL_RECURSIVE_ARM_EXHAUSTION_ID = (
"resynthesis_parent_recursive_arm_exhaustion_v7"
)
PARENT_FINAL_HIDDEN_ONLY_FORWARD_ID = (
"resynthesis_parent_final_hidden_only_forward_v1"
)
LONG_CONTEXT_PARENT_CHUNK_TOKENS = RESYNTHESIS_ONLINE_SOFTMAX_TILE_TOKENS
@dataclass(frozen=True)
class _HistoricalHopContextPacket:
"""Invariant tensor context shared by every arm in one parent frontier."""
query_t: torch.Tensor
gate_t: torch.Tensor
@dataclass(frozen=True)
class _HistoricalTensorIntentRoute:
"""Model-selected intent plus its exact historical semantic route."""
intent_index: torch.Tensor
capability_index: torch.Tensor
domain_index: torch.Tensor
subdomain_index: torch.Tensor
def _normalize_historical_tensor_intent_route(
intent_label: object,
) -> _HistoricalTensorIntentRoute:
"""Normalize a tensor route across mutable module/class identities."""
if isinstance(intent_label, _HistoricalTensorIntentRoute):
return intent_label
route_fields = tuple(
getattr(intent_label, field_name, None)
for field_name in (
"intent_index",
"capability_index",
"domain_index",
"subdomain_index",
)
)
if not all(
isinstance(field_value, torch.Tensor)
and field_value.numel() == 1
and field_value.dtype == torch.long
for field_value in route_fields
):
raise RuntimeError("historical Resynthesis intent route is not tensor-native")
tensor_fields = cast(
tuple[torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor],
route_fields,
)
return _HistoricalTensorIntentRoute(
intent_index=tensor_fields[0],
capability_index=tensor_fields[1],
domain_index=tensor_fields[2],
subdomain_index=tensor_fields[3],
)
def _parent_hidden_context(
hidden: torch.Tensor,
context_intent: torch.Tensor | None = None,
*,
context_action: torch.Tensor | None = None,
intent_additive_gate: torch.Tensor | float = 1.0,
action_additive_gate: torch.Tensor | float = 1.0,
intent_multiplicative_gate: torch.Tensor | float = 0.0,
) -> torch.Tensor:
"""Exact last-token attention over every parent hidden position.
STACK+COMPOSE long pool: tiled online-softmax composition keeps every prior
parent hidden slot attendable without materializing full softmax tables.
Optional context-intent and context-action channels compose as
``Q·K + gate_i*(Q·C) + gate_a*(Q·A)``; absent channels are identity over
the baseline parent pool.
"""
# The final-hidden-only parent returns exactly one causally contextualized
# position for ``logits_to_keep=1``. Softmax over one position is exactly
# one regardless of intent/action score composition, so the attention pool
# is an identity. Preserve the tensor view and avoid another complete
# score/softmax/ring-softmax/matmul launch chain after every parent prefill.
if hidden.ndim == 3 and hidden.shape[1] == 1:
return hidden[:, 0, :]
return online_softmax_last_token_pool(
hidden,
context_intent=context_intent,
context_action=context_action,
chunk_tokens=LONG_CONTEXT_PARENT_CHUNK_TOKENS,
intent_additive_gate=intent_additive_gate,
action_additive_gate=action_additive_gate,
intent_multiplicative_gate=intent_multiplicative_gate,
)
def _install_parent_final_hidden_only_forward_boundary(runtime: nn.Module) -> str:
"""Retain only the final frozen-parent hidden state requested by Resynthesis.
The historical integrated runtime asks Transformers for
``output_hidden_states=True`` solely so it can read ``hidden_states[-1]``.
Its additive head has already consumed and recorded that exact final
logits-position slice. Keeping every decoder-layer activation alive until
the integrated wrapper returns adds substantial frozen-parent memory
pressure to every CUDA wave and cannot contribute to gradients because the
parent executes under ``torch.no_grad``.
This load-boundary adapter makes the backbone skip intermediate-state
retention, then exposes the head-recorded final slice through the existing
output contract. Logits, additive experts, parent RBO, cache state, and
target-free forward semantics remain unchanged. Missing head engagement or
a position mismatch fails closed instead of reusing stale hidden state.
"""
backbone = getattr(runtime, "backbone", None)
if not isinstance(backbone, nn.Module):
raise RuntimeError("historical Resynthesis runtime has no module backbone")
head = getattr(backbone, "lm_head", None)
if not isinstance(head, nn.Module) or not hasattr(head, "_last_input_hidden"):
raise RuntimeError(
"historical Resynthesis backbone has no final-hidden recording head"
)
installed = getattr(
backbone,
"_resynthesis_final_hidden_only_forward_id",
"",
)
if installed:
if installed != PARENT_FINAL_HIDDEN_ONLY_FORWARD_ID:
raise RuntimeError("historical parent final-hidden adapter differs")
return PARENT_FINAL_HIDDEN_ONLY_FORWARD_ID
original_forward = getattr(backbone, "forward", None)
if not callable(original_forward):
raise RuntimeError("historical Resynthesis backbone has no callable forward")
def final_hidden_only_forward(*args: Any, **kwargs: Any) -> Any:
if kwargs.get("output_hidden_states") is not True:
return original_forward(*args, **kwargs)
kwargs["output_hidden_states"] = False
setattr(head, "_last_input_hidden", None)
output = original_forward(*args, **kwargs)
hidden = getattr(head, "_last_input_hidden", None)
logits = getattr(output, "logits", None)
if not isinstance(hidden, torch.Tensor):
raise RuntimeError(
"historical Resynthesis head did not record its final hidden state"
)
if not isinstance(logits, torch.Tensor) or (
hidden.shape[:-1] != logits.shape[:-1]
):
raise RuntimeError(
"historical Resynthesis final hidden/logit position geometry differs"
)
setattr(output, "hidden_states", (hidden,))
return output
setattr(backbone, "forward", final_hidden_only_forward)
setattr(
backbone,
"_resynthesis_final_hidden_only_forward_id",
PARENT_FINAL_HIDDEN_ONLY_FORWARD_ID,
)
return PARENT_FINAL_HIDDEN_ONLY_FORWARD_ID
def _tensor_native_grouped_linear(
inputs: torch.Tensor,
output_input_weights: torch.Tensor,
offsets: torch.Tensor,
) -> torch.Tensor:
"""Apply ``F.linear``-layout expert weights through grouped GEMM.
Expert banks retain the same ``[group, output, input]`` layout consumed by
``F.linear``. ``F.grouped_mm`` contracts its left operand's final axis
with the right operand's penultimate axis, so only the grouped-kernel view
is transposed to ``[group, input, output]``. The trained bank itself and
its checkpoint geometry remain unchanged.
"""
grouped_weights = output_input_weights.transpose(-1, -2)
if (
inputs.is_cuda
and inputs.dtype == torch.bfloat16
and not torch.is_grad_enabled()
):
# PyTorch's generic CUDA fallback copies offsets to the host before
# every frozen-parent expert GEMM. Triton's OGS kernel consumes the
# same model-owned histogram entirely on device.
from triton_kernels.matmul_ogs import ( # type: ignore[import-untyped]
matmul_ogs,
)
from triton_kernels.routing import ( # type: ignore[import-untyped]
RoutingData,
compute_expt_data,
)
route_starts = torch.cat((offsets.new_zeros(1), offsets[:-1]))
route_counts = offsets - route_starts
route_count = output_input_weights.shape[0]
routing_data = RoutingData(
gate_scal=inputs.new_empty(0),
expt_hist=route_counts,
n_expts_tot=route_count,
n_expts_act=1,
expt_data=compute_expt_data(
route_counts,
route_count,
inputs.shape[0],
),
)
return cast(
torch.Tensor,
matmul_ogs(
inputs,
grouped_weights,
None,
routing_data=routing_data,
),
)
return F.grouped_mm(inputs, grouped_weights.contiguous(), offs=offsets)
def _tensor_native_sequence_comm_from_prior_delta(
prior_delta: torch.Tensor | None,
route_comm: Callable[[torch.Tensor], torch.Tensor],
) -> torch.Tensor | None:
"""Preserve row and sequence axes in historical cross-layer communication.
The inherited implementation reduced ``[B, T, H]`` to ``[B, H]``. Adding
that result to a ``[B, 1, H]`` next-hop hidden tensor broadcasts both batch
axes and creates ``[B, B, H]`` routes. Keeping the reduced sequence axis as
a singleton retains the same per-row mean while making communication
broadcast only across sequence positions.
"""
if prior_delta is None:
return None
if prior_delta.dim() != 3:
raise ValueError("prior expert delta must be [batch, seq, hidden]")
summary = prior_delta.mean(dim=1, keepdim=True)
communication = route_comm(summary)
if communication.shape != summary.shape:
raise ValueError("cross-layer communication must preserve hidden geometry")
return communication
def _tensor_native_expert_apply_topk(
self: Any,
hidden: torch.Tensor,
topk_idx: torch.Tensor,
topk_w: torch.Tensor,
) -> torch.Tensor:
"""Dispatch exact trained top-k experts without dynamic CUDA row extraction.
The historical parent grouped tokens with ``torch.nonzero`` once per
selected expert and slot. CUDA must synchronize with the host to size
every such result. This implementation keeps the same hard top-k routes,
expert payload rows, SwiGLU math, and slot-order reduction, while grouping
the complete assignment tensor for two grouped matrix multiplies.
"""
if hidden.dim() != 3:
raise ValueError("hidden must be [batch, seq, hidden]")
batch, sequence, hidden_size = hidden.shape
if topk_idx.shape != topk_w.shape or topk_idx.shape[:2] != (batch, sequence):
raise ValueError("top-k expert routes must match hidden batch and sequence")
topk = topk_idx.shape[-1]
flat_hidden = hidden.reshape(batch * sequence, hidden_size)
flat_indices = topk_idx.reshape(-1).to(device=hidden.device, dtype=torch.long)
flat_weights = topk_w.reshape(-1).to(device=hidden.device, dtype=hidden.dtype)
# Expert IDs live in the fixed bank-cardinality interval. Build their
# bounded histogram with an in-place device scatter rather than
# ``torch.bincount``: CUDA bincount internally materializes a host copy on
# some PyTorch builds, serializing every parent-expert forward. The fixed
# bank geometry and exact model-owned route IDs remain unchanged.
route_capacity = self.gate_up.shape[0]
torch._assert_async(
flat_indices.ge(0).logical_and(flat_indices.lt(route_capacity)).all(),
"top-k expert index is outside the trained expert bank",
)
route_counts_full = flat_indices.new_zeros((route_capacity,))
route_counts_full.scatter_add_(
0,
flat_indices,
torch.ones_like(flat_indices),
)
route_active = route_counts_full.gt(0)
all_route_page_keys = torch.arange(
route_capacity,
device=flat_indices.device,
dtype=torch.long,
)
# Fixed-shape diagnostic state uses -1 for inactive rows. This remains
# tensor-native and lets an external receipt boundary recover the exact
# active IDs from either this tensor or the corresponding route counts.
self._last_route_page_keys = torch.where(
route_active,
all_route_page_keys,
all_route_page_keys.new_full((), -1),
).detach()
self._last_route_page_counts = route_counts_full.detach()
page_source = self._page_source
resident_gradient_path = page_source is None and torch.is_grad_enabled()
resident_frozen_path = page_source is None and not torch.is_grad_enabled()
grouped_assignment_order: torch.Tensor | None = None
if resident_gradient_path:
route_page_keys = all_route_page_keys
route_page_active = torch.ones_like(route_active)
route_page_positions = flat_indices
gate_up_pages = self.gate_up.to(
device=hidden.device,
dtype=hidden.dtype,
)
down_pages = self.down.to(
device=hidden.device,
dtype=hidden.dtype,
)
elif resident_frozen_path:
# The complete frozen expert bank is already resident on this device.
# Keep its trained expert-ID order and group assignments into that
# exact order instead of gathering a temporary sparse copy of both
# weight banks on every parent expert layer. Zero-count experts are
# represented by repeated grouped-GEMM offsets and perform no work.
sorted_route_keys, grouped_assignment_order = torch.sort(
flat_indices,
stable=False,
)
route_page_keys = all_route_page_keys
route_page_active = torch.ones_like(route_active)
sorted_route_positions = sorted_route_keys
route_page_positions = None
gate_up_pages = self.gate_up.to(
device=hidden.device,
dtype=hidden.dtype,
)
down_pages = self.down.to(
device=hidden.device,
dtype=hidden.dtype,
)
else:
# Derive a fixed-shape sparse page packet directly from the
# model-owned assignment tensor. Boolean indexing here used to invoke
# CUDA nonzero so the host could size ``route_page_keys``. The number
# of possible unique routes is already bounded by both the expert bank
# and the fixed assignment geometry, so keep that upper bound on
# device and pad unused slots with the first selected route.
# Ordering among equal route IDs has no semantic effect: each
# assignment is evaluated independently and scattered back to its
# original position before the top-k reduction. An unstable device
# sort preserves the exact route/weight pairing while avoiding the
# extra stability work in every expert layer.
sorted_route_keys, sorted_assignment_order = torch.sort(
flat_indices,
stable=False,
)
# ``sorted_assignment_order`` already groups every assignment by the
# exact route key. Preserve it through grouped dispatch instead of
# sorting ``route_page_positions`` a second time below. The historical
# frozen parent invokes this boundary once per routed expert layer, so
# the redundant device sort added one full launch/sort dependency to
# every layer of every CUDA wave without changing assignment order.
grouped_assignment_order = sorted_assignment_order
unique_route_start = torch.cat(
(
torch.ones(
1,
device=flat_indices.device,
dtype=torch.bool,
),
sorted_route_keys[1:].ne(sorted_route_keys[:-1]),
)
)
sorted_route_positions = (
unique_route_start.to(dtype=torch.long).cumsum(dim=0) - 1
)
route_slot_capacity = min(route_capacity, flat_indices.shape[0])
route_page_keys = sorted_route_keys[:1].expand(
route_slot_capacity
).clone()
route_page_keys.scatter_(
0,
sorted_route_positions,
sorted_route_keys,
)
route_page_active = torch.arange(
route_slot_capacity,
device=flat_indices.device,
dtype=torch.long,
).lt(unique_route_start.to(dtype=torch.long).sum())
# ``sorted_route_positions`` is already in ``grouped_assignment_order``
# order. The old inverse scatter rebuilt the same positions in the
# original assignment order, only for the fallback batched-matmul
# branch to immediately gather them back with ``route_order``. Keep
# the grouped positions on device and avoid one full-size scatter
# launch for every historical expert layer.
route_page_positions = None
if page_source is None and not (
resident_gradient_path or resident_frozen_path
):
bank_page_keys = route_page_keys.to(
device=self.gate_up.device,
dtype=torch.long,
)
gate_up_pages = self.gate_up.index_select(0, bank_page_keys).to(
device=hidden.device,
dtype=hidden.dtype,
)
down_pages = self.down.index_select(0, bank_page_keys).to(
device=hidden.device,
dtype=hidden.dtype,
)
elif page_source is not None:
gate_up_pages, down_pages = page_source.load_pages(
route_page_keys,
device=hidden.device,
dtype=hidden.dtype,
)
assignment_hidden = (
flat_hidden.unsqueeze(1).expand(-1, topk, -1).reshape(-1, hidden_size)
)
route_order = (
grouped_assignment_order
if grouped_assignment_order is not None
else torch.argsort(cast(torch.Tensor, route_page_positions))
)
sorted_hidden = assignment_hidden.index_select(0, route_order).contiguous()
sorted_weights = flat_weights.index_select(0, route_order).unsqueeze(-1)
route_counts = (
route_counts_full
if resident_gradient_path or resident_frozen_path
else route_counts_full.index_select(
0,
route_page_keys,
)
* route_page_active.to(dtype=route_counts_full.dtype)
)
route_offsets = route_counts.cumsum(dim=0).to(dtype=torch.int32)
grouped_cuda = (
sorted_hidden.is_cuda
and sorted_hidden.dtype == torch.bfloat16
and not torch.is_grad_enabled()
)
if grouped_cuda:
gate_up = _tensor_native_grouped_linear(
sorted_hidden,
gate_up_pages.contiguous(),
route_offsets,
)
gate, up = gate_up.chunk(2, dim=-1)
middle = F.silu(gate) * up
sorted_values = _tensor_native_grouped_linear(
middle.contiguous(),
down_pages.contiguous(),
route_offsets,
)
else:
sorted_page_positions = (
sorted_route_positions
if grouped_assignment_order is not None
else cast(torch.Tensor, route_page_positions).index_select(
0,
route_order,
)
)
selected_gate_up = gate_up_pages.index_select(0, sorted_page_positions)
gate_up = torch.bmm(
selected_gate_up,
sorted_hidden.unsqueeze(-1),
).squeeze(-1)
gate, up = gate_up.chunk(2, dim=-1)
middle = F.silu(gate) * up
selected_down = down_pages.index_select(0, sorted_page_positions)
sorted_values = torch.bmm(
selected_down,
middle.unsqueeze(-1),
).squeeze(-1)
weighted_values = sorted_values * sorted_weights
# ``route_order`` is a permutation of every assignment row. The old
# ``index_copy`` reconstruction has an ``index_add`` backward, which is
# one of the dominant CUDA stacks for the GPU3 fanout lane. A dense
# permutation scatter has the same exact one-to-one assignment semantics
# but a gather-style backward (no duplicate-index accumulation). Keep the
# route tensor model-owned and preserve assignment order before reduction.
assignment_order_t = route_order.unsqueeze(-1).expand_as(weighted_values)
# ``route_order`` covers every assignment row exactly once, so an
# uninitialized destination is safe and avoids a full-device zero fill
# before the permutation write.
assignment_values = torch.empty_like(weighted_values).scatter(
0,
assignment_order_t,
weighted_values,
)
return assignment_values.reshape(batch, sequence, topk, hidden_size).sum(dim=2)
def _tensor_hotpath_index_boundary(
value: int | torch.Tensor,
*,
reference: torch.Tensor,
upper_bound: int,
) -> torch.Tensor:
"""Normalize one model-owned index without reading a device scalar."""
if (
not isinstance(upper_bound, int)
or isinstance(upper_bound, bool)
or upper_bound < 1
):
raise RuntimeError("historical tensor index upper bound is invalid")
if isinstance(value, torch.Tensor):
if value.numel() != 1:
raise RuntimeError("historical tensor index must contain one element")
index_t = value.detach().to(
device=reference.device,
dtype=torch.long,
).reshape(())
elif isinstance(value, int) and not isinstance(value, bool):
index_t = reference.new_empty((), dtype=torch.long).fill_(value)
else:
raise RuntimeError("historical tensor index boundary is malformed")
return index_t.clamp(min=0, max=upper_bound - 1)
def _tensor_native_cap_weights_from_domain(
self: Any,
domain_idx: int | torch.Tensor,
device: torch.device,
dtype: torch.dtype,
) -> torch.Tensor:
"""Build the exact physical-domain one-hot without a CUDA scalar write."""
capability_embed = self.capability_embed
if not isinstance(capability_embed, torch.Tensor):
raise RuntimeError("historical capability embedding is unavailable")
domain_t = _tensor_hotpath_index_boundary(
domain_idx,
reference=capability_embed,
upper_bound=int(self.n_domains),
).to(device=device)
one_hot = (
torch.arange(self.n_domains, device=device)
.eq(domain_t)
.to(dtype=dtype)
)
logits = self.physical_to_capability(one_hot.unsqueeze(0)).squeeze(0)
return F.softmax(logits, dim=-1)
def _tensor_native_pressure_on_capabilities(
self: Any,
domain_idx: int | torch.Tensor,
outcome_pressure: torch.Tensor | None,
) -> torch.Tensor:
"""Map outcome pressure without indexing through a Python domain scalar."""
device = self.capability_embed.device
dtype = self.capability_embed.dtype
domain_t = _tensor_hotpath_index_boundary(
domain_idx,
reference=self.capability_embed,
upper_bound=int(self.n_domains),
)
cap_weights = cast(
torch.Tensor,
self._cap_weights_from_domain(domain_t, device, dtype),
)
if outcome_pressure is None or outcome_pressure.numel() == 0:
return cap_weights * 0.0
pressure_t = outcome_pressure.to(device=device, dtype=dtype).reshape(-1)
if pressure_t.numel() == self.n_domains:
domain_weakness = pressure_t.index_select(
0,
domain_t.reshape(1),
).reshape(())
else:
domain_weakness = pressure_t.mean()
return cap_weights * domain_weakness.clamp_min(0.0)
def _tensor_native_donor_weights_for_pressure(
self: Any,
domain_idx: int | torch.Tensor,
outcome_pressure: torch.Tensor | None,
*,
hidden: torch.Tensor | None = None,
) -> torch.Tensor:
"""Build the exact learned donor mix without CUDA scalar indexing."""
weak = self._pressure_on_capabilities(domain_idx, outcome_pressure)
weak_index = weak.argmax().to(dtype=torch.long).reshape(1)
visit_scores, _ = self.neural_dag(
hidden=hidden,
start_cap=weak_index.reshape(()),
)
related_transfer = self.related_transfer.index_select(0, weak_index).squeeze(0)
relatedness_prior = self.relatedness_prior.index_select(0, weak_index).squeeze(0)
related = F.softmax(
related_transfer + relatedness_prior * 0.5,
dim=-1,
)
combined = (
0.7 * visit_scores.to(device=related.device, dtype=related.dtype)
+ 0.3 * related
)
improve = torch.sigmoid(
self.improvement_coupling.index_select(0, weak_index).squeeze(0)
)
donor = combined * (1.0 + improve)
active = (weak.max().detach() >= 1e-6).to(
device=donor.device,
dtype=donor.dtype,
)
return cast(torch.Tensor, active * donor + (1.0 - active) * weak)
def _historical_transfer_init_with_tensor_constants(
self: Any,
*args: Any,
**kwargs: Any,
) -> None:
"""Construct immutable NoNE geometry once, before the module moves to CUDA."""
original = getattr(type(self), "_resynthesis_original_transfer_init", None)
if not callable(original):
raise RuntimeError("historical transfer constructor was not preserved")
original(self, *args, **kwargs)
capability_embed = getattr(self, "capability_embed", None)
n_capabilities = getattr(self, "n_capabilities", None)
if (
not isinstance(capability_embed, torch.Tensor)
or not isinstance(n_capabilities, int)
or isinstance(n_capabilities, bool)
or n_capabilities < 1
):
raise RuntimeError("historical NoNE capability geometry is malformed")
self.register_buffer(
"_resynthesis_none_source_identity_t",
torch.eye(
n_capabilities,
device=capability_embed.device,
dtype=capability_embed.dtype,
),
persistent=False,
)
self.register_buffer(
"_resynthesis_none_phase_axis_t",
capability_embed.new_tensor((-1.0, -0.25, 0.25, 1.0)),
persistent=False,
)
def _tensor_native_build_none_pathway_packet(
self: Any,
domain_idx: int | torch.Tensor,
*,
donor: torch.Tensor,
hidden: torch.Tensor | None,
outcome_pressure: torch.Tensor | None,
n_layers: int,
n_experts: int,
) -> Any:
"""Build the exact historical NoNE packet without per-hop host tensors."""
device = self.capability_embed.device
dtype = self.capability_embed.dtype
source_identity_t = self._resynthesis_none_source_identity_t
phase_axis_t = self._resynthesis_none_phase_axis_t
if (
not isinstance(source_identity_t, torch.Tensor)
or source_identity_t.shape
!= (self.n_capabilities, self.n_capabilities)
or source_identity_t.device != device
or source_identity_t.dtype != dtype
or not isinstance(phase_axis_t, torch.Tensor)
or phase_axis_t.shape != (4,)
or phase_axis_t.device != device
or phase_axis_t.dtype != dtype
):
raise RuntimeError("historical NoNE tensor constants differ")
donor = donor.to(device=device, dtype=dtype).reshape(self.n_capabilities)
donor_mass = donor.sum()
domain_t = _tensor_hotpath_index_boundary(
domain_idx,
reference=source_identity_t,
upper_bound=int(self.n_capabilities),
)
source_default = source_identity_t.index_select(
0,
domain_t.reshape(1),
).squeeze(0)
epsilon = torch.finfo(dtype).eps
has_donor = donor_mass.abs().gt(epsilon).to(dtype=dtype)
source_mix = (
has_donor * donor / donor_mass.abs().clamp_min(epsilon)
+ (1.0 - has_donor) * source_default
)
if hidden is None:
pooled = torch.zeros(self.hidden_size, device=device, dtype=dtype)
else:
hidden_t = hidden.to(device=device, dtype=dtype)
pooled = hidden_t.reshape(-1, hidden_t.shape[-1]).mean(dim=0)
hidden_state = torch.tanh(
self.none_hidden_to_pathway(pooled.unsqueeze(0)).squeeze(0)
)
source_state = source_mix @ self.capability_embed
pathway_state = torch.tanh(hidden_state + source_state)
phase = F.softmax(
self.none_phase_head(pathway_state.unsqueeze(0)).squeeze(0),
dim=-1,
)
transfer_logits = (
self.none_transfer_head(pathway_state.unsqueeze(0)).squeeze(0)
+ source_mix @ self.related_transfer
)
transfer = F.softmax(transfer_logits, dim=-1)
transfer_state = transfer @ self.capability_embed
pathway_state = torch.tanh(pathway_state + transfer_state)
if outcome_pressure is None or outcome_pressure.numel() == 0:
pressure = pathway_state.new_zeros(())
else:
pressure = (
outcome_pressure.to(device=device, dtype=dtype)
.reshape(-1)
.mean()
.clamp_min(0.0)
)
gap = torch.sigmoid(
self.none_gap_head(pathway_state.unsqueeze(0)).squeeze(0).squeeze(-1)
+ pressure
)
completion = torch.sigmoid(
self.none_completion_head(pathway_state.unsqueeze(0))
.squeeze(0)
.squeeze(-1)
- pressure
)
gain = torch.tanh(self.none_runtime_gain)
active_work = gap + (1.0 - completion)
domain_raw = (
self.none_domain_head(pathway_state.unsqueeze(0)).squeeze(0)
* active_work
)
layer_full = (
self.none_layer_head(pathway_state.unsqueeze(0)).squeeze(0)
* active_work
)
expert_full = (
self.none_expert_head(pathway_state.unsqueeze(0)).squeeze(0)
* active_work
)
layer_raw = layer_full.repeat(
(n_layers + layer_full.numel() - 1) // layer_full.numel()
)[:n_layers]
expert_raw = expert_full.repeat(
(n_experts + expert_full.numel() - 1) // expert_full.numel()
)[:n_experts]
phase_drive = torch.sum(phase * phase_axis_t)
stop_drive = completion - gap + phase_drive
slice_drive = gap + phase[1] + phase[2] - completion
stop_raw = (
torch.tanh(
self.none_stop_head(pathway_state.unsqueeze(0)).squeeze()
)
* stop_drive
)
slice_raw = (
torch.tanh(
self.none_slice_head(pathway_state.unsqueeze(0)).squeeze()
)
* slice_drive
)
packet_type = self._resynthesis_none_pathway_packet_type
return packet_type(
phase=phase,
transfer=transfer,
gap=gap,
completion=completion,
domain_delta=self._neutral_trainable_residual(domain_raw, gain),
layer_delta=self._neutral_trainable_residual(layer_raw, gain),
expert_delta=self._neutral_trainable_residual(expert_raw, gain),
stop_delta=self._neutral_trainable_residual(stop_raw, gain),
slice_delta=self._neutral_trainable_residual(slice_raw, gain),
runtime_gain=gain,
)
def _tensor_native_compute_hop_signal(
self: Any,
domain_idx: int | torch.Tensor,
subdomain_idx: int | torch.Tensor,
*,
hidden: torch.Tensor | None = None,
intent_probs: torch.Tensor | None = None,
outcome_pressure: torch.Tensor | None = None,
hop_idx: int = 0,
step_idx: int = 0,
n_layers: int | None = None,
n_experts: int | None = None,
prepared_context: _HistoricalHopContextPacket | None = None,
) -> Any:
"""Select the exact learned hop pair with tensor indexing, never ``item``."""
layer_count = int(n_layers if n_layers is not None else self.n_layers)
expert_count = int(n_experts if n_experts is not None else self.n_experts)
domain_t = _tensor_hotpath_index_boundary(
domain_idx,
reference=self.transition_logits,
upper_bound=int(self.n_domains),
)
subdomain_t = _tensor_hotpath_index_boundary(
subdomain_idx,
reference=self.subdomain_coupling,
upper_bound=int(self.n_subdomains),
)
hop_slot = int(hop_idx) % int(self.hop_slots)
from_index_t = domain_t.clamp(
max=int(self.hop_domain_dim) - 1,
).reshape(1)
hop_route = self.cross_domain_hop.index_select(
0,
from_index_t,
).squeeze(0).clone()
destination = hop_route.argmax(dim=-1).reshape(1).to(dtype=torch.long)
from_embedding = self.hop_embed_from.index_select(
0,
from_index_t,
).squeeze(0)
to_embedding = self.hop_embed_to.index_select(0, destination).squeeze(0)
depth = int(step_idx + hop_idx) % int(self.hop_max_depth)
from_hidden = F.silu(self.hop_from_proj(from_embedding.unsqueeze(0))).squeeze(0)
to_hidden = F.silu(self.hop_to_proj(to_embedding.unsqueeze(0))).squeeze(0)
mixed = F.silu(
self.hop_mix(torch.cat((from_hidden, to_hidden), dim=-1).unsqueeze(0))
).squeeze(0)
mixed = mixed + self.hop_depth_embed[depth]
bonus = self.hop_bonus_head(mixed.unsqueeze(0)).squeeze(0).squeeze(-1)
confidence = torch.sigmoid(
self.hop_confidence_head(mixed.unsqueeze(0)).squeeze(0).squeeze(-1)
)
hop_route = hop_route + confidence * bonus.expand_as(hop_route) * 0.1
destination = hop_route.argmax(dim=-1).reshape(1).to(dtype=torch.long)
to_embedding = self.hop_embed_to.index_select(0, destination).squeeze(0)
context = from_embedding @ self.hop_context_core + to_embedding
pair_layers = (
self.cross_domain_hop_layer.index_select(0, from_index_t)
.squeeze(0)
.index_select(0, destination)
)
pair_experts = (
self.cross_domain_hop_expert.index_select(0, from_index_t)
.squeeze(0)
.index_select(0, destination)
)
layer_full = pair_layers.squeeze(0) + context @ self.hop_context_to_layer
expert_full = pair_experts.squeeze(0) + context @ self.hop_context_to_expert
layer_bias = layer_full.repeat(
(layer_count + layer_full.numel() - 1) // layer_full.numel()
)[:layer_count]
expert_bias = expert_full.repeat(
(expert_count + expert_full.numel() - 1) // expert_full.numel()
)[:expert_count]
stop_delta = (
self.cross_domain_hop_stop.index_select(0, from_index_t)
.squeeze(0)
.index_select(0, destination)
.squeeze(0)
)
slice_delta = (
self.cross_domain_hop_slice.index_select(0, from_index_t)
.squeeze(0)
.index_select(0, destination)
.squeeze(0)
)
transition_row = self.transition_logits.index_select(
0,
domain_t.reshape(1),
).squeeze(0)
subdomain_row = (
self.subdomain_coupling.index_select(0, domain_t.reshape(1))
.squeeze(0)
.index_select(0, subdomain_t.reshape(1))
.squeeze(0)
)
base = (
transition_row
+ subdomain_row
+ self.hop_depth_bias[hop_slot]
+ self.hop_to_domain(hop_route.unsqueeze(0)).squeeze(0)
)
if hidden is not None and intent_probs is not None:
context = prepared_context
if context is None:
pooled = _parent_hidden_context(hidden).to(
dtype=self.context_query.weight.dtype
)
intent_mean = intent_probs.mean(dim=1).to(dtype=pooled.dtype)
fused = torch.cat((pooled, intent_mean), dim=-1)
if fused.dim() == 1:
fused = fused.unsqueeze(0)
context = _HistoricalHopContextPacket(
query_t=self.context_query(fused).squeeze(0),
gate_t=torch.sigmoid(self.context_gate(fused)).squeeze(0),
)
base = base + context.gate_t * context.query_t
if outcome_pressure is not None:
pressure = outcome_pressure.to(dtype=base.dtype, device=base.device)
if pressure.numel() == self.n_domains:
base = base + self.pressure_scale * pressure
base, layer_bias, expert_bias, none_pathway = (
self.knowledge_transfer.apply_transfer(
domain_t,
domain_logits=base,
layer_bias=layer_bias,
expert_bias=expert_bias,
outcome_pressure=outcome_pressure,
n_layers=layer_count,
n_experts=expert_count,
hidden=hidden,
)
)
stop_delta = stop_delta + none_pathway.stop_delta.to(stop_delta)
slice_delta = slice_delta + none_pathway.slice_delta.to(slice_delta)
signal_type = self._resynthesis_hop_signal_type
return signal_type(
domain_logits=base,
layer_bias=layer_bias,
expert_bias=expert_bias,
stop_delta=stop_delta,
slice_delta=slice_delta,
hop_bonus=bonus,
hop_confidence=confidence,
none_pathway=none_pathway,
from_hop_idx=from_index_t.reshape(()),
dest_hop_idx=destination.reshape(()),
)
def _historical_plan_layer_traversal_with_arm_exhaustion(
self: Any,
step_idx: int,
domain_idx: int,
subdomain_idx: int,
*,
last_rubric: float | None,
last_layer: int | None,
device: torch.device,
hidden: torch.Tensor | None = None,
intent_probs: torch.Tensor | None = None,
extra_pressure: torch.Tensor | None = None,
transfer_weights: torch.Tensor | None = None,
stage_layer_bias: torch.Tensor | None = None,
) -> Any:
"""Traverse the historical learned layer frontier exactly once per arm.
The July 13 planner documented layer-arm exhaustion as a terminal model
authority, but its low-rubric branch repeatedly selected ``last_layer``
without consuming another arm. Its local ``visited_layers`` set therefore
remained at one element and the function itself could never return for a
low-confidence prompt. A wrapper around its return value was necessarily
too late to repair that state machine.
This compatibility owner retains all eight trained routing signals from
the historical planner and masks only layer arms already consumed in the
current graph frontier. The topology-derived frontier has exactly one arm
per attached expert layer; exhausting those model-owned arms is not a host
hop, token, step, or time cap. The final list conversion is the explicit
legacy boundary required by the immutable parent's recursive-pass API.
"""
original = getattr(type(self), "_resynthesis_original_plan_layer_traversal", None)
if not callable(original):
raise RuntimeError("historical Resynthesis traversal owner was not preserved")
exhaustion = getattr(self, "_resynthesis_recursive_arm_exhausted", None)
layer_pick_logits = getattr(self, "layer_pick_logits", None)
experts = getattr(self, "experts", None)
if not isinstance(exhaustion, torch.Tensor) or not isinstance(
layer_pick_logits, torch.Tensor
):
raise RuntimeError("historical Resynthesis arm-exhaustion tensor is unavailable")
if experts is None:
raise RuntimeError("historical Resynthesis traversal expert frontier is unavailable")
layer_count = len(experts)
if layer_count <= 0:
exhaustion.zero_()
return [], []
rc = self.rbo_cfg
destination_pressure = self.arm_registry.domain_slot_pressure().sum(dim=-1)
if extra_pressure is not None:
pressure = extra_pressure.to(
device=destination_pressure.device,
dtype=destination_pressure.dtype,
)
if pressure.numel() == destination_pressure.numel():
destination_pressure = destination_pressure + pressure
# Keep the legacy route triples and the learned terminal decision in one
# contiguous packet. The immutable parent requires Python route tuples at
# its recursive-pass boundary, so transferring this packet once avoids a
# second CUDA synchronization for the stop reason without changing either
# model-owned decision.
legacy_boundary_t = torch.empty(
(layer_count * 3 + 1,),
device=device,
dtype=torch.long,
)
routes = legacy_boundary_t[:-1].view(layer_count, 3)
visited = torch.zeros(layer_count, device=device, dtype=torch.bool)
frontier_scores = layer_pick_logits.new_full(
(layer_count,),
-torch.inf,
device=device,
)
hop_signals: list[Any] = []
active_domain_t = _tensor_hotpath_index_boundary(
domain_idx,
reference=layer_pick_logits,
upper_bound=int(self.arm_layout.n_domains),
).to(device=device)
active_subdomain_t = _tensor_hotpath_index_boundary(
subdomain_idx,
reference=layer_pick_logits,
upper_bound=int(self.arm_layout.n_subdomains),
).to(device=device)
struggling = last_rubric is not None and last_rubric < rc.anti_floor
final_stop_delta = layer_pick_logits.new_zeros(())
final_slice_delta = layer_pick_logits.new_zeros(())
# The content, intent, transfer packet, expert frontier, and learned layer
# identities are immutable for this planner invocation. The historical
# implementation rebuilt the same profile/identity query once for every
# arm in the frontier (21 times in the native parent). Reuse that exact
# model-owned tensor while still evaluating every hop, domain pressure,
# arm posterior, and layer score independently below. This changes no
# route, parameter, expert participation, or stopping authority.
base = torch.sigmoid(layer_pick_logits[:layer_count].to(device=device))
base_score = torch.log(base.clamp(min=1e-4))
identity_layer = self._layer_identity_bias(
hidden,
intent_probs,
transfer_weights,
layer_count,
).to(device=device, dtype=base.dtype)
stage_route_scale = (
2.0
* torch.sigmoid(self.swe_stage_route_logit).to(
device=device,
dtype=base.dtype,
)
if stage_layer_bias is not None
else None
)
stage_layer_score = (
stage_route_scale
* stage_layer_bias[:layer_count].to(
device=device,
dtype=base.dtype,
)
if stage_layer_bias is not None and stage_route_scale is not None
else None
)
layer_slots_t = torch.arange(
layer_count,
device=device,
dtype=torch.long,
)
# Subdomain slot geometry is fixed for the complete learned frontier.
# Only ``active_domain_t`` changes between hops. Rebuilding this identical
# axis inside every hop added one allocation/dispatch (and one CUDA kernel
# launch on the live parent) per traversed layer without influencing the
# anti-Thompson posterior, pressure, route, or stop decision.
subdomain_slots_t = (
torch.arange(
self.arm_layout.n_subdomains,
device=device,
dtype=torch.long,
)
if layer_count > 1 and last_rubric is not None and not struggling
else None
)
domain_layer_affinity = self.domain_layer_affinity
if (
not isinstance(domain_layer_affinity, torch.Tensor)
or domain_layer_affinity.dim() != 2
or domain_layer_affinity.shape[0] < self.arm_layout.n_domains
or domain_layer_affinity.shape[1] < layer_count
):
raise RuntimeError("historical Resynthesis domain-layer affinity differs")
# Hidden content and task intent do not change while the planner consumes
# its model-owned layer frontier. The historical hop owner previously
# repeated the same full-context attention pool and the same two learned
# projections once per arm. Prepare that tensor packet once under the
# historical planner's unchanged no-gradient authority and reuse it; each
# arm still computes its own hop destination,
# domain pressure, transfer pathway, posterior, layer score, and stop
# evidence below.
prepared_hop_context: _HistoricalHopContextPacket | None = None
if hidden is not None and intent_probs is not None:
with torch.no_grad():
pooled = _parent_hidden_context(hidden).to(
dtype=self.cross_domain_router.context_query.weight.dtype
)
intent_mean = intent_probs.mean(dim=1).to(dtype=pooled.dtype)
fused = torch.cat((pooled, intent_mean), dim=-1)
if fused.dim() == 1:
fused = fused.unsqueeze(0)
prepared_hop_context = _HistoricalHopContextPacket(
query_t=self.cross_domain_router.context_query(fused).squeeze(0),
gate_t=torch.sigmoid(
self.cross_domain_router.context_gate(fused)
).squeeze(0),
)
for hop_idx in range(layer_count):
with torch.no_grad():
hop_signal = self.cross_domain_router.compute_hop_signal(
active_domain_t,
active_subdomain_t,
hidden=hidden,
intent_probs=intent_probs,
outcome_pressure=destination_pressure,
hop_idx=hop_idx,
step_idx=step_idx,
n_layers=layer_count,
n_experts=self.cfg.total_experts,
prepared_context=prepared_hop_context,
)
hop_signals.append(hop_signal)
self._last_hop_signal = hop_signal
final_stop_delta = hop_signal.stop_delta.detach().reshape(())
final_slice_delta = hop_signal.slice_delta.detach().reshape(())
self._hop_stop_delta_accum = (
self._hop_stop_delta_accum.to(
device=final_stop_delta.device,
dtype=final_stop_delta.dtype,
)
+ final_stop_delta
)
self._hop_slice_delta_accum = (
self._hop_slice_delta_accum.to(
device=final_slice_delta.device,
dtype=final_slice_delta.dtype,
)
+ final_slice_delta
)
if hop_idx > 0 and last_rubric is not None and not struggling:
if subdomain_slots_t is None:
raise RuntimeError(
"historical Resynthesis subdomain slot axis is unavailable"
)
active_domain_t = _tensor_hotpath_index_boundary(
hop_signal.domain_logits.detach().argmax(),
reference=layer_pick_logits,
upper_bound=int(self.arm_layout.n_domains),
)
subdomain_arm_ids_t = (
active_domain_t * self.arm_layout.n_subdomains
+ subdomain_slots_t
)
subdomain_posterior_t = self.arm_registry.posterior_mean(
subdomain_arm_ids_t
)
subdomain_memory_t = (
0.5 - self.arm_registry.success_ema[subdomain_arm_ids_t]
).clamp_min(0.0)
subdomain_pressure_t = (
self.arm_registry.anti_fails[subdomain_arm_ids_t]
+ subdomain_memory_t
+ (0.5 - subdomain_posterior_t).clamp_min(0.0)
)
best_subdomain_t = subdomain_pressure_t.argmax().reshape(())
if self.arm_layout.n_subdomains > 1:
best_subdomain_t = torch.where(
best_subdomain_t.eq(active_subdomain_t),
(active_subdomain_t + 1).remainder(
self.arm_layout.n_subdomains
),
best_subdomain_t,
)
active_subdomain_t = best_subdomain_t
domain_subdomain_t = (
active_domain_t * self.arm_layout.n_subdomains
+ active_subdomain_t
)
layer_arms = (
self.arm_layout.n_domain_subdomain
+ (
(
domain_subdomain_t * self.arm_layout.n_layers
+ layer_slots_t
)
* self.arm_layout.step_slots
+ (int(step_idx) % self.arm_layout.step_slots)
)
* self.arm_layout.hop_slots
+ (hop_idx % self.arm_layout.hop_slots)
)
anti = self.arm_registry.suppression(layer_arms)
explore = self.arm_registry.explore_boost(
layer_arms,
weight=rc.thompson_explore_weight,
sample=not self.training,
)
memory = self.arm_registry.memory_log_bias(layer_arms) * 0.15
domain_bias = torch.sigmoid(
domain_layer_affinity.index_select(
0,
active_domain_t.reshape(1),
).squeeze(0)[:layer_count]
)
hop_layer = hop_signal.layer_bias[:layer_count].to(
device=device,
dtype=base.dtype,
)
scores = (
base_score
+ torch.log(domain_bias.to(device=device).clamp(min=1e-4))
+ 0.35 * torch.log(anti.to(device=device).clamp(min=1e-4))
+ 0.25 * torch.log(explore.to(device=device).clamp(min=1e-4))
+ memory.to(device=device)
+ 0.3 * hop_layer
+ rc.identity_layer_strength * identity_layer
)
if stage_layer_bias is not None:
if stage_layer_score is None:
raise RuntimeError("historical Resynthesis stage-route scale is unavailable")
scores = scores + stage_layer_score.to(
device=scores.device,
dtype=scores.dtype,
)
selected_layer = scores.masked_fill(visited, -torch.inf).argmax().reshape(())
selected_score = scores.gather(0, selected_layer.reshape(1)).reshape(())
frontier_scores.scatter_(
0,
selected_layer.reshape(1),
selected_score.reshape(1),
)
routes[hop_idx] = torch.stack(
(
selected_layer,
active_domain_t,
active_subdomain_t,
)
)
visited.scatter_(0, selected_layer.reshape(1), True)
exhausted = visited.all()
exhaustion.copy_(exhausted.to(device=exhaustion.device, dtype=torch.bool))
torch._assert_async(
torch.isfinite(frontier_scores).all(),
"historical Resynthesis frontier blend has an unscored layer",
)
# Every model-selected layer arm executes, but its residual is blended by
# the same learned route evidence that selected its place in the frontier.
# Normalizing the complete frontier prevents 21 independently trained
# expert residuals from being treated as 21 full-strength sequential model
# replacements. This is model-owned MoE aggregation, not a host cap: every
# finite arm retains non-zero contribution and the weights sum to one.
# The historical checkpoint runs this control surface in bfloat16. A
# bfloat16 softmax followed by a bfloat16 renormalization can leave the
# represented 21-arm mass roughly 1e-3 away from one after promotion back
# to float32. Keep the learned scores and all selected arms unchanged,
# but perform the aggregation in float32 so the model-owned mixture has a
# numerically faithful unit-mass contract before each scalar is cast at the
# legacy expert-residual boundary.
frontier_blend = torch.softmax(
frontier_scores.to(dtype=torch.float32),
dim=0,
)
# A finite learned score can still underflow to exactly zero after
# exponentiation. Arm exhaustion promises that every finite model-owned
# layer contributes, so retain the dtype's smallest representable positive
# mass and renormalize. This does not select, skip, or cap an arm.
frontier_blend = frontier_blend.clamp_min(torch.finfo(torch.float32).tiny)
frontier_blend = frontier_blend / frontier_blend.sum()
setattr(
self,
"_resynthesis_recursive_frontier_blend",
frontier_blend.detach(),
)
learned_cycle_stop = torch.sigmoid(final_stop_delta).ge(0.5) | final_slice_delta.gt(
0.5
)
legacy_boundary_t[-1].copy_(learned_cycle_stop)
legacy_boundary = legacy_boundary_t.detach().to(device="cpu").tolist()
self._last_traversal_stop_reason = (
"trained_gate" if bool(legacy_boundary[-1]) else "arm_exhaustion"
)
plan = [
(legacy_boundary[offset], legacy_boundary[offset + 1], legacy_boundary[offset + 2])
for offset in range(0, layer_count * 3, 3)
]
return plan, hop_signals
def _historical_layer_gate_scale_with_frontier_blend(
self: Any,
layer_idx: int,
) -> torch.Tensor:
"""Blend every traversed layer by its trained frontier route evidence."""
original = getattr(type(self), "_resynthesis_original_layer_gate_scale", None)
if not callable(original):
raise RuntimeError("historical Resynthesis layer gate owner was not preserved")
base_scale = original(self, layer_idx)
frontier_blend = getattr(self, "_resynthesis_recursive_frontier_blend", None)
experts = getattr(self, "experts", None)
if not isinstance(base_scale, torch.Tensor) or not isinstance(
frontier_blend,
torch.Tensor,
):
raise RuntimeError("historical Resynthesis frontier blend tensor is unavailable")
if experts is None or frontier_blend.shape != (len(experts),):
raise RuntimeError("historical Resynthesis frontier blend geometry differs")
blend = frontier_blend[layer_idx].to(
device=base_scale.device,
dtype=base_scale.dtype,
)
return base_scale * blend
def _historical_joint_stop_logit_with_arm_exhaustion(
self: Any,
signals: Any,
) -> torch.Tensor:
"""Fuse graph-frontier exhaustion into the parent's existing stop logit.
A complete frontier is an independent native stop authority in the parent
contract. Raising the existing logit above its unchanged 0.5 decision
boundary lets the historical outer loop consume that authority. Its
original low-rubric-and-improving exception remains intact, so measurable
correction progress may still continue without any host-derived cap.
"""
original = getattr(type(self), "_resynthesis_original_joint_stop_logit", None)
if not callable(original):
raise RuntimeError("historical Resynthesis stop-logit owner was not preserved")
joint_logit = original(self, signals)
exhaustion = getattr(self, "_resynthesis_recursive_arm_exhausted", None)
if not isinstance(joint_logit, torch.Tensor) or not isinstance(
exhaustion, torch.Tensor
):
raise RuntimeError("historical Resynthesis arm-exhaustion stop tensor is unavailable")
exhausted_stop_logit = torch.maximum(
joint_logit,
torch.ones_like(joint_logit),
)
return torch.where(
exhaustion.to(device=joint_logit.device, dtype=torch.bool),
exhausted_stop_logit,
joint_logit,
)
def _historical_batch_safe_correction_training_loss(self: Any) -> torch.Tensor:
"""Align each historical policy target with its row-local policy logits.
The inherited correction controller intentionally derives one aggregate
mode target per recursive step. Its loss path concatenated row-local logits
as ``[steps * batch, modes]`` but retained only ``steps`` scalar targets.
Expand each aggregate step target across that step's rows so every row
contributes gradient without mixing or selecting a representative row.
"""
original = getattr(
type(self),
"_resynthesis_original_correction_training_loss",
None,
)
if not callable(original):
raise RuntimeError("historical Resynthesis correction-loss owner was not preserved")
mode_logits = getattr(self, "_last_policy_mode_logits", None)
mode_targets = getattr(self, "_last_policy_mode_targets", None)
if not isinstance(mode_logits, list) or not isinstance(mode_targets, list):
raise RuntimeError("historical Resynthesis correction-policy history differs")
if not mode_logits and not mode_targets:
return cast(torch.Tensor, original(self))
if len(mode_logits) != len(mode_targets):
raise RuntimeError("historical correction-policy history lengths differ")
expanded_targets: list[torch.Tensor] = []
expansion_required = False
for logits_t, target_t in zip(mode_logits, mode_targets, strict=True):
if not isinstance(logits_t, torch.Tensor) or not isinstance(
target_t,
torch.Tensor,
):
raise RuntimeError("historical correction-policy history is not tensor-owned")
if logits_t.dim() != 2 or logits_t.shape[0] < 1:
raise RuntimeError("historical correction-policy logits geometry differs")
flat_target = target_t.reshape(-1)
if flat_target.numel() == 1:
row_targets = flat_target.expand(logits_t.shape[0])
expansion_required = expansion_required or logits_t.shape[0] > 1
elif flat_target.numel() == logits_t.shape[0]:
row_targets = flat_target
else:
raise RuntimeError("historical correction-policy target geometry differs")
expanded_targets.extend(row_targets.unbind(0))
if not expansion_required:
return cast(torch.Tensor, original(self))
setattr(self, "_last_policy_mode_targets", expanded_targets)
try:
return cast(torch.Tensor, original(self))
finally:
setattr(self, "_last_policy_mode_targets", mode_targets)
def _tensor_native_self_correction_mode_target(
signals: Mapping[str, torch.Tensor],
*,
verification_deficit: torch.Tensor,
domain_collapse: torch.Tensor,
rubric_floor: float,
) -> torch.Tensor:
"""Preserve the historical mode target without a per-hop H2D scalar copy."""
rubric = signals["rubric"]
# The source-era implementation used ``rubric.new_tensor(Python_float)``.
# On CUDA that enters the host-to-device copy path at every recursive hop.
# Build the same scalar from a device-native factory and pass the unchanged
# configuration value as a scalar kernel argument instead.
floor = rubric.new_empty(()).fill_(rubric_floor)
zero = rubric.new_zeros(())
route_pressure = torch.maximum(
domain_collapse.reshape(()),
signals.get("history_pressure", zero).reshape(()),
)
hidden_pressure = torch.maximum(
(floor - rubric).clamp(min=0.0).reshape(()),
signals.get("stalled", zero).reshape(()),
)
verify_pressure = verification_deficit.reshape(())
transfer_pressure = signals.get("tool_pressure", zero).reshape(())
execution_pressure = signals.get("execution_pressure", zero).reshape(())
grounding_confidence = signals.get("grounding_confidence", zero).reshape(())
growth_pressure = torch.maximum(
signals.get("history_pressure", zero).reshape(()),
(signals["contradiction"] - (torch.ones_like(floor) - floor))
.clamp(min=0.0)
.reshape(()),
)
verify_pressure = torch.maximum(verify_pressure, execution_pressure)
transfer_pressure = torch.maximum(
transfer_pressure,
execution_pressure * grounding_confidence,
)
growth_pressure = torch.maximum(growth_pressure, execution_pressure)
return torch.stack(
(
zero,
route_pressure,
hidden_pressure,
verify_pressure,
transfer_pressure,
growth_pressure,
)
).argmax(dim=0)
def _tensor_native_domain_collapse_forward(
self: Any,
domain_probs: torch.Tensor,
) -> tuple[torch.Tensor, torch.Tensor]:
"""Preserve collapse correction without a per-forward host scalar copy."""
entropy = -(
domain_probs * (domain_probs + 1e-8).log()
).sum(dim=-1)
maximum_entropy = (
domain_probs.new_ones(())
.mul_(self.n_domains)
.log_()
)
normalized_entropy = entropy / maximum_entropy
threshold = torch.sigmoid(self.entropy_threshold)
collapse_score = torch.sigmoid(
(threshold - normalized_entropy) * 8.0
)
strength = torch.sigmoid(self.redistribution_strength)
uniform = torch.ones_like(domain_probs) / self.n_domains
blend = strength * collapse_score.unsqueeze(-1)
adjusted = (1 - blend) * domain_probs + blend * uniform
return adjusted, collapse_score.mean()
def _tensor_native_rubric_apply_bias(
self: Any,
logits: torch.Tensor,
hidden: torch.Tensor,
intent_probs: torch.Tensor,
strength: float = 1.0,
) -> torch.Tensor:
"""Scatter historical rubric activations in the logits' exact dtype.
The frozen parent may expose BF16 hidden/rubric activations while its
logits surface is FP32. ``scatter_add_`` requires destination and source
dtypes to match, so the source-era method failed before the first r152
wave. The cast changes only representation at this additive logit
boundary; token IDs, learned rubric weights, and routing remain exact.
"""
_, weighted = self.forward(hidden, intent_probs)
if not isinstance(weighted, torch.Tensor) or weighted.ndim != 4:
raise RuntimeError("historical rubric activation geometry differs")
if logits.ndim != 3 or weighted.shape[:2] != logits.shape[:2]:
raise RuntimeError("historical rubric logits geometry differs")
bias = torch.zeros_like(logits)
active_mask = cast(torch.Tensor, self.rubric_mask).to(
device=weighted.device,
dtype=torch.bool,
)
rubric_token_ids = cast(torch.Tensor, self.rubric_token_ids).to(
device=weighted.device,
dtype=torch.long,
)
# The rubric registry has a fixed ``[intent, term]`` geometry. Boolean
# indexing that registry asks CUDA ``nonzero`` to size a dynamic result,
# synchronizing the parent stream during every frozen prefill. Scatter the
# same fixed slots and make unregistered contributions exactly zero
# instead: inactive token IDs are zero-initialized by the historical
# checkpoint, and zero-valued duplicate scatters cannot alter token zero.
# ``where`` also prevents an inactive non-finite projection from leaking
# through multiplication by zero.
flat_token_ids = rubric_token_ids.reshape(-1)
weighted_flat = weighted.flatten(start_dim=2)
weighted_active = torch.where(
active_mask.reshape(1, 1, -1),
weighted_flat * strength,
torch.zeros_like(weighted_flat),
).to(
device=bias.device,
dtype=bias.dtype,
)
index_t = (
flat_token_ids.to(device=bias.device, dtype=torch.long)
.view(1, 1, -1)
.expand(logits.shape[0], logits.shape[1], -1)
)
bias.scatter_add_(2, index_t, weighted_active)
return logits + bias
def _tensor_native_historical_domain_route(
adjusted_domain_probabilities: torch.Tensor,
subdomain_indices: torch.Tensor,
) -> tuple[torch.Tensor, torch.Tensor]:
"""Keep the historical model-selected route on its active device.
The source-era parent selected the same two argmax values and immediately
called ``Tensor.item`` on each. That serialized the complete CUDA stream
before the recursive graph could enqueue any work. The downstream
Resynthesis compatibility owners already accept scalar tensor indices, so
retain the two exact model decisions as tensors through that graph.
"""
if (
adjusted_domain_probabilities.dim() != 2
or adjusted_domain_probabilities.shape[0] < 1
or adjusted_domain_probabilities.shape[1] < 1
or subdomain_indices.dim() != 1
or subdomain_indices.shape[0] < 1
):
raise RuntimeError("historical Resynthesis domain route geometry differs")
return (
# Only the first route row is consumed by the historical recursive
# owner below. Reducing every batch row and then discarding all but
# row zero needlessly launches a full-batch CUDA reduction.
adjusted_domain_probabilities[0].argmax()
.to(
device=adjusted_domain_probabilities.device,
dtype=torch.long,
)
.reshape(()),
subdomain_indices[0]
.to(
device=adjusted_domain_probabilities.device,
dtype=torch.long,
)
.reshape(()),
)
def _tensor_preserving_historical_int(
value: Any,
*args: Any,
) -> int | torch.Tensor:
"""Preserve scalar tensors only inside the cloned historical recursion.
Python's source-era ``_recursive_pass`` begins with ``int(domain_idx)`` and
``int(subdomain_idx)``. Its installed traversal owner is tensor-native,
while every genuine Python integer in that function must retain normal
``int`` behavior. A private globals table for that one function lets the
two scalar tensors survive without changing process-wide builtins or the
immutable parent module.
"""
if isinstance(value, torch.Tensor):
if args or value.numel() != 1:
raise RuntimeError("historical recursive tensor index is malformed")
return value.reshape(())
return int(value, *args)
def _tensor_preserving_historical_float(
value: Any = 0.0,
) -> float | torch.Tensor:
"""Keep frozen-parent diagnostic scalars on their current device.
The source-era recursive pass converts every diagnostic signal to
``float`` before appending it to a Python history list. On CUDA each
conversion executes ``Tensor.item()``, synchronizing the complete stream.
The frozen parent does not train from those scalar histories; its active
output is the routed hidden/logit tensor and its compact proof deliberately
omits the boundary-only histories. Preserve scalar tensors in that exact
path while retaining normal Python ``float`` behavior for configuration
constants and for the trainable historical path.
"""
if isinstance(value, torch.Tensor):
if value.numel() != 1:
raise RuntimeError("historical recursive float tensor is malformed")
return value.reshape(())
return float(value)
class _HistoricalRecursiveTorchFacade:
"""Keep the frozen recursive pass from staging one scalar per layer.
The verified source-era function has one ``torch.tensor`` call. It wraps
the already-known Python layer-arm identifier in a one-element CUDA tensor
after every model-selected expert. Constructing that tensor through the
generic host-data factory enters the synchronous H2D copy path. The arm
identifier is a kernel scalar, so an empty device tensor followed by
``fill_`` has identical values without a host buffer or stream fence.
All other torch attributes are delegated unchanged. This facade is
installed only in the private globals table of the cloned historical
recursive function; process-wide torch behavior is never modified.
"""
def __getattr__(self, name: str) -> Any:
return getattr(torch, name)
@staticmethod
def tensor(
data: object,
*,
device: torch.device,
dtype: torch.dtype,
) -> torch.Tensor:
if (
not isinstance(data, (list, tuple))
or len(data) != 1
or not isinstance(data[0], int)
or isinstance(data[0], bool)
or dtype != torch.long
):
raise RuntimeError(
"historical recursive device-scalar construction differs"
)
return torch.empty((1,), device=device, dtype=dtype).fill_(data[0])
def _historical_recursive_pass_with_tensor_route(
original: Callable[..., Any],
) -> Callable[..., Any]:
"""Clone one verified historical function with tensor-preserving scalars."""
if not isinstance(original, FunctionType):
raise RuntimeError("historical Resynthesis recursive pass is not a Python function")
recursive_globals = dict(original.__globals__)
recursive_globals["int"] = _tensor_preserving_historical_int
recursive_globals["torch"] = _HistoricalRecursiveTorchFacade()
tensor_route = FunctionType(
original.__code__,
recursive_globals,
original.__name__,
original.__defaults__,
original.__closure__,
)
frozen_globals = dict(recursive_globals)
frozen_globals["float"] = _tensor_preserving_historical_float
frozen_route = FunctionType(
original.__code__,
frozen_globals,
original.__name__,
original.__defaults__,
original.__closure__,
)
def recursive_pass(
self: Any,
*args: Any,
**kwargs: Any,
) -> Any:
implementation = (
frozen_route
if not self.training and not torch.is_grad_enabled()
else tensor_route
)
return implementation(self, *args, **kwargs)
recursive_pass.__kwdefaults__ = (
dict(original.__kwdefaults__)
if original.__kwdefaults__ is not None
else {}
)
recursive_pass.__annotations__ = dict(original.__annotations__)
recursive_pass.__qualname__ = original.__qualname__
recursive_pass.__doc__ = original.__doc__
return cast(Callable[..., Any], recursive_pass)
def _tensor_native_outcome_history_pack_step(
self: Any,
signals: Mapping[str, torch.Tensor],
domain_idx: int | torch.Tensor,
subdomain_idx: int | torch.Tensor,
stop_logit: torch.Tensor,
step_idx: int,
step_slots: int,
trace_summary: torch.Tensor,
) -> torch.Tensor:
"""Pack routed history without scalar indexing through the CPU."""
device = self.carry_h.device
dtype = self.domain_head.weight.dtype
utility = signals["utility"].reshape(()).to(device=device, dtype=dtype)
contradiction = signals["contradiction"].reshape(()).to(
device=device,
dtype=dtype,
)
rubric = signals["rubric"].reshape(()).to(device=device, dtype=dtype)
intent = signals["intent_conf"].reshape(()).to(device=device, dtype=dtype)
stop = stop_logit.reshape(()).to(device=device, dtype=dtype)
domain_t = _tensor_hotpath_index_boundary(
domain_idx,
reference=self.carry_h,
upper_bound=int(self.n_domains),
).to(device=device)
subdomain_t = _tensor_hotpath_index_boundary(
subdomain_idx,
reference=self.carry_h,
upper_bound=int(self.n_subdomains),
).to(device=device)
domain_one_hot = torch.zeros(
self.n_domains,
device=device,
dtype=dtype,
).scatter(
0,
domain_t.reshape(1),
torch.ones(1, device=device, dtype=dtype),
)
subdomain_one_hot = torch.zeros(
self.n_subdomains,
device=device,
dtype=dtype,
).scatter(
0,
subdomain_t.reshape(1),
torch.ones(1, device=device, dtype=dtype),
)
step_fraction = stop.new_full(
(),
float(step_idx + 1) / float(max(1, step_slots)),
)
trace = trace_summary.to(
device=device,
dtype=dtype,
).reshape(self.TRACE_FEAT_DIM)
return torch.cat(
(
utility.view(1),
contradiction.view(1),
rubric.view(1),
intent.view(1),
stop.view(1),
domain_one_hot,
subdomain_one_hot,
step_fraction.view(1),
trace,
),
dim=0,
)
def _tensor_native_historical_rbo_forward(
self: Any,
hidden: torch.Tensor,
baseline_logits: torch.Tensor | None = None,
intent_label: str | None = None,
swe_stage_target: torch.Tensor | None = None,
swe_stage_hidden: torch.Tensor | None = None,
project_logits: bool = True,
) -> Any:
"""Run the source-era parent while retaining model routes as tensors."""
self._hop_stop_delta_accum = hidden.new_zeros(())
self._hop_slice_delta_accum = hidden.new_zeros(())
control_hidden = hidden if swe_stage_hidden is None else swe_stage_hidden
intent_probs = self._intent_probs(control_hidden)
(
swe_stage_logits,
swe_stage_probs,
stage_domain_bias,
stage_subdomain_bias,
) = self.swe_stage_router(control_hidden)
stage_layer_bias = (
swe_stage_probs.mean(dim=0)
@ self.swe_stage_layer_affinity.to(
device=swe_stage_probs.device,
dtype=swe_stage_probs.dtype,
)
)
intent_idx_t = self._intent_index(intent_probs)
if intent_label is None:
intent_label = (
"model_owned_tensor"
if self.training
else self._intent_label(intent_probs, self.intent_names)
)
else:
intent_boundary_idx = (
self.intent_names.index(str(intent_label))
if str(intent_label) in self.intent_names
else self.intent_names.index("general")
if "general" in self.intent_names
else 0
)
intent_idx_t = intent_idx_t.new_empty((), dtype=torch.long).fill_(
intent_boundary_idx
)
hist_d, hist_s, _ = self.outcome_history.current_biases()
ctx_signals = self.context_encoder(control_hidden)
domain_conf_bias = ctx_signals["domain_conf"][0]
domain_count = self.arm_layout.n_domains
domain_confidence = domain_conf_bias[:domain_count]
domain_confidence = domain_confidence / (
domain_confidence.sum() + 1e-8
)
domain_log_boost = torch.log(domain_confidence.clamp(min=1e-6))
domain_probabilities, _, _, subdomain_indices = self.domain_router(
control_hidden,
intent_probs,
domain_log_bias=(
self.correction_domain_bias
+ hist_d.detach()
+ 0.1 * domain_log_boost
+ 0.2 * stage_domain_bias
),
subdomain_log_bias=(
self.correction_subdomain_bias
+ hist_s.detach()
+ 0.2 * stage_subdomain_bias
),
)
(
adjusted_domain_probabilities,
domain_collapse_score,
) = self.domain_collapse_detector(domain_probabilities)
self._last_domain_collapse_tensor = domain_collapse_score.detach()
domain_idx, subdomain_idx = _tensor_native_historical_domain_route(
adjusted_domain_probabilities,
subdomain_indices,
)
(
shaped_hidden,
routing_history,
feedback_history,
stop_logit_history,
traversal_history,
arms_used,
self_correction_steps,
) = self._recursive_pass(
hidden,
intent_probs,
domain_idx,
subdomain_idx,
ctx_signals,
stage_layer_bias,
control_hidden,
)
# The scalar history list is external receipt telemetry. Keep the current
# model value tensor-resident in the active graph; boundary snapshots can
# materialize it after the forward rather than synchronizing every wave.
self._last_domain_collapse_history = []
frozen_parent_hot_path = not self.training and not torch.is_grad_enabled()
if frozen_parent_hot_path:
# These four scalars supervise the historical parent only. Resynthesis
# freezes that parent and consumes shaped hidden/logit tensors, so
# rebuilding their BCE/CE stacks per retained wave cannot influence a
# routed-page gradient. Keep stateful history/context replay below.
dead_training_loss = hidden.new_zeros((), dtype=torch.float32)
contra_loss = dead_training_loss
stop_loss = dead_training_loss
correction_loss = dead_training_loss
stage_training_loss = dead_training_loss
else:
rubric_tensor = (
torch.stack([feedback["rubric"] for feedback in feedback_history])
if feedback_history
else hidden.new_zeros(0)
)
contrareactive_repulsion = getattr(
type(self),
"_resynthesis_contrareactive_repulsion",
None,
)
if not callable(contrareactive_repulsion):
raise RuntimeError(
"historical Resynthesis contrareactive owner is unavailable"
)
contra_loss = contrareactive_repulsion(
rubric_tensor,
floor=self.rbo_cfg.anti_floor,
weight=self.rbo_cfg.anti_weight,
)
stop_loss = self.stop_gate_training_loss()
correction_loss = self._correction_training_loss()
stage_training_loss = self._swe_stage_training_loss(
swe_stage_logits,
swe_stage_probs,
swe_stage_target,
)
history_loss = self.outcome_history_training_loss(
feedback_history,
stop_logit_history,
domain_idx=domain_idx,
subdomain_idx=subdomain_idx,
)
context_loss = (
dead_training_loss
if frozen_parent_hot_path
else self._context_dag_training_loss(
hidden,
intent_probs,
ctx_signals,
intent_idx_t,
feedback_history,
domain_idx=domain_idx,
subdomain_idx=subdomain_idx,
stage_domain_bias=stage_domain_bias,
stage_subdomain_bias=stage_subdomain_bias,
)
)
steps = len(feedback_history)
traversal_stop_reason = getattr(
self,
"_last_traversal_stop_reason",
"trained_gate",
)
shaped_logits = (
self.shaped_logits_from_hidden(
shaped_hidden,
baseline_logits,
hidden,
)
if project_logits
else None
)
if not self.training and shaped_logits is not None:
shaped_logits = self.rubric_bias.apply_bias(
shaped_logits,
shaped_hidden.detach(),
intent_probs.detach(),
strength=float(self.rbo_cfg.rubric_bias_strength),
)
self._detach_outcome_history_carry()
result_type = getattr(type(self), "_resynthesis_rbo_result_type", None)
stages = getattr(type(self), "_resynthesis_swe_stages", None)
if not callable(result_type) or not isinstance(stages, tuple):
raise RuntimeError("historical Resynthesis result boundary is unavailable")
stage_index_boundary = (
0
if frozen_parent_hot_path
else int(
swe_stage_probs.mean(dim=0)
.argmax()
.detach()
.to(device="cpu", dtype=torch.long)
)
)
return result_type(
shaped_hidden=shaped_hidden,
baseline_hidden=hidden,
shaped_logits=shaped_logits,
baseline_logits=baseline_logits,
steps=steps,
stop_reason=traversal_stop_reason,
intent_probs=intent_probs,
intent_label=intent_label,
swe_stage_probs=swe_stage_probs,
swe_stage_training_loss=stage_training_loss,
routing_history=routing_history,
feedback_history=feedback_history,
stop_logit_history=stop_logit_history,
traversal_history=traversal_history,
arms_used=arms_used,
domain_idx=domain_idx,
subdomain_idx=subdomain_idx,
self_correction_steps=self_correction_steps,
contrareactive_loss=contra_loss,
correction_loss=correction_loss,
stop_gate_loss=stop_loss,
outcome_history_loss=history_loss,
context_training_loss=context_loss,
proof=(
{
"resynthesisRBOActive": True,
"none_architecture": "Nest of Native Experts",
"none_model_owned": True,
"none_transfer_active": self._last_none_phase is not None,
}
if frozen_parent_hot_path
else {
"resynthesisRBOActive": True,
"none_architecture": "Nest of Native Experts",
"none_model_owned": True,
"none_transfer_active": self._last_none_phase is not None,
"none_outcome_writes": int(
self.none_fabric.outcome_writes.detach().cpu()
),
"none_phase_visits": (
self.none_fabric.phase_visits.detach().cpu().tolist()
),
"none_shaped_logits_present": isinstance(
shaped_logits,
torch.Tensor,
),
"none_pathway_phase": (
self._last_hop_signal.none_pathway.phase.detach().cpu().tolist()
if self._last_hop_signal is not None
else []
),
"none_pathway_gap": (
float(
self._last_hop_signal.none_pathway.gap.detach().cpu()
)
if self._last_hop_signal is not None
else 0.0
),
"none_pathway_completion": (
float(
self._last_hop_signal.none_pathway.completion.detach().cpu()
)
if self._last_hop_signal is not None
else 0.0
),
"none_pathway_runtime_gain": (
float(
self._last_hop_signal.none_pathway.runtime_gain.detach().cpu()
)
if self._last_hop_signal is not None
else 0.0
),
"intent": intent_label,
"swe_stage_idx": stage_index_boundary,
"swe_stage": stages[stage_index_boundary],
"domain_idx": domain_idx.detach().clone(),
"subdomain_idx": subdomain_idx.detach().clone(),
"steps": steps,
# The recursive pass returns the count owned by this exact forward.
# Reuse it in both the typed result and proof so mutable parent
# attribute layouts cannot substitute stale or default telemetry.
"self_correction_steps": self_correction_steps,
"traversal_layers": (
traversal_history[-1] if traversal_history else []
),
"traversal_stop_reason": traversal_stop_reason,
"arms_used": arms_used,
"outcome_memory": self.outcome_memory_snapshot(),
"identity_state": {
"boundary_only": True,
"identity_dim": int(self.rbo_cfg.identity_dim),
"n_layers": len(self.experts),
},
"outcome_history_steps": len(feedback_history),
"outcome_history_carry_norm": float(
self.outcome_history.carry_h.norm().detach(),
),
"hop_traversal": {
"from_hop_idx": (
self._last_hop_signal.from_hop_idx
if self._last_hop_signal
else 0
),
"dest_hop_idx": (
self._last_hop_signal.dest_hop_idx
if self._last_hop_signal
else 0
),
"hop_confidence": (
float(self._last_hop_signal.hop_confidence.detach())
if self._last_hop_signal is not None
else 0.0
),
"hop_bonus": (
float(self._last_hop_signal.hop_bonus.detach())
if self._last_hop_signal is not None
else 0.0
),
"none_stop_delta": (
float(
self._last_hop_signal.none_pathway.stop_delta.detach()
)
if self._last_hop_signal is not None
else 0.0
),
"none_slice_delta": (
float(
self._last_hop_signal.none_pathway.slice_delta.detach()
)
if self._last_hop_signal is not None
else 0.0
),
"stop_delta_accum": self._boundary_float(
self._hop_stop_delta_accum
),
"slice_delta_accum": self._boundary_float(
self._hop_slice_delta_accum
),
},
}
),
)
def _historical_rbo_init_with_tensor_intent_mapping(
self: Any,
*args: Any,
**kwargs: Any,
) -> None:
"""Materialize the source-era intent map once, before the RBO moves to CUDA."""
owner = type(self)
original = getattr(owner, "_resynthesis_original_rbo_init", None)
intent_to_capability = getattr(
owner,
"_resynthesis_intent_to_capability_boundary",
None,
)
capability_count = getattr(
owner,
"_resynthesis_intent_capability_count",
None,
)
if not callable(original) or not callable(intent_to_capability):
raise RuntimeError("historical Resynthesis intent constructor was not preserved")
if (
not isinstance(capability_count, int)
or isinstance(capability_count, bool)
or capability_count < 1
):
raise RuntimeError("historical Resynthesis capability count is invalid")
original(self, *args, **kwargs)
intent_names = getattr(self, "intent_names", None)
arm_layout = getattr(self, "arm_layout", None)
reference = getattr(self, "layer_pick_logits", None)
if (
not isinstance(intent_names, (list, tuple))
or not intent_names
or not all(isinstance(name, str) for name in intent_names)
or not isinstance(reference, torch.Tensor)
):
raise RuntimeError("historical Resynthesis intent geometry is malformed")
n_domains = getattr(arm_layout, "n_domains", None)
n_subdomains = getattr(arm_layout, "n_subdomains", None)
if (
not isinstance(n_domains, int)
or isinstance(n_domains, bool)
or n_domains < 1
or not isinstance(n_subdomains, int)
or isinstance(n_subdomains, bool)
or n_subdomains < 1
):
raise RuntimeError("historical Resynthesis intent route geometry is malformed")
# This is the one explicit source-compatibility boundary. It exhaustively
# resolves every classifier row through the hash-verified source-era map.
# The active forward only gathers these tensors; it has no string/default
# route and never reads the selected CUDA scalar on the host.
capability_indices = tuple(
intent_to_capability(intent_name) for intent_name in intent_names
)
if not all(
isinstance(index, int)
and not isinstance(index, bool)
and 0 <= index < capability_count
for index in capability_indices
):
raise RuntimeError("historical Resynthesis intent capability map is invalid")
capability_map = torch.tensor(
capability_indices,
device=reference.device,
dtype=torch.long,
)
domain_map = capability_map.remainder(n_domains)
subdomain_map = torch.div(
capability_map,
n_domains,
rounding_mode="floor",
).remainder(n_subdomains)
for name, value in (
("_resynthesis_intent_capability_map", capability_map),
("_resynthesis_intent_domain_map", domain_map),
("_resynthesis_intent_subdomain_map", subdomain_map),
):
if hasattr(self, name):
raise RuntimeError("historical Resynthesis intent map is already attached")
self.register_buffer(name, value, persistent=False)
none_fabric = getattr(self, "none_fabric", None)
none_geometry = getattr(none_fabric, "geometry", None)
phase_count = getattr(none_geometry, "n_phases", None)
if (
not isinstance(phase_count, int)
or isinstance(phase_count, bool)
or phase_count < 1
):
raise RuntimeError("historical Resynthesis phase geometry is malformed")
self.register_buffer(
"_resynthesis_none_phase_selector_t",
torch.eye(
phase_count,
device=reference.device,
dtype=reference.dtype,
),
persistent=False,
)
def _tensor_native_historical_none_step_phase(
self: Any,
*,
commit: Any,
phase_hidden: torch.Tensor,
observation_hidden: torch.Tensor,
step_idx: int,
) -> Any:
"""Run one historical Fabric phase without a host-created CUDA scalar."""
phase_selector_t = getattr(
self,
"_resynthesis_none_phase_selector_t",
None,
)
none_fabric = getattr(self, "none_fabric", None)
none_geometry = getattr(none_fabric, "geometry", None)
phase_count = getattr(none_geometry, "n_phases", None)
step_phase = getattr(none_fabric, "step_phase", None)
if (
not isinstance(step_idx, int)
or isinstance(step_idx, bool)
or not isinstance(phase_count, int)
or isinstance(phase_count, bool)
or phase_count < 1
or not isinstance(phase_selector_t, torch.Tensor)
or phase_selector_t.shape != (phase_count, phase_count)
or phase_selector_t.device != phase_hidden.device
or phase_selector_t.dtype != phase_hidden.dtype
or not callable(step_phase)
):
raise RuntimeError("historical Resynthesis phase selector differs")
expert_state, layer_state = self._none_expert_layer_state(phase_hidden)
phase_weights = phase_selector_t[
step_idx % phase_count
].reshape(1, phase_count).expand(phase_hidden.shape[0], -1)
compact_phase = torch.tanh(self.none_hidden_in(phase_hidden))
compact_observation = torch.tanh(
self.none_hidden_in(observation_hidden)
)
return step_phase(
commit,
compact_phase,
compact_observation,
expert_state,
layer_state,
phase_weights,
)
def _tensor_native_historical_intent_route(
self: Any,
intent_probs: torch.Tensor,
intents: list[str] | tuple[str, ...],
) -> _HistoricalTensorIntentRoute:
"""Select and map the exact historical intent without a CUDA host read."""
capability_map = getattr(self, "_resynthesis_intent_capability_map", None)
domain_map = getattr(self, "_resynthesis_intent_domain_map", None)
subdomain_map = getattr(self, "_resynthesis_intent_subdomain_map", None)
expected_shape = (len(intents),)
if (
intent_probs.ndim != 3
or intent_probs.shape[-1] != len(intents)
or not isinstance(capability_map, torch.Tensor)
or capability_map.shape != expected_shape
or capability_map.device != intent_probs.device
or capability_map.dtype != torch.long
or not isinstance(domain_map, torch.Tensor)
or domain_map.shape != expected_shape
or domain_map.device != intent_probs.device
or domain_map.dtype != torch.long
or not isinstance(subdomain_map, torch.Tensor)
or subdomain_map.shape != expected_shape
or subdomain_map.device != intent_probs.device
or subdomain_map.dtype != torch.long
):
raise RuntimeError("historical Resynthesis tensor intent map differs")
intent_index = intent_probs.mean(dim=(0, 1)).argmax().to(
dtype=torch.long
).reshape(())
selector = intent_index.reshape(1)
return _HistoricalTensorIntentRoute(
intent_index=intent_index,
capability_index=capability_map.index_select(0, selector).reshape(()),
domain_index=domain_map.index_select(0, selector).reshape(()),
subdomain_index=subdomain_map.index_select(0, selector).reshape(()),
)
def _tensor_native_none_pathway_training_loss(
self: Any,
packet: Any,
*,
receiver_capability_idx: int | torch.Tensor,
phase_target: torch.Tensor,
gap_target: torch.Tensor,
completion_target: torch.Tensor,
) -> torch.Tensor:
"""Preserve the historical NoNE loss with tensor receiver indexing."""
transfer = packet.transfer
receiver = _tensor_hotpath_index_boundary(
receiver_capability_idx,
reference=transfer,
upper_bound=int(self.n_capabilities),
)
epsilon = torch.finfo(transfer.dtype).eps
phase_target = phase_target.to(
device=packet.phase.device,
dtype=packet.phase.dtype,
).reshape_as(packet.phase)
phase_target = phase_target / phase_target.sum().clamp_min(epsilon)
gap_target = gap_target.to(
device=packet.gap.device,
dtype=packet.gap.dtype,
).reshape(())
completion_target = completion_target.to(
device=packet.completion.device,
dtype=packet.completion.dtype,
).reshape(())
phase_loss = -(phase_target * packet.phase.clamp_min(epsilon).log()).sum()
transfer_loss = -transfer.index_select(
0,
receiver.reshape(1),
).reshape(()).clamp_min(epsilon).log()
state_loss = F.binary_cross_entropy(
packet.gap,
gap_target.clamp(0.0, 1.0),
)
state_loss = state_loss + F.binary_cross_entropy(
packet.completion,
completion_target.clamp(0.0, 1.0),
)
desired_gain = torch.maximum(
gap_target,
torch.ones_like(completion_target) - completion_target,
).clamp(0.0, 1.0)
gain_loss = F.mse_loss(packet.runtime_gain, desired_gain)
stop_target = completion_target - gap_target
slice_target = gap_target - completion_target
traversal_loss = F.mse_loss(packet.stop_delta, stop_target)
traversal_loss = traversal_loss + F.mse_loss(
packet.slice_delta,
slice_target,
)
route_target = (gap_target - completion_target).reshape(())
route_loss = F.mse_loss(packet.domain_delta.mean(), route_target)
route_loss = route_loss + F.mse_loss(packet.layer_delta.mean(), route_target)
route_loss = route_loss + F.mse_loss(packet.expert_delta.mean(), route_target)
return cast(
torch.Tensor,
phase_loss
+ transfer_loss
+ state_loss
+ 0.25 * gain_loss
+ 0.10 * traversal_loss
+ 0.05 * route_loss,
)
def _historical_tensor_native_identity_supervision_loss(
self: Any,
hidden: torch.Tensor,
intent_probs: torch.Tensor,
ctx_signals: Mapping[str, torch.Tensor],
*,
target_domain_idx: int | torch.Tensor,
target_capability_idx: int | torch.Tensor,
) -> torch.Tensor:
"""Preserve historical identity supervision with tensor route indices."""
n_layers = len(self.experts)
if n_layers <= 0:
return hidden.new_zeros(())
transfer_weights = ctx_signals.get("transfer_weights")
layer_scores = self._layer_identity_bias(
hidden,
intent_probs,
transfer_weights,
n_layers,
)
target_domain = _tensor_hotpath_index_boundary(
target_domain_idx,
reference=self.domain_layer_affinity,
upper_bound=int(self.domain_layer_affinity.shape[0]),
)
layer_target = F.softmax(
self.domain_layer_affinity.index_select(
0,
target_domain.reshape(1),
)
.squeeze(0)[:n_layers]
.detach(),
dim=-1,
)
layer_loss = -(layer_target * F.log_softmax(layer_scores, dim=-1)).sum()
capability_index = _tensor_hotpath_index_boundary(
target_capability_idx,
reference=hidden,
upper_bound=10,
)
capability_target = F.one_hot(
capability_index,
num_classes=10,
).to(device=hidden.device, dtype=hidden.dtype)
capability_losses = tuple(
F.mse_loss(
torch.sigmoid(
expert.flat_expert_capability.to(
device=hidden.device,
dtype=hidden.dtype,
)
),
capability_target.unsqueeze(0).expand_as(
expert.flat_expert_capability
),
)
for expert in self.experts
)
capability_loss = (
torch.stack(capability_losses).mean()
if capability_losses
else hidden.new_zeros(())
)
layer_capability = torch.stack(
tuple(expert.capability_scores for expert in self.experts),
dim=0,
).to(device=hidden.device, dtype=hidden.dtype)
layer_capability_loss = F.mse_loss(
torch.sigmoid(layer_capability),
capability_target.unsqueeze(0).expand_as(layer_capability),
)
specialization = torch.stack(
tuple(expert.specialization for expert in self.experts),
dim=0,
).to(device=hidden.device, dtype=hidden.dtype)
specialization_target = (layer_target * float(n_layers)).clamp(0.0, 1.0)
specialization_loss = F.mse_loss(
torch.sigmoid(specialization),
specialization_target,
)
routing_fingerprints = torch.stack(
tuple(expert.routing_fingerprint for expert in self.experts),
dim=0,
)
routing_fingerprint_loss = self._identity_orthogonality_loss(
routing_fingerprints
)
return cast(
torch.Tensor,
layer_loss
+ 0.10 * capability_loss
+ 0.10 * layer_capability_loss
+ 0.05 * specialization_loss
+ self.rbo_cfg.identity_diversity_weight
* self._identity_diversity_loss()
+ self.rbo_cfg.identity_diversity_weight * routing_fingerprint_loss
+ self.rbo_cfg.profile_diversity_weight * self._profile_diversity_loss(),
)
def _historical_tensor_native_context_dag_training_loss(
self: Any,
hidden: torch.Tensor,
intent_probs: torch.Tensor,
ctx_signals: Mapping[str, torch.Tensor],
intent_label: str | torch.Tensor | _HistoricalTensorIntentRoute,
feedback_history: list[Mapping[str, torch.Tensor]],
*,
domain_idx: int | torch.Tensor,
subdomain_idx: int | torch.Tensor,
stage_domain_bias: torch.Tensor | None = None,
stage_subdomain_bias: torch.Tensor | None = None,
) -> torch.Tensor:
"""Run the exact source-era semantic loss from a tensor intent route."""
if isinstance(intent_label, str):
original = getattr(
type(self),
"_resynthesis_original_context_dag_training_loss",
None,
)
if not callable(original):
raise RuntimeError("historical Resynthesis context loss was not preserved")
return cast(
torch.Tensor,
original(
self,
hidden,
intent_probs,
ctx_signals,
intent_label,
feedback_history,
domain_idx=domain_idx,
subdomain_idx=subdomain_idx,
stage_domain_bias=stage_domain_bias,
stage_subdomain_bias=stage_subdomain_bias,
),
)
capability_map = self._resynthesis_intent_capability_map
domain_map = self._resynthesis_intent_domain_map
subdomain_map = self._resynthesis_intent_subdomain_map
if isinstance(intent_label, torch.Tensor):
intent_route: _HistoricalTensorIntentRoute | None = None
route_intent_index = intent_label
else:
intent_route = _normalize_historical_tensor_intent_route(intent_label)
route_intent_index = intent_route.intent_index
intent_index = _tensor_hotpath_index_boundary(
route_intent_index,
reference=capability_map,
upper_bound=int(capability_map.numel()),
)
selector = intent_index.reshape(1)
capability_index = capability_map.index_select(0, selector).reshape(())
target_domain = domain_map.index_select(0, selector).reshape(())
target_subdomain = subdomain_map.index_select(0, selector).reshape(())
if intent_route is not None:
# Older parent generations returned a typed route packet and therefore
# carry redundant mapped indices that remain exact integrity proofs.
# Current native parents already pass the model-owned scalar tensor
# directly; its maps are gathered above without a host readback.
torch._assert_async(
capability_index.eq(intent_route.capability_index),
"historical Resynthesis intent capability route differs",
)
torch._assert_async(
target_domain.eq(intent_route.domain_index),
"historical Resynthesis intent domain route differs",
)
torch._assert_async(
target_subdomain.eq(intent_route.subdomain_index),
"historical Resynthesis intent subdomain route differs",
)
zero = hidden.new_zeros(())
epsilon = torch.finfo(hidden.float().dtype).eps
domain_confidence = ctx_signals["domain_conf"].clamp_min(epsilon)
capability_loss = -torch.log(
domain_confidence.index_select(
1,
capability_index.reshape(1),
).clamp_min(epsilon)
).mean()
depth_probabilities = ctx_signals["depth_probs"].clamp_min(epsilon)
depth_index = min(
max(len(feedback_history) - 1, 0),
depth_probabilities.shape[-1] - 1,
)
depth_loss = -torch.log(
depth_probabilities[:, depth_index].clamp_min(epsilon)
).mean()
if feedback_history:
rubric_values = torch.stack(
tuple(
feedback["rubric"]
.detach()
.reshape(())
.to(device=hidden.device, dtype=hidden.dtype)
for feedback in feedback_history
)
)
completion_target = rubric_values.mean().clamp(0.0, 1.0)
complexity_target = torch.ones_like(completion_target) - completion_target
else:
completion_target = hidden.new_ones(()).mul_(0.5)
complexity_target = hidden.new_ones(()).mul_(0.5)
complexity_loss = F.mse_loss(
ctx_signals["complexity"],
complexity_target.expand_as(ctx_signals["complexity"]),
)
transfer_owner = self.cross_domain_router.knowledge_transfer
relatedness = (
transfer_owner.relatedness_prior.index_select(
0,
capability_index.reshape(1),
)
.squeeze(0)
.to(device=hidden.device, dtype=hidden.dtype)
.clone()
)
capability_mask = torch.arange(
transfer_owner.n_capabilities,
device=hidden.device,
).ne(capability_index)
relatedness = relatedness * capability_mask.to(dtype=relatedness.dtype)
relatedness_mass = relatedness.sum()
self_capability = F.one_hot(
capability_index,
num_classes=transfer_owner.n_capabilities,
).to(device=hidden.device, dtype=hidden.dtype)
relatedness = torch.where(
relatedness_mass.le(epsilon),
self_capability,
relatedness / relatedness_mass.clamp_min(epsilon),
)
transfer_weights = ctx_signals["transfer_weights"].clamp_min(epsilon)
transfer_loss = F.kl_div(
transfer_weights.log(),
relatedness.unsqueeze(0).expand_as(transfer_weights),
reduction="batchmean",
)
visit_scores, _ = transfer_owner.neural_dag(
hidden=hidden,
start_cap=capability_index,
)
dag_loss = F.kl_div(
visit_scores.clamp_min(epsilon).log(),
relatedness,
reduction="sum",
)
outcome_pressure = self.arm_registry.domain_slot_pressure().sum(dim=-1)
donor = transfer_owner.donor_weights_for_pressure(
target_domain,
outcome_pressure,
hidden=hidden,
)
none_packet = transfer_owner.build_none_pathway_packet(
target_domain,
donor=donor,
hidden=hidden,
outcome_pressure=outcome_pressure,
n_layers=len(self.experts),
n_experts=self.cfg.total_experts,
)
gap_target = complexity_target.clamp(0.0, 1.0)
phase_target = torch.stack(
(
0.10 * (torch.ones_like(gap_target) - gap_target),
0.35 * gap_target,
0.65 * gap_target,
completion_target,
)
)
receiver_capability = relatedness.argmax().to(dtype=torch.long).reshape(())
none_pathway_loss = transfer_owner.none_pathway_training_loss(
none_packet,
receiver_capability_idx=receiver_capability,
phase_target=phase_target,
gap_target=gap_target,
completion_target=completion_target,
)
domain_probabilities, subdomain_probabilities, _, _ = self.domain_router(
hidden,
intent_probs,
domain_log_bias=stage_domain_bias,
subdomain_log_bias=stage_subdomain_bias,
)
route_loss = -torch.log(
domain_probabilities.index_select(
1,
target_domain.reshape(1),
).clamp_min(epsilon)
).mean()
route_loss = route_loss - torch.log(
subdomain_probabilities.index_select(
1,
target_subdomain.reshape(1),
).clamp_min(epsilon)
).mean()
observed_domain = (
domain_idx.detach().to(device=hidden.device, dtype=torch.long).reshape(())
if isinstance(domain_idx, torch.Tensor)
else hidden.new_empty((), dtype=torch.long).fill_(domain_idx)
)
observed_subdomain = (
subdomain_idx.detach().to(
device=hidden.device,
dtype=torch.long,
).reshape(())
if isinstance(subdomain_idx, torch.Tensor)
else hidden.new_empty((), dtype=torch.long).fill_(subdomain_idx)
)
observed_valid = (
observed_domain.ge(0)
.logical_and(observed_domain.lt(domain_probabilities.shape[-1]))
.logical_and(observed_subdomain.ge(0))
.logical_and(
observed_subdomain.lt(subdomain_probabilities.shape[-1])
)
)
observed_domain = observed_domain.clamp(
min=0,
max=domain_probabilities.shape[-1] - 1,
)
observed_subdomain = observed_subdomain.clamp(
min=0,
max=subdomain_probabilities.shape[-1] - 1,
)
observed_slot_loss = -torch.log(
domain_probabilities.index_select(
1,
observed_domain.reshape(1),
).clamp_min(epsilon)
).mean()
observed_slot_loss = observed_slot_loss - torch.log(
subdomain_probabilities.index_select(
1,
observed_subdomain.reshape(1),
).clamp_min(epsilon)
).mean()
observed_slot_loss = (
observed_slot_loss
* observed_valid.to(device=hidden.device, dtype=hidden.dtype)
)
identity_loss = self._identity_supervision_loss(
hidden,
intent_probs,
ctx_signals,
target_domain_idx=target_domain,
target_capability_idx=capability_index,
)
return cast(
torch.Tensor,
0.35 * capability_loss
+ 0.20 * depth_loss
+ 0.15 * complexity_loss
+ 0.25 * transfer_loss
+ 0.25 * dag_loss
+ 0.20 * none_pathway_loss
+ 0.20 * route_loss
+ 0.05 * observed_slot_loss
+ 0.08 * identity_loss
+ zero,
)
def _historical_none_fabric_init_with_tensor_constants(
self: Any,
geometry: Any,
) -> None:
"""Install the immutable Fabric's device-resident scalar geometry.
The historical ``form_commit`` rebuilt the atomic-graph width with
``Tensor.new_tensor`` on every parent forward. On CUDA that host-to-device
scalar copy synchronizes the parent stream. Registering the exact scalar
once keeps it with the module across device/dtype moves. It is
nonpersistent because it is derived entirely from the already-authoritative
geometry and therefore must not alter checkpoint identity.
"""
original = getattr(
type(self),
"_resynthesis_original_none_fabric_init",
None,
)
if not callable(original):
raise RuntimeError("historical NoNE Fabric constructor was not preserved")
original(self, geometry)
graph_node_credit = getattr(self, "graph_node_credit", None)
if not isinstance(graph_node_credit, torch.Tensor):
raise RuntimeError("historical NoNE Fabric graph geometry differs")
self.register_buffer(
"_resynthesis_atomic_graph_slots_scale_t",
graph_node_credit.new_full((), geometry.atomic_graph_slots),
persistent=False,
)
def _tensor_native_historical_none_fabric_form_commit(
self: Any,
surface_glyphs: torch.Tensor,
evidence_logits: torch.Tensor,
capability_gap: torch.Tensor,
) -> Any:
"""Form the exact historical Fabric commit without CUDA scalar copies."""
if surface_glyphs.ndim != 3:
raise ValueError(
"surface_glyphs must have shape [batch, obligations, hidden]"
)
if evidence_logits.ndim != 2:
raise ValueError(
"evidence_logits must have shape [batch, obligations]"
)
if capability_gap.ndim != 2 or capability_gap.shape[-1] != 3:
raise ValueError("capability_gap must have shape [batch, 3]")
if surface_glyphs.shape[1] != self.geometry.n_obligations:
raise ValueError(
"surface_glyphs obligation axis does not match Fabric geometry"
)
identity = self.architecture_identity.unsqueeze(1)
encoded = torch.tanh(self.surface_encoder(surface_glyphs + identity))
preliminary_state = self.commit_encoder(encoded.mean(dim=1))
learned_evidence = self.obligation_evidence(encoded).squeeze(-1)
support = torch.sigmoid(evidence_logits + learned_evidence)
requirements = torch.sigmoid(self.commit_requirement_logits).unsqueeze(0)
obligation_margin = support - requirements
learned_memory = self.outcome_memory_reader(
self.outcome_memory.mean(dim=0, keepdim=True)
).expand(surface_glyphs.shape[0], -1)
state = torch.tanh(
preliminary_state
+ self.commit_obligation_encoder(obligation_margin)
+ self.capability_pressure_encoder(capability_gap)
+ learned_memory
)
atomic_graph = self.atomic_task_graph(
state,
encoded,
support,
capability_gap,
)
graph_credit_logits = (
self.graph_node_credit - self.graph_rollback_pressure
).to(device=state.device, dtype=state.dtype)
graph_slots_scale_t = self._resynthesis_atomic_graph_slots_scale_t.to(
device=state.device,
dtype=state.dtype,
)
graph_credit_scale = (
torch.softmax(graph_credit_logits, dim=-1) * graph_slots_scale_t
)
graph_ready_weights = (
atomic_graph.ready_weights * graph_credit_scale.unsqueeze(0)
)
graph_repair_scale = 2.0 * torch.sigmoid(
self.graph_repair_credit.to(device=state.device, dtype=state.dtype)
)
graph_repair_weights = (
atomic_graph.repair_weights * graph_repair_scale.unsqueeze(0)
)
session_present = self.session_graph_present.to(
device=state.device,
dtype=state.dtype,
)
if self.training:
session_present = torch.zeros_like(session_present)
prior_frozen = self.session_graph_frozen_weights.to(
device=state.device,
dtype=state.dtype,
)
node_keep = (session_present * prior_frozen).reshape(1, -1)
graph_node_state = torch.lerp(
atomic_graph.node_state,
self.session_graph_node_state.to(
device=state.device,
dtype=state.dtype,
)
.unsqueeze(0)
.expand_as(atomic_graph.node_state),
node_keep.unsqueeze(-1),
)
edge_keep = node_keep.unsqueeze(-1) * node_keep.unsqueeze(-2)
graph_edge_weights = torch.lerp(
atomic_graph.edge_weights,
self.session_graph_edge_weights.to(
device=state.device,
dtype=state.dtype,
)
.unsqueeze(0)
.expand_as(atomic_graph.edge_weights),
edge_keep,
)
graph_validated_weights = torch.maximum(
atomic_graph.validated_weights,
self.session_graph_validated_weights.to(
device=state.device,
dtype=state.dtype,
).unsqueeze(0)
* session_present,
)
graph_frozen_weights = torch.maximum(
atomic_graph.frozen_weights,
prior_frozen.unsqueeze(0) * session_present,
)
graph_repair_weights = graph_repair_weights * (
1.0 - graph_frozen_weights
)
graph_summary_weights = (
graph_ready_weights + graph_repair_weights + graph_frozen_weights
)
graph_summary = self.atomic_task_graph.graph_summary_head(
torch.einsum(
"bn,bnh->bh",
graph_summary_weights,
graph_node_state,
)
/ graph_summary_weights.sum(dim=-1, keepdim=True).clamp_min(
torch.finfo(graph_summary_weights.dtype).tiny
)
)
state = torch.tanh(state + graph_summary)
if not self.training:
with torch.no_grad():
self.session_graph_node_state.copy_(
graph_node_state.detach().mean(dim=0)
)
self.session_graph_edge_weights.copy_(
graph_edge_weights.detach().mean(dim=0)
)
self.session_graph_validated_weights.copy_(
graph_validated_weights.detach().mean(dim=0)
)
self.session_graph_frozen_weights.copy_(
graph_frozen_weights.detach().mean(dim=0)
)
self.session_graph_repair_weights.copy_(
graph_repair_weights.detach().mean(dim=0)
)
self.session_graph_present.fill_(True)
scope_completeness = obligation_margin.amin(dim=-1, keepdim=True)
edit_authorization = (scope_completeness >= 0).to(surface_glyphs.dtype)
mutating_probability = torch.einsum(
"bno,o->bn",
atomic_graph.operation_probs,
self.atomic_task_graph.mutating_operation_mask.to(
device=state.device,
dtype=state.dtype,
),
)
graph_execution_authorization = (
1.0
- mutating_probability
+ mutating_probability * edit_authorization
).clamp(min=0.0, max=1.0)
unresolved_tickets = (
encoded * torch.relu(-obligation_margin).unsqueeze(-1)
)
commit_expert_source = self._record_causal_gradient(
self.commit_expert_source(state),
0,
)
commit_expert_destination = self._record_causal_gradient(
self.commit_expert_destination(state),
1,
)
commit_layer_source = self._record_causal_gradient(
self.commit_layer_source(state),
2,
)
commit_layer_destination = self._record_causal_gradient(
self.commit_layer_destination(state),
3,
)
expert_logits = self._pair_logits(
commit_expert_source,
commit_expert_destination,
self.expert_pair_prior,
self.expert_transfer_credit,
)
layer_logits = self._pair_logits(
commit_layer_source,
commit_layer_destination,
self.layer_pair_prior,
self.layer_transfer_credit,
)
research_branch_logits = self._record_causal_gradient(
self.research_branch_head(state),
4,
)
unresolved_pressure = torch.relu(-scope_completeness)
research_pressure = torch.sigmoid(
self.research_pressure_head(state)
+ unresolved_pressure
+ capability_gap[:, 0:1]
)
tool_build_pressure = torch.sigmoid(
self.tool_build_head(state)
+ unresolved_pressure
+ capability_gap[:, 1:2]
)
environment_build_pressure = torch.sigmoid(
self.environment_build_head(state)
+ unresolved_pressure
+ capability_gap[:, 2:3]
)
commit_type = getattr(
type(self),
"_resynthesis_none_fabric_commit_type",
None,
)
if not isinstance(commit_type, type):
raise RuntimeError("historical NoNE Fabric commit type is unavailable")
return commit_type(
state=state,
obligation_support=support,
unresolved_tickets=unresolved_tickets,
scope_completeness=scope_completeness,
edit_authorization=edit_authorization,
expert_source_state=commit_expert_source,
expert_destination_state=commit_expert_destination,
layer_source_state=commit_layer_source,
layer_destination_state=commit_layer_destination,
expert_transfer_intent=self._off_diagonal_softmax(expert_logits),
layer_transfer_intent=self._off_diagonal_softmax(layer_logits),
research_branch_weights=F.softmax(research_branch_logits, dim=-1),
research_pressure=research_pressure,
tool_build_pressure=tool_build_pressure,
environment_build_pressure=environment_build_pressure,
graph_node_state=graph_node_state,
graph_operation_logits=atomic_graph.operation_logits,
graph_operation_probs=atomic_graph.operation_probs,
graph_argument_state=atomic_graph.argument_state,
graph_observation_binding=atomic_graph.observation_binding,
graph_execution_authorization=graph_execution_authorization,
graph_edge_weights=graph_edge_weights,
graph_refinement_parent_weights=(
atomic_graph.refinement_parent_weights
),
graph_atomicity=atomic_graph.atomicity,
graph_input_interfaces=atomic_graph.input_interfaces,
graph_output_interfaces=atomic_graph.output_interfaces,
graph_interface_compatibility=(
atomic_graph.interface_compatibility
),
graph_simulation_risk=atomic_graph.simulation_risk,
graph_ready_weights=graph_ready_weights,
graph_parallel_weights=atomic_graph.parallel_weights,
graph_validated_weights=graph_validated_weights,
graph_affected_weights=atomic_graph.affected_weights,
graph_frozen_weights=graph_frozen_weights,
graph_repair_weights=graph_repair_weights,
graph_summary=graph_summary,
)
def _install_historical_tensor_native_hotpaths(
rbo_module: Any,
experts_module: Any,
transfer_module: Any,
traversal_module: Any,
) -> str:
"""Bind hash-verified historical classes to equivalent tensor hot paths."""
expert_bank = getattr(experts_module, "FactorizedGranularExpertBank", None)
transfer = getattr(transfer_module, "KnowledgeTransferSurfaces", None)
intent_to_capability = getattr(
transfer_module,
"intent_to_capability",
None,
)
capability_count = getattr(transfer_module, "N_CAPABILITIES", None)
pathway_packet_type = getattr(transfer_module, "NoNEPathwayPacket", None)
hop_router = getattr(traversal_module, "CrossDomainHopRouter", None)
signal_type = getattr(traversal_module, "HopTraversalSignal", None)
outcome_history = getattr(traversal_module, "OutcomeHistoryEncoder", None)
rbo_type = getattr(rbo_module, _legacy_parent_class_name("RBO"), None)
rbo_result_type = getattr(
rbo_module,
_legacy_parent_class_name("RBOResult"),
None,
)
rubric_bias_type = getattr(rbo_module, "RubricLogitBias", None)
contrareactive_repulsion = getattr(
rbo_module,
"contrareactive_repulsion",
None,
)
swe_stages = getattr(rbo_module, "SWE_STAGES", None)
correction_controller = getattr(
rbo_module,
"SelfCorrectionController",
None,
)
domain_collapse_detector = getattr(
rbo_module,
"DomainCollapseDetector",
None,
)
none_fabric_type = getattr(
rbo_module,
_legacy_parent_class_name("NoNEFabric"),
None,
)
none_fabric_commit_type = getattr(rbo_module, "NoNEFabricCommit", None)
if not all(
isinstance(value, type)
for value in (
rbo_type,
rbo_result_type,
expert_bank,
transfer,
hop_router,
outcome_history,
rubric_bias_type,
none_fabric_type,
none_fabric_commit_type,
)
):
raise RuntimeError("historical Resynthesis hot-path classes are unavailable")
if (
signal_type is None
or not isinstance(pathway_packet_type, type)
or not callable(intent_to_capability)
or not callable(contrareactive_repulsion)
or not isinstance(swe_stages, tuple)
or not isinstance(capability_count, int)
or isinstance(capability_count, bool)
or capability_count < 1
):
raise RuntimeError("historical Resynthesis hop-signal type is unavailable")
if not all(
callable(getattr(owner, method, None))
for owner, method in (
(expert_bank, "apply_topk"),
(transfer, "_cap_weights_from_domain"),
(transfer, "_pressure_on_capabilities"),
(transfer, "donor_weights_for_pressure"),
(transfer, "build_none_pathway_packet"),
(transfer, "none_pathway_training_loss"),
(hop_router, "compute_hop_signal"),
(rbo_type, "_intent_label"),
(rbo_type, "_context_dag_training_loss"),
(rbo_type, "_identity_supervision_loss"),
(rbo_type, "_plan_layer_traversal"),
(rbo_type, "_comm_from_prior_delta"),
(rbo_type, "_correction_training_loss"),
(rbo_type, "layer_gate_scale"),
(rbo_type, "_joint_stop_logit"),
(rbo_type, "_none_step_phase"),
(rbo_type, "_recursive_pass"),
(rbo_type, "forward"),
(outcome_history, "_pack_step"),
(none_fabric_type, "form_commit"),
)
):
raise RuntimeError("historical Resynthesis hot-path method contract differs")
if not bool(
getattr(
none_fabric_type,
"_resynthesis_tensor_commit_installed",
False,
)
):
setattr(
none_fabric_type,
"_resynthesis_original_none_fabric_init",
getattr(none_fabric_type, "__init__"),
)
setattr(
none_fabric_type,
"_resynthesis_none_fabric_commit_type",
none_fabric_commit_type,
)
setattr(
none_fabric_type,
"__init__",
_historical_none_fabric_init_with_tensor_constants,
)
setattr(
none_fabric_type,
"form_commit",
_tensor_native_historical_none_fabric_form_commit,
)
setattr(
none_fabric_type,
"_resynthesis_tensor_commit_installed",
True,
)
setattr(expert_bank, "apply_topk", _tensor_native_expert_apply_topk)
setattr(
transfer, "_cap_weights_from_domain", _tensor_native_cap_weights_from_domain
)
setattr(
transfer,
"_pressure_on_capabilities",
_tensor_native_pressure_on_capabilities,
)
setattr(
transfer,
"donor_weights_for_pressure",
_tensor_native_donor_weights_for_pressure,
)
setattr(
transfer,
"none_pathway_training_loss",
_tensor_native_none_pathway_training_loss,
)
if not bool(
getattr(
transfer,
"_resynthesis_none_tensor_constants_installed",
False,
)
):
setattr(
transfer,
"_resynthesis_original_transfer_init",
getattr(transfer, "__init__"),
)
setattr(
transfer,
"__init__",
_historical_transfer_init_with_tensor_constants,
)
setattr(
transfer,
"_resynthesis_none_tensor_constants_installed",
True,
)
setattr(
transfer,
"_resynthesis_none_pathway_packet_type",
pathway_packet_type,
)
setattr(
transfer,
"build_none_pathway_packet",
_tensor_native_build_none_pathway_packet,
)
setattr(hop_router, "_resynthesis_hop_signal_type", signal_type)
setattr(hop_router, "compute_hop_signal", _tensor_native_compute_hop_signal)
setattr(
rbo_type,
"_comm_from_prior_delta",
staticmethod(_tensor_native_sequence_comm_from_prior_delta),
)
if not bool(
getattr(rbo_type, "_resynthesis_tensor_intent_mapping_installed", False)
):
setattr(
rbo_type,
"_resynthesis_original_rbo_init",
getattr(rbo_type, "__init__"),
)
setattr(
rbo_type,
"_resynthesis_intent_to_capability_boundary",
staticmethod(intent_to_capability),
)
setattr(
rbo_type,
"_resynthesis_intent_capability_count",
capability_count,
)
setattr(
rbo_type,
"_resynthesis_original_context_dag_training_loss",
getattr(rbo_type, "_context_dag_training_loss"),
)
setattr(
rbo_type,
"__init__",
_historical_rbo_init_with_tensor_intent_mapping,
)
setattr(
rbo_type,
"_intent_label",
_tensor_native_historical_intent_route,
)
setattr(
rbo_type,
"_none_step_phase",
_tensor_native_historical_none_step_phase,
)
setattr(
rbo_type,
"_context_dag_training_loss",
_historical_tensor_native_context_dag_training_loss,
)
setattr(
rbo_type,
"_identity_supervision_loss",
_historical_tensor_native_identity_supervision_loss,
)
setattr(
rbo_type,
"_resynthesis_tensor_intent_mapping_installed",
True,
)
if isinstance(correction_controller, type):
setattr(
correction_controller,
"mode_target",
staticmethod(_tensor_native_self_correction_mode_target),
)
if (
isinstance(domain_collapse_detector, type)
and callable(getattr(domain_collapse_detector, "forward", None))
):
setattr(
domain_collapse_detector,
"forward",
_tensor_native_domain_collapse_forward,
)
if not bool(
getattr(rbo_type, "_resynthesis_tensor_route_forward_installed", False)
):
original_recursive_pass = getattr(rbo_type, "_recursive_pass")
setattr(
rbo_type,
"_resynthesis_original_rbo_forward",
getattr(rbo_type, "forward"),
)
setattr(
rbo_type,
"_resynthesis_original_recursive_pass",
original_recursive_pass,
)
setattr(
rbo_type,
"_resynthesis_rbo_result_type",
rbo_result_type,
)
setattr(
rbo_type,
"_resynthesis_contrareactive_repulsion",
staticmethod(contrareactive_repulsion),
)
setattr(rbo_type, "_resynthesis_swe_stages", swe_stages)
setattr(
rbo_type,
"_recursive_pass",
_historical_recursive_pass_with_tensor_route(
original_recursive_pass
),
)
setattr(
rbo_type,
"forward",
_tensor_native_historical_rbo_forward,
)
setattr(
rbo_type,
"_resynthesis_tensor_route_forward_installed",
True,
)
setattr(
outcome_history,
"_pack_step",
_tensor_native_outcome_history_pack_step,
)
setattr(
rubric_bias_type,
"apply_bias",
_tensor_native_rubric_apply_bias,
)
if not bool(
getattr(rbo_type, "_resynthesis_batch_correction_loss_installed", False)
):
setattr(
rbo_type,
"_resynthesis_original_correction_training_loss",
getattr(rbo_type, "_correction_training_loss"),
)
setattr(
rbo_type,
"_correction_training_loss",
_historical_batch_safe_correction_training_loss,
)
setattr(rbo_type, "_resynthesis_batch_correction_loss_installed", True)
if not bool(getattr(rbo_type, "_resynthesis_arm_exhaustion_installed", False)):
setattr(
rbo_type,
"_resynthesis_original_plan_layer_traversal",
getattr(rbo_type, "_plan_layer_traversal"),
)
setattr(
rbo_type,
"_resynthesis_original_joint_stop_logit",
getattr(rbo_type, "_joint_stop_logit"),
)
setattr(
rbo_type,
"_resynthesis_original_layer_gate_scale",
getattr(rbo_type, "layer_gate_scale"),
)
setattr(
rbo_type,
"_plan_layer_traversal",
_historical_plan_layer_traversal_with_arm_exhaustion,
)
setattr(
rbo_type,
"_joint_stop_logit",
_historical_joint_stop_logit_with_arm_exhaustion,
)
setattr(
rbo_type,
"layer_gate_scale",
_historical_layer_gate_scale_with_frontier_blend,
)
setattr(rbo_type, "_resynthesis_arm_exhaustion_installed", True)
return HISTORICAL_TENSOR_NATIVE_HOTPATH_ID
@dataclass(frozen=True)
class LegacyRBOCapabilitySessionState:
"""Session-owned observable-outcome state for cold continuation."""
durable_outcome: torch.Tensor
shot_one_route: torch.Tensor
shot_two_route: torch.Tensor
last_route: torch.Tensor
awaiting_second_shot: torch.Tensor
durable_outcome_updates: torch.Tensor
@dataclass(frozen=True)
class LegacyRBOCapabilityEngagementPacket:
"""Tensor-only evidence from the preserved legacy capability bank."""
authority_trained: torch.Tensor
retention_passed: torch.Tensor
forward_id: torch.Tensor
forward_calls: torch.Tensor
engaged_calls: torch.Tensor
delta_l2: torch.Tensor
route_id: torch.Tensor
shot_one_route: torch.Tensor
shot_two_route: torch.Tensor
shot_one_route_id: torch.Tensor
shot_two_route_id: torch.Tensor
route_shift_l1: torch.Tensor
durable_outcome_updates: torch.Tensor
payload_tensor_count: torch.Tensor
def _capture_legacy_second_shot_route_(
*,
last_route: torch.Tensor,
shot_two_route: torch.Tensor,
awaiting_second_shot: torch.Tensor,
) -> None:
"""Capture a pending route without synchronizing the hot path to the host."""
if last_route.shape != shot_two_route.shape:
raise RuntimeError("legacy RBO second-shot route geometry differs")
if awaiting_second_shot.shape != torch.Size([]):
raise RuntimeError("legacy RBO second-shot pending state must be scalar")
shot_two_route.copy_(
torch.where(
awaiting_second_shot.to(
device=shot_two_route.device,
dtype=torch.bool,
),
last_route,
shot_two_route,
)
)
awaiting_second_shot.zero_()
class _FrozenLegacyRBOTensor(nn.Module):
"""One dehydrated parent tensor that remains immutable and CPU-resident."""
def __init__(self, value: torch.Tensor) -> None:
super().__init__()
frozen = value.detach().to(device="cpu").contiguous()
frozen.requires_grad_(False)
self.register_buffer("payload", frozen, persistent=True)
def _apply(self, fn: Any, recurse: bool = True) -> "_FrozenLegacyRBOTensor":
del fn, recurse
return self
class LegacyRBOCapabilityBank(nn.Module):
"""Pinned compatibility owner for the parent's embedded 553-tensor RBO.
The parent artifact contains an older RBO geometry that cannot be overlaid
onto the authoritative 1,043-tensor graph. This implementation is owned by
Resynthesis so continuation never imports a changing Resynthesis worktree. Its
tensor names, initialization, routing, outcome memory, and persistence match
the exact source generation recorded by
``LEGACY_RBO_CAPABILITY_PROVENANCE_SHA256``.
"""
schema = _RESYNTHESIS_PARENT_CAPABILITY_SCHEMA
durable_outcome_width = 10
_payloads: nn.ModuleList
capability_profiles: torch.Tensor
route_query: nn.Linear
outcome_query: nn.Linear
residual_projection: nn.Linear
fusion_logit: nn.Parameter
authority_trained: torch.Tensor
retention_passed: torch.Tensor
forward_id: torch.Tensor
forward_calls: torch.Tensor
engaged_calls: torch.Tensor
last_delta_l2: torch.Tensor
last_route: torch.Tensor
shot_one_route: torch.Tensor
shot_two_route: torch.Tensor
durable_outcome: torch.Tensor
awaiting_second_shot: torch.Tensor
durable_outcome_updates: torch.Tensor
def __init__(self, state: Mapping[str, torch.Tensor]) -> None:
super().__init__()
normalized = {
str(name).removeprefix(_LEGACY_PARENT_STATE_PREFIX): value
for name, value in state.items()
}
if len(normalized) != LEGACY_RBO_CAPABILITY_TENSOR_COUNT:
raise RuntimeError(
"legacy RBO capability tensor count differs: "
f"expected={LEGACY_RBO_CAPABILITY_TENSOR_COUNT} "
f"actual={len(normalized)}"
)
if len(normalized) != len(state) or any(
not isinstance(value, torch.Tensor) for value in normalized.values()
):
raise RuntimeError(
"legacy RBO capability state has duplicate or non-tensor entries"
)
names = tuple(sorted(normalized))
layer_indices = sorted(
{
int(parts[1])
for name in names
if (parts := name.split("."))[:1] == ["experts"]
and len(parts) > 3
and parts[1].isdigit()
and parts[2:] == [parts[2], parts[3]]
and parts[2] == "bank"
and parts[3] == "gate_up"
}
)
if layer_indices != list(range(len(layer_indices))) or not layer_indices:
raise RuntimeError("legacy RBO capability expert layers are not contiguous")
profile_rows: list[torch.Tensor] = []
hidden_size = 0
intermediate_size = 0
route_count = 0
profile_width = 0
profile_suffixes = (
"flat_expert_identity",
"flat_expert_what_profile",
"flat_expert_when_profile",
"flat_expert_how_profile",
"flat_expert_capability",
)
for layer_idx in layer_indices:
prefix = f"experts.{layer_idx}."
gate_name = prefix + "bank.gate_up"
down_name = prefix + "bank.down"
if gate_name not in normalized or down_name not in normalized:
raise RuntimeError("legacy RBO capability expert payload is incomplete")
gate = normalized[gate_name]
down = normalized[down_name]
if gate.dim() != 3 or down.dim() != 3:
raise RuntimeError(
"legacy RBO capability experts are not rank-three banks"
)
layer_routes = int(gate.shape[0])
layer_hidden = int(gate.shape[-1])
layer_intermediate = int(down.shape[-1])
if tuple(gate.shape) != (
layer_routes,
2 * layer_intermediate,
layer_hidden,
) or tuple(down.shape) != (
layer_routes,
layer_hidden,
layer_intermediate,
):
raise RuntimeError(
"legacy RBO capability expert geometry is inconsistent"
)
if layer_idx == 0:
hidden_size = layer_hidden
intermediate_size = layer_intermediate
route_count = layer_routes
elif (
layer_hidden != hidden_size
or layer_intermediate != intermediate_size
or layer_routes != route_count
):
raise RuntimeError(
"legacy RBO capability expert geometry drifts by layer"
)
profile_parts: list[torch.Tensor] = []
for suffix in profile_suffixes:
name = prefix + suffix
value = normalized.get(name)
if not isinstance(value, torch.Tensor) or value.dim() != 2:
raise RuntimeError(
f"legacy RBO capability profile is missing or invalid: {name}"
)
if int(value.shape[0]) != route_count:
raise RuntimeError(
"legacy RBO capability profile route count differs"
)
profile_parts.append(
value.detach().to(device="cpu", dtype=torch.float32)
)
layer_profile = torch.cat(profile_parts, dim=-1)
if layer_idx == 0:
profile_width = int(layer_profile.shape[-1])
elif int(layer_profile.shape[-1]) != profile_width:
raise RuntimeError(
"legacy RBO capability profile width drifts by layer"
)
profile_rows.append(layer_profile)
classifier = normalized.get("intent_module.classifier.weight")
feedback_down = normalized.get("feedback_head.down.weight")
if (
not isinstance(classifier, torch.Tensor)
or tuple(classifier.shape[1:]) != (hidden_size,)
or not isinstance(feedback_down, torch.Tensor)
or feedback_down.dim() != 2
or int(feedback_down.shape[1]) != hidden_size
):
raise RuntimeError(
"legacy RBO capability intent/feedback geometry is invalid"
)
self._payload_names = names
self._payloads = nn.ModuleList(
[_FrozenLegacyRBOTensor(normalized[name]) for name in names]
)
profiles = F.normalize(torch.cat(profile_rows, dim=0), dim=-1, eps=1e-6)
self.register_buffer("capability_profiles", profiles, persistent=False)
self.route_query = nn.Linear(hidden_size, profile_width, bias=False)
self.outcome_query = nn.Linear(
self.durable_outcome_width,
profile_width,
bias=False,
)
self.residual_projection = nn.Linear(profile_width, hidden_size, bias=False)
self.fusion_logit = nn.Parameter(torch.zeros(()))
signal_weight = normalized.get("feedback_head.signal.weight")
trauma_values = [
normalized.get(f"trauma_gate.{layer_idx}") for layer_idx in layer_indices
]
learned_query = torch.cat(
(
feedback_down.detach().to(dtype=torch.float32),
classifier.detach().to(dtype=torch.float32),
),
dim=0,
)
if tuple(learned_query.shape) != tuple(self.route_query.weight.shape):
raise RuntimeError(
"legacy RBO capability learned query/profile geometry differs"
)
if (
not isinstance(signal_weight, torch.Tensor)
or signal_weight.dim() != 2
or tuple(signal_weight.shape) != (4, int(feedback_down.shape[0]))
or any(
not isinstance(value, torch.Tensor) or value.numel() != 1
for value in trauma_values
)
):
raise RuntimeError("legacy RBO capability feedback/trauma state is invalid")
with torch.no_grad():
self.route_query.weight.copy_(learned_query)
self.residual_projection.weight.copy_(learned_query.transpose(0, 1))
self.outcome_query.weight.zero_()
self.outcome_query.weight[
: int(feedback_down.shape[0]),
: int(signal_weight.shape[0]),
].copy_(signal_weight.detach().to(dtype=torch.float32).transpose(0, 1))
self.fusion_logit.copy_(
torch.stack(
[
value.detach().to(dtype=torch.float32).reshape(())
for value in trauma_values
if isinstance(value, torch.Tensor)
]
).mean()
)
self.register_buffer("authority_trained", torch.zeros((), dtype=torch.bool))
self.register_buffer("retention_passed", torch.zeros((), dtype=torch.bool))
self.register_buffer(
"forward_id",
torch.zeros((), dtype=torch.long),
persistent=False,
)
self.register_buffer(
"forward_calls",
torch.zeros((), dtype=torch.long),
persistent=False,
)
self.register_buffer(
"engaged_calls",
torch.zeros((), dtype=torch.long),
persistent=False,
)
self.register_buffer(
"last_delta_l2",
torch.zeros((), dtype=torch.float32),
persistent=False,
)
self.register_buffer(
"last_route",
torch.zeros(profiles.shape[0]),
persistent=False,
)
self.register_buffer(
"shot_one_route",
torch.zeros_like(self.last_route),
persistent=False,
)
self.register_buffer(
"shot_two_route",
torch.zeros_like(self.last_route),
persistent=False,
)
self.register_buffer(
"durable_outcome",
torch.zeros(self.durable_outcome_width),
persistent=False,
)
self.register_buffer(
"awaiting_second_shot",
torch.zeros((), dtype=torch.bool),
persistent=False,
)
self.register_buffer(
"durable_outcome_updates",
torch.zeros((), dtype=torch.long),
persistent=False,
)
self.hidden_size = hidden_size
self.intermediate_size = intermediate_size
self.n_expert_layers = len(layer_indices)
self.routes_per_layer = route_count
self.n_intents = int(classifier.shape[0])
self.feedback_hidden_size = int(feedback_down.shape[0])
self.profile_width = profile_width
self.payload_key_set_sha256 = hashlib.sha256(
"\n".join(names).encode("utf-8")
).hexdigest()
geometry_rows = [
(name, tuple(normalized[name].shape), str(normalized[name].dtype))
for name in names
]
self.payload_geometry_sha256 = hashlib.sha256(
json.dumps(geometry_rows, separators=(",", ":")).encode("utf-8")
).hexdigest()
def trainable_parameters(self) -> Iterator[nn.Parameter]:
yield from self.route_query.parameters()
yield from self.outcome_query.parameters()
yield from self.residual_projection.parameters()
yield self.fusion_logit
def reset_authority_from_boundary(self) -> None:
with torch.no_grad():
self.authority_trained.zero_()
self.retention_passed.zero_()
def begin_forward(self, forward_id: torch.Tensor) -> None:
if forward_id.numel() != 1:
raise RuntimeError("legacy RBO capability forward ID must be scalar")
with torch.no_grad():
self.forward_id.copy_(
forward_id.detach()
.to(device=self.forward_id.device, dtype=self.forward_id.dtype)
.reshape(())
)
self.forward_calls.zero_()
self.engaged_calls.zero_()
self.last_delta_l2.zero_()
def mark_authority_trained_from_boundary(
self,
update_proven: torch.Tensor,
retention_proven: torch.Tensor,
) -> None:
if update_proven.numel() != 1 or retention_proven.numel() != 1:
raise RuntimeError(
"legacy RBO capability authority receipts must be scalar"
)
with torch.no_grad():
self.authority_trained.copy_(update_proven.to(dtype=torch.bool).reshape(()))
self.retention_passed.copy_(
retention_proven.to(dtype=torch.bool).reshape(())
)
def apply_durable_outcome_state(self, observation: torch.Tensor) -> None:
"""Apply a persisted observable outcome, never a target or gold answer."""
if observation.numel() != self.durable_outcome_width:
raise RuntimeError("legacy RBO durable outcome geometry differs")
with torch.no_grad():
self.shot_one_route.copy_(self.last_route)
self.durable_outcome.copy_(
observation.detach()
.to(
device=self.durable_outcome.device,
dtype=self.durable_outcome.dtype,
)
.reshape_as(self.durable_outcome)
)
self.awaiting_second_shot.fill_(True)
self.durable_outcome_updates.add_(
torch.ones_like(self.durable_outcome_updates)
)
def capture_session_state(self) -> LegacyRBOCapabilitySessionState:
return LegacyRBOCapabilitySessionState(
durable_outcome=self.durable_outcome.detach().clone(),
shot_one_route=self.shot_one_route.detach().clone(),
shot_two_route=self.shot_two_route.detach().clone(),
last_route=self.last_route.detach().clone(),
awaiting_second_shot=self.awaiting_second_shot.detach().clone(),
durable_outcome_updates=self.durable_outcome_updates.detach().clone(),
)
def reset_session_state(self) -> None:
with torch.no_grad():
self.durable_outcome.zero_()
self.shot_one_route.zero_()
self.shot_two_route.zero_()
self.last_route.zero_()
self.awaiting_second_shot.zero_()
self.durable_outcome_updates.zero_()
def hydrate_session_state(self, state: LegacyRBOCapabilitySessionState) -> None:
pairs = (
(self.durable_outcome, state.durable_outcome),
(self.shot_one_route, state.shot_one_route),
(self.shot_two_route, state.shot_two_route),
(self.last_route, state.last_route),
(self.awaiting_second_shot, state.awaiting_second_shot),
(self.durable_outcome_updates, state.durable_outcome_updates),
)
with torch.no_grad():
for target, source in pairs:
if tuple(target.shape) != tuple(source.shape):
raise RuntimeError("legacy RBO capability session geometry differs")
target.copy_(source.to(device=target.device, dtype=target.dtype))
def forward(self, hidden: torch.Tensor) -> torch.Tensor:
if hidden.shape[-1] != self.hidden_size:
raise RuntimeError("legacy RBO capability hidden geometry differs")
with torch.no_grad():
self.forward_calls.add_(torch.ones_like(self.forward_calls))
active = self.authority_trained & self.retention_passed
if not self.training and not bool(active.detach().to(device="cpu")):
self.last_delta_l2.zero_()
return hidden
query_dtype = self.route_query.weight.dtype
query_hidden = hidden.to(dtype=query_dtype)
query = self.route_query(query_hidden)
outcome = self.outcome_query(
self.durable_outcome.to(device=query.device, dtype=query_dtype)
)
query = F.normalize(query + outcome, dim=-1, eps=1e-6)
profiles = self.capability_profiles.to(
device=query.device,
dtype=query_dtype,
)
route_logits = torch.einsum("btd,rd->btr", query, profiles)
route_weights = route_logits.softmax(dim=-1)
capability = torch.einsum("btr,rd->btd", route_weights, profiles)
proposed_delta = (
self.residual_projection(capability) * self.fusion_logit.sigmoid()
)
route = route_weights.mean(dim=(0, 1)).detach()
with torch.no_grad():
self.last_route.copy_(route.to(device=self.last_route.device))
_capture_legacy_second_shot_route_(
last_route=self.last_route,
shot_two_route=self.shot_two_route,
awaiting_second_shot=self.awaiting_second_shot,
)
if self.training:
inactive_delta = proposed_delta - proposed_delta.detach()
else:
inactive_delta = torch.zeros_like(proposed_delta)
active_scale = active.to(device=hidden.device, dtype=hidden.dtype)
delta = active_scale * proposed_delta.to(dtype=hidden.dtype) + (
1.0 - active_scale
) * inactive_delta.to(dtype=hidden.dtype)
with torch.no_grad():
self.engaged_calls.add_(active.to(dtype=torch.long))
self.last_delta_l2.copy_(delta.detach().float().square().sum().sqrt())
output: torch.Tensor = hidden + delta
return output
def engagement_packet(self) -> LegacyRBOCapabilityEngagementPacket:
return LegacyRBOCapabilityEngagementPacket(
authority_trained=self.authority_trained.detach().clone(),
retention_passed=self.retention_passed.detach().clone(),
forward_id=self.forward_id.detach().clone(),
forward_calls=self.forward_calls.detach().clone(),
engaged_calls=self.engaged_calls.detach().clone(),
delta_l2=self.last_delta_l2.detach().clone(),
route_id=self.last_route.argmax().detach().clone(),
shot_one_route=self.shot_one_route.detach().clone(),
shot_two_route=self.shot_two_route.detach().clone(),
shot_one_route_id=self.shot_one_route.argmax().detach().clone(),
shot_two_route_id=self.shot_two_route.argmax().detach().clone(),
route_shift_l1=(self.shot_two_route - self.shot_one_route)
.abs()
.sum()
.detach()
.clone(),
durable_outcome_updates=self.durable_outcome_updates.detach().clone(),
payload_tensor_count=torch.full_like(
self.forward_calls,
len(self._payload_names),
),
)
def _verified_legacy_capability_provenance(cfg: ResynthesisConfig) -> str:
"""Return the launch's code-provenance observation for receipts only."""
observed = cfg.legacy_capability_source_sha256
return observed if observed else "unavailable"
def _resolve_path(base_dir: str, filename: str) -> Path:
p = Path(filename)
if p.is_absolute():
return p
return Path(base_dir) / filename
def load_resynthesis_parent_config(cfg: ResynthesisConfig) -> dict[str, Any]:
"""Load the Resynthesis config.json (boundary I/O — JSON is allowed here)."""
config_path = _resolve_path(cfg.base_model_dir, cfg.config_path)
with open(config_path, encoding="utf-8") as f:
return cast(dict[str, Any], json.load(f))
def load_resynthesis_parent_manifest(cfg: ResynthesisConfig) -> dict[str, Any]:
"""Load the integrated checkpoint manifest at the external I/O boundary."""
manifest_path = (
Path(cfg.parent_checkpoint_manifest_path).expanduser().resolve()
)
manifest_sha256 = _streamed_file_sha256(manifest_path)
if (
cfg.parent_checkpoint_manifest_sha256
!= RESYNTHESIS_PARENT_CHECKPOINT_MANIFEST_SHA256
or manifest_sha256 != cfg.parent_checkpoint_manifest_sha256
):
raise RuntimeError(
"Resynthesis parent checkpoint manifest SHA-256 differs"
)
with manifest_path.open(encoding="utf-8") as handle:
manifest = json.load(handle)
if not isinstance(manifest, dict):
raise RuntimeError("Resynthesis checkpoint manifest is not a JSON object")
expected_payload_sha256 = str(manifest.get("manifest_payload_sha256", ""))
payload = dict(manifest)
payload.pop("manifest_payload_sha256", None)
actual_payload_sha256 = hashlib.sha256(
json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
).hexdigest()
if expected_payload_sha256 != actual_payload_sha256:
raise RuntimeError("Resynthesis checkpoint manifest payload SHA-256 differs")
return manifest
def _streamed_file_sha256(path: Path) -> str:
"""Hash an immutable model artifact at the explicit checkpoint boundary."""
from resynthesis.none_migration import file_sha256_boundary
return file_sha256_boundary(path)
def _safetensors_parameter_elements(path: Path) -> int:
"""Count exact tensor elements from a safetensors header boundary."""
with path.open("rb") as handle:
header_width = int.from_bytes(handle.read(8), byteorder="little")
if header_width < 2:
raise RuntimeError("native parent safetensors header is malformed")
header = json.loads(handle.read(header_width).decode("utf-8"))
if not isinstance(header, dict):
raise RuntimeError("native parent safetensors header is not an object")
parameter_elements = 0
tensor_count = 0
for name, record in header.items():
if name == "__metadata__":
continue
if not isinstance(record, dict):
raise RuntimeError("native parent safetensors tensor record is malformed")
shape = record.get("shape")
if not isinstance(shape, list) or not all(
isinstance(width, int) and not isinstance(width, bool) and width >= 0
for width in shape
):
raise RuntimeError("native parent safetensors tensor shape is malformed")
elements = 1
for width in shape:
elements *= width
parameter_elements += elements
tensor_count += 1
if tensor_count < 1 or parameter_elements < 1:
raise RuntimeError("native parent safetensors contains no model tensors")
return parameter_elements
def _verified_resynthesis_native_parent_manifest(
cfg: ResynthesisConfig,
native_root: Path,
weights_path: Path,
parameter_elements: int,
inherited_manifest: Mapping[str, Any],
) -> tuple[str, bool]:
"""Verify exact parent bytes while allowing their public release location."""
manifest_path = Path(cfg.native_parent_manifest_path).expanduser().resolve()
manifest_sha256 = _streamed_file_sha256(manifest_path)
if (
cfg.native_parent_manifest_sha256 != RESYNTHESIS_NATIVE_PARENT_MANIFEST_SHA256
or manifest_sha256 != cfg.native_parent_manifest_sha256
or cfg.native_parent_model_sha256
!= RESYNTHESIS_NATIVE_PARENT_MODEL_SHA256
):
raise RuntimeError("Resynthesis native-parent release identity differs")
loaded = json.loads(manifest_path.read_text(encoding="utf-8"))
if not isinstance(loaded, dict):
raise RuntimeError("Resynthesis native-parent manifest is not an object")
model_artifact = loaded.get("modelArtifact")
inherited_lineage = loaded.get("inheritedCheckpointLineage")
source_runtime = loaded.get("sourceEraRuntime")
migration = loaded.get("migrationValidation")
inherited_model_artifact = inherited_manifest.get("model_artifact")
if not all(
isinstance(value, dict)
for value in (
model_artifact,
inherited_lineage,
source_runtime,
migration,
inherited_model_artifact,
)
):
raise RuntimeError("Resynthesis native-parent manifest lacks required records")
assert isinstance(model_artifact, dict)
assert isinstance(inherited_lineage, dict)
assert isinstance(source_runtime, dict)
assert isinstance(migration, dict)
assert isinstance(inherited_model_artifact, dict)
declared_root = Path(str(loaded.get("nativeRoot", ""))).resolve()
declared_model_path = Path(str(model_artifact.get("path", "")))
checks = (
native_root.is_dir(),
loaded.get("schema") == "nnf.resynthesis.native_parent_generation.v1",
loaded.get("owner") == "Resynthesis",
loaded.get("generation") == RESYNTHESIS_NATIVE_PARENT_GENERATION,
declared_root == Path(RESYNTHESIS_NATIVE_PARENT_ROOT).resolve(),
loaded.get("externalProductCheckpointDependency") is False,
declared_model_path == Path("model.safetensors"),
weights_path.name == declared_model_path.name,
model_artifact.get("sizeBytes") == weights_path.stat().st_size,
model_artifact.get("parameterElements") == parameter_elements,
model_artifact.get("sha256") == cfg.native_parent_model_sha256,
model_artifact.get("sha256")
== inherited_model_artifact.get("artifact_sha256"),
model_artifact.get("tensorCount")
== inherited_model_artifact.get("tensor_count"),
model_artifact.get("contentTensorCount")
== inherited_model_artifact.get("content_tensor_count"),
inherited_lineage.get("checkpointId")
== inherited_manifest.get("checkpoint_id"),
inherited_lineage.get("manifestPayloadSha256")
== inherited_manifest.get("manifest_payload_sha256"),
migration.get("independentModelSha256Recomputed") is True,
migration.get("safetensorsHeaderParameterCountVerified") is True,
migration.get("promotionEligible") is False,
)
if not all(checks):
raise RuntimeError(
"Resynthesis native-parent manifest differs from live geometry or lineage"
)
return manifest_sha256, bool(migration.get("promotionEligible"))
def _verified_parent_source_hashes(cfg: ResynthesisConfig) -> dict[str, str]:
"""Observe mutable parent source bytes without creating launch authority.
The historical modules themselves still have to import successfully when
the parent is loaded. Hashing them is only version telemetry, however, so
a concurrent edit or transient diagnostic read failure is recorded in the
observation instead of pre-empting that real import/load boundary.
"""
def observe(path: Path) -> str:
try:
return _streamed_file_sha256(path) if path.is_file() else "missing"
except (OSError, RuntimeError) as error:
return f"unavailable:{type(error).__name__}"
source_root = Path(cfg.parent_runtime_source_dir).expanduser().resolve()
fingerprint_path = Path(cfg.parent_source_fingerprint_path).expanduser().resolve()
runtime_files = tuple(
f"{_LEGACY_PARENT_PACKAGE_COMPONENT}/{filename}"
for filename in (
"additive_moe.py",
"capability_corpus.py",
"config.py",
"experts.py",
"glyph_vge.py",
"integrated.py",
"knowledge_transfer_surfaces.py",
"loader.py",
"master_context_orchestration.py",
"moe_growth_controller.py",
f"{_legacy_parent_module_name('rbo')}.py",
"none_fabric.py",
"rbo_checkpoint.py",
"rbo_outcome_continual.py",
"rbo_traversal_tensors.py",
"training_doctrine.py",
"swe_agent_system_prompt.txt",
"tool_schema.json",
)
)
outcome_source_name = (
f"{_LEGACY_PARENT_PACKAGE_COMPONENT}/rbo_outcome_continual.py"
)
actual: dict[str, str] = {}
for relative_path in runtime_files:
path = (
Path(cfg.parent_rbo_outcome_source_path).expanduser().resolve()
if relative_path == outcome_source_name
else source_root / relative_path
)
actual[relative_path] = observe(path)
actual["@source_fingerprint_shard"] = observe(fingerprint_path)
actual["@source_id"] = cfg.parent_source_id
return actual
def _source_bundle_sha256(source_hashes: Mapping[str, str]) -> str:
rows = tuple(
sorted((str(name), str(value)) for name, value in source_hashes.items())
)
return hashlib.sha256(
json.dumps(rows, separators=(",", ":")).encode("utf-8")
).hexdigest()
def _exec_historical_source_module(
module_key: str,
module_spec: importlib.machinery.ModuleSpec,
module: Any,
) -> None:
"""Execute verified historical source without writing into its artifact."""
if module_spec.loader is None:
raise RuntimeError("historical source module has no loader")
previous = sys.dont_write_bytecode
sys.dont_write_bytecode = True
try:
module_spec.loader.exec_module(module)
except BaseException:
sys.modules.pop(module_key, None)
raise
finally:
sys.dont_write_bytecode = previous
def _load_aliased_parent_module(
alias: str,
source_root: Path,
module_name: str,
) -> Any:
"""Import one Resynthesis source generation under an isolated package identity."""
package_dir = source_root / _LEGACY_PARENT_PACKAGE_COMPONENT
package_name = alias
package = sys.modules.get(package_name)
if package is None:
if not package_dir.is_dir():
raise RuntimeError("historical Resynthesis package directory is unavailable")
# The source-era package initializer imported unrelated mutable modules
# that were absent from the checkpoint fingerprint. Build an isolated
# namespace package instead, then import only manifest-verified modules.
package_spec = importlib.machinery.ModuleSpec(
package_name,
loader=None,
is_package=True,
)
package_spec.submodule_search_locations = [str(package_dir)]
package = importlib.util.module_from_spec(package_spec)
sys.modules[package_name] = package
module_key = f"{package_name}.{module_name}"
existing = sys.modules.get(module_key)
if existing is not None:
actual_file = Path(str(getattr(existing, "__file__", ""))).resolve()
expected_file = (package_dir / f"{module_name}.py").resolve()
if actual_file != expected_file:
raise RuntimeError("isolated Resynthesis module alias is already bound elsewhere")
return existing
module_path = package_dir / f"{module_name}.py"
source_loader = importlib.machinery.SourceFileLoader(module_key, str(module_path))
module_spec = importlib.util.spec_from_loader(module_key, source_loader)
if module_spec is None or module_spec.loader is None:
raise RuntimeError(
f"could not create Resynthesis module specification: {module_path}"
)
module = importlib.util.module_from_spec(module_spec)
sys.modules[module_key] = module
_exec_historical_source_module(module_key, module_spec, module)
return module
def _load_aliased_parent_module_from_path(
alias: str,
module_name: str,
module_path: Path,
) -> Any:
"""Install one recovered hash-exact module into an isolated package."""
module_key = f"{alias}.{module_name}"
existing = sys.modules.get(module_key)
if existing is not None:
if (
Path(str(getattr(existing, "__file__", ""))).resolve()
!= module_path.resolve()
):
raise RuntimeError(
"recovered Resynthesis module alias is already bound elsewhere"
)
return existing
source_loader = importlib.machinery.SourceFileLoader(module_key, str(module_path))
module_spec = importlib.util.spec_from_loader(module_key, source_loader)
if module_spec is None or module_spec.loader is None:
raise RuntimeError(
f"could not create recovered Resynthesis module spec: {module_path}"
)
module = importlib.util.module_from_spec(module_spec)
sys.modules[module_key] = module
_exec_historical_source_module(module_key, module_spec, module)
return module
class _FrozenWideIntent(nn.Module):
"""Exact eight-way trained intent geometry embedded in the graph parent."""
def __init__(self, hidden_size: int = 4096, carry_dim: int = 512) -> None:
super().__init__()
self.hidden_size = int(hidden_size)
self.carry_dim = int(carry_dim)
self.n_intents = 8
self.carry_down = nn.Linear(self.hidden_size, self.carry_dim, bias=False)
self.classifier = nn.Linear(self.carry_dim, self.n_intents)
self.requires_grad_(False)
def forward_intent(self, hidden: torch.Tensor) -> torch.Tensor:
carry = self.carry_down(hidden.to(dtype=self.carry_down.weight.dtype))
logits: torch.Tensor = self.classifier(carry)
return logits
def _split_embedded_legacy_rbo_state(
state: Mapping[str, torch.Tensor],
) -> tuple[dict[str, torch.Tensor], dict[str, torch.Tensor]]:
additive = {
name: value
for name, value in state.items()
if not name.startswith(_LEGACY_PARENT_STATE_PREFIX)
}
legacy = {
name.removeprefix(_LEGACY_PARENT_STATE_PREFIX): value
for name, value in state.items()
if name.startswith(_LEGACY_PARENT_STATE_PREFIX)
}
return additive, legacy
def _normalize_legacy_ladder_state(
state: dict[str, torch.Tensor],
) -> tuple[dict[str, torch.Tensor], bool]:
"""Normalize the one source-era nested ladder namespace."""
if not state:
return state, False
legacy = tuple(name.startswith("ladder.") for name in state)
if any(legacy) and not all(legacy):
raise RuntimeError("integrated Resynthesis ladder checkpoint mixes legacy namespaces")
if not all(legacy):
return state, False
normalized = {name.removeprefix("ladder."): value for name, value in state.items()}
if len(normalized) != len(state):
raise RuntimeError("integrated Resynthesis ladder namespace migration collided")
return normalized, True
def _validate_embedded_legacy_rbo_split(
source: Mapping[str, torch.Tensor],
additive_state: Mapping[str, torch.Tensor],
legacy_state: Mapping[str, torch.Tensor],
*,
expected_source_tensor_count: int,
) -> str:
"""Validate the one historical additive-embedded RBO capability schema."""
if len(source) != expected_source_tensor_count:
raise RuntimeError("integrated additive tensor count differs from its manifest")
if len(additive_state) + len(legacy_state) != len(source):
raise RuntimeError("embedded legacy RBO split lost or duplicated tensors")
if len(legacy_state) != 553:
raise RuntimeError(
"embedded legacy RBO capability tensor count is not the exact schema"
)
embedded_names = {
name.removeprefix(_LEGACY_PARENT_STATE_PREFIX)
for name in source
if name.startswith(_LEGACY_PARENT_STATE_PREFIX)
}
if embedded_names != set(legacy_state):
raise RuntimeError("embedded legacy RBO capability key ownership differs")
if not additive_state:
raise RuntimeError("embedded legacy RBO split left no additive expert tensors")
return hashlib.sha256("\n".join(sorted(legacy_state)).encode("utf-8")).hexdigest()
def _prepare_empty_legacy_capability_module_slot(model: nn.Module) -> None:
"""Remove Resynthesis's ``None`` placeholder so its official add_module can own it."""
name = _legacy_parent_attribute("legacy_rbo_capability_bank")
existing = getattr(model, name, None)
if existing is not None:
raise RuntimeError("legacy RBO capability module slot is already occupied")
if hasattr(model, name):
delattr(model, name)
def _load_additive_state_into_resident_owner_boundary(
load_state: Callable[..., None],
model: Any,
state: Mapping[str, torch.Tensor],
) -> None:
"""Strictly copy CPU checkpoint tensors once into resident parameters.
Historical Resynthesis constructs the additive owner beside the already-resident
parent before invoking this callback. Supplying that owner's CUDA device
asks the source-era loader to build a second device state map and then copy
it again through ``load_state_dict``. Omitting the staging device retains
the source tensors and lets PyTorch perform the single required copy into
each target parameter. Key/shape checking remains owned by the exact
historical loader; no checkpoint, dtype, or lineage authority changes.
"""
load_state(model, state, device=None)
def _load_resynthesis_parent_with_boundary_migration(
integrated_module: Any,
load_parent_full: Any,
checkpoint_dir: str,
*,
device: str,
expected_additive_tensor_count: int,
) -> tuple[Any, ResynthesisBoundaryMigrationReceipt]:
"""Run Resynthesis's loader with a temporary, exact legacy-key boundary adapter."""
original_prefixed_loader = getattr(
integrated_module, "_load_prefixed_tensors", None
)
ladder_prefix = getattr(integrated_module, "LADDER_PREFIX", None)
if not callable(original_prefixed_loader) or not isinstance(ladder_prefix, str):
raise RuntimeError(
"Resynthesis integrated runtime lacks its checkpoint tensor loader"
)
additive_module = importlib.import_module(
f"{_LEGACY_PARENT_PACKAGE_COMPONENT}.additive_moe"
)
original_additive_loader = getattr(
additive_module, "load_additive_state_dict", None
)
split_embedded = getattr(integrated_module, "split_embedded_legacy_rbo_state", None)
wire_legacy_bank = getattr(
integrated_module, "wire_legacy_rbo_capability_bank", None
)
if not all(
callable(value)
for value in (
original_additive_loader,
split_embedded,
wire_legacy_bank,
)
):
raise RuntimeError(
"Resynthesis integrated runtime lacks its legacy capability migration"
)
ladder_migration_applied = False
embedded_legacy_rbo_applied = False
embedded_legacy_rbo_tensor_count = 0
additive_tensor_count_after_split = 0
embedded_legacy_rbo_key_set_sha256 = ""
def compatible_prefixed_loader(
model_path: str | Path,
prefix: str,
target_device: str,
) -> dict[str, torch.Tensor]:
nonlocal ladder_migration_applied
state = original_prefixed_loader(model_path, prefix, target_device)
if not isinstance(state, dict) or not all(
isinstance(name, str) and isinstance(value, torch.Tensor)
for name, value in state.items()
):
raise RuntimeError(
"Resynthesis checkpoint tensor loader returned an invalid state"
)
if prefix == ladder_prefix:
state, migrated = _normalize_legacy_ladder_state(state)
ladder_migration_applied = ladder_migration_applied or migrated
return cast(dict[str, torch.Tensor], state)
def compatible_additive_loader(
model: Any,
state: Mapping[str, torch.Tensor],
*,
device: str | torch.device | None = None,
) -> None:
nonlocal embedded_legacy_rbo_applied
nonlocal embedded_legacy_rbo_tensor_count
nonlocal additive_tensor_count_after_split
nonlocal embedded_legacy_rbo_key_set_sha256
assert callable(original_additive_loader)
assert callable(split_embedded)
assert callable(wire_legacy_bank)
split = split_embedded(state)
additive_state = getattr(split, "additive_state", None)
legacy_state = getattr(split, "legacy_state", None)
if not isinstance(additive_state, Mapping) or not isinstance(
legacy_state, Mapping
):
raise RuntimeError(
"Resynthesis legacy capability splitter returned an invalid packet"
)
if not legacy_state:
original_additive_loader(model, state, device=device)
return
if embedded_legacy_rbo_applied:
raise RuntimeError(
"embedded legacy RBO capability migration ran more than once"
)
embedded_legacy_rbo_key_set_sha256 = _validate_embedded_legacy_rbo_split(
state,
additive_state,
legacy_state,
expected_source_tensor_count=expected_additive_tensor_count,
)
# ``wire_additive_moe`` has already constructed the additive modules
# on the selected device. Passing that device into the historical
# loader first materializes an entire second GPU-resident state map,
# then ``load_state_dict`` copies the same tensors again into those
# resident parameters. Preserve the exact key/shape validation while
# letting PyTorch perform the one required CPU-to-resident copy during
# strict state loading. This is a load-boundary optimization only;
# it does not alter parent bytes, routing, trainability, or lineage.
_load_additive_state_into_resident_owner_boundary(
original_additive_loader,
model,
additive_state,
)
if not isinstance(model, nn.Module):
raise RuntimeError("Resynthesis additive owner is not an nn.Module")
_prepare_empty_legacy_capability_module_slot(model)
bank = wire_legacy_bank(model, legacy_state)
if not isinstance(bank, nn.Module):
raise RuntimeError("embedded legacy RBO capability bank was not attached")
embedded_legacy_rbo_applied = True
embedded_legacy_rbo_tensor_count = len(legacy_state)
additive_tensor_count_after_split = len(additive_state)
setattr(integrated_module, "_load_prefixed_tensors", compatible_prefixed_loader)
setattr(additive_module, "load_additive_state_dict", compatible_additive_loader)
try:
loaded = load_parent_full(
checkpoint_dir,
device=device,
dtype=torch.bfloat16,
)
finally:
setattr(integrated_module, "_load_prefixed_tensors", original_prefixed_loader)
setattr(additive_module, "load_additive_state_dict", original_additive_loader)
return loaded, ResynthesisBoundaryMigrationReceipt(
ladder_namespace_applied=ladder_migration_applied,
embedded_legacy_rbo_applied=embedded_legacy_rbo_applied,
embedded_legacy_rbo_tensor_count=embedded_legacy_rbo_tensor_count,
additive_tensor_count_after_split=additive_tensor_count_after_split,
embedded_legacy_rbo_key_set_sha256=embedded_legacy_rbo_key_set_sha256,
)
@contextmanager
def _relocated_parent_checkpoint_view(
cfg: ResynthesisConfig,
) -> Iterator[Path]:
"""Expose separated release artifacts as one read-only checkpoint view.
The source-era constructor expects the model, tokenizer, and configuration
under one directory. Public releases keep the multi-gigabyte model in the
weights tree and lexical artifacts in the tokenizer tree. A temporary
symlink view preserves those single physical files; it is created only
after the exact manifests and model identity have been verified.
"""
native_root = Path(cfg.base_model_dir).expanduser().resolve()
weights_path = _resolve_path(cfg.base_model_dir, cfg.base_weights).resolve()
config_path = _resolve_path(cfg.base_model_dir, cfg.config_path).resolve()
tokenizer_path = _resolve_path(cfg.base_model_dir, cfg.tokenizer_path).resolve()
checkpoint_manifest_path = Path(
cfg.parent_checkpoint_manifest_path
).expanduser().resolve()
expected_paths = {
"model.safetensors": weights_path,
"config.json": config_path,
"tokenizer.json": tokenizer_path,
f"{_LEGACY_PARENT_PACKAGE_COMPONENT}_manifest.json": (
checkpoint_manifest_path
),
}
if all((native_root / name).resolve() == path for name, path in expected_paths.items()):
yield native_root
return
if not native_root.is_dir() or any(
not path.is_file() for path in expected_paths.values()
):
raise RuntimeError(
"relocated Resynthesis parent release is missing a required artifact"
)
with tempfile.TemporaryDirectory(
prefix="resynthesis-native-parent-view-"
) as temporary:
view = Path(temporary)
for source in native_root.iterdir():
if source.name in expected_paths:
continue
(view / source.name).symlink_to(
source,
target_is_directory=source.is_dir(),
)
for name, source in expected_paths.items():
(view / name).symlink_to(source)
yield view
def _load_exact_resynthesis_parent(
cfg: ResynthesisConfig,
*,
device: str,
expected_rbo_tensor_count: int,
expected_additive_tensor_count: int,
) -> tuple[Any, ResynthesisBoundaryMigrationReceipt, nn.Module, str]:
"""Load the July 13 graph parent with its exact source-era architecture.
The current rolling Resynthesis source has additional trained-authority tensors
and cannot strictly represent this artifact. This path imports the source
snapshot under an isolated package name, reconstructs the checkpoint's
eight-way wide-intent/21-layer RBO exactly, and rejects every missing,
extra, or shape-different authoritative tensor.
"""
source_root = Path(cfg.parent_runtime_source_dir).expanduser().resolve()
alias = "_resynthesis_parent_artifact_20260713"
_load_aliased_parent_module(alias, source_root, "config")
_load_aliased_parent_module_from_path(
alias,
"rbo_outcome_continual",
Path(cfg.parent_rbo_outcome_source_path).expanduser().resolve(),
)
integrated_module = _load_aliased_parent_module(alias, source_root, "integrated")
additive_module = _load_aliased_parent_module(alias, source_root, "additive_moe")
rbo_module = _load_aliased_parent_module(
alias,
source_root,
_legacy_parent_module_name("rbo"),
)
checkpoint_module = _load_aliased_parent_module(alias, source_root, "rbo_checkpoint")
experts_module = _load_aliased_parent_module(alias, source_root, "experts")
transfer_module = _load_aliased_parent_module(
alias,
source_root,
"knowledge_transfer_surfaces",
)
traversal_module = _load_aliased_parent_module(
alias,
source_root,
"rbo_traversal_tensors",
)
hotpath_id = _install_historical_tensor_native_hotpaths(
rbo_module,
experts_module,
transfer_module,
traversal_module,
)
if hotpath_id != HISTORICAL_TENSOR_NATIVE_HOTPATH_ID:
raise RuntimeError("historical Resynthesis tensor-native hot-path identity differs")
load_promoted = getattr(
integrated_module,
f"load_{_LEGACY_PARENT_PACKAGE_COMPONENT}_promoted",
None,
)
original_prefixed_loader = getattr(
integrated_module, "_load_prefixed_tensors", None
)
legacy_integrated_rbo_loader_name = (
f"_load_integrated_{_LEGACY_PARENT_PACKAGE_COMPONENT}_rbo"
)
original_rbo_loader = getattr(
integrated_module,
legacy_integrated_rbo_loader_name,
None,
)
original_additive_loader = getattr(
additive_module, "load_additive_state_dict", None
)
original_wire_additive_moe = getattr(additive_module, "wire_additive_moe", None)
additive_lm_head = getattr(additive_module, "additive_lm_head", None)
ladder_prefix = getattr(integrated_module, "LADDER_PREFIX", None)
rbo_prefix = getattr(
integrated_module,
_legacy_parent_upper_name("RBO_PREFIX"),
None,
)
build_parent_rbo = getattr(
rbo_module,
f"build_{_LEGACY_PARENT_PACKAGE_COMPONENT}_rbo",
None,
)
rbo_config_type = getattr(
rbo_module,
_legacy_parent_class_name("RBOConfig"),
None,
)
detect_layers = getattr(checkpoint_module, "detect_expert_layers", None)
detect_intermediate = getattr(checkpoint_module, "detect_moe_intermediate", None)
wire_parent_rbo = getattr(
additive_module,
f"wire_{_LEGACY_PARENT_PACKAGE_COMPONENT}_rbo",
None,
)
required = (
load_promoted,
original_prefixed_loader,
original_rbo_loader,
original_additive_loader,
original_wire_additive_moe,
additive_lm_head,
build_parent_rbo,
rbo_config_type,
detect_layers,
detect_intermediate,
wire_parent_rbo,
)
if not all(callable(value) for value in required):
raise RuntimeError("historical Resynthesis runtime lacks an exact graph constructor")
if not isinstance(ladder_prefix, str) or not isinstance(rbo_prefix, str):
raise RuntimeError("historical Resynthesis runtime prefixes are unavailable")
ladder_migration_applied = False
authoritative_rbo_tensor_count = 0
authoritative_rbo_key_set_sha256 = ""
additive_tensor_count_after_split = 0
embedded_legacy_rbo_key_set_sha256 = ""
legacy_state: dict[str, torch.Tensor] = {}
skipped_additive_initializer_targets: list[torch.Tensor] = []
additive_wire_invocations = 0
def compatible_prefixed_loader(
model_path: str | Path,
prefix: str,
target_device: str,
) -> dict[str, torch.Tensor]:
nonlocal ladder_migration_applied
assert callable(original_prefixed_loader)
state = original_prefixed_loader(model_path, prefix, target_device)
if not isinstance(state, dict) or not all(
isinstance(name, str) and isinstance(value, torch.Tensor)
for name, value in state.items()
):
raise RuntimeError("historical Resynthesis tensor loader returned invalid state")
if prefix == ladder_prefix:
state, migrated = _normalize_legacy_ladder_state(state)
ladder_migration_applied = ladder_migration_applied or migrated
return cast(dict[str, torch.Tensor], state)
def exact_rbo_loader(
model: Any,
*,
model_path: Path,
cfg: Any,
device: str,
dtype: torch.dtype,
) -> nn.Module:
nonlocal authoritative_rbo_tensor_count
nonlocal authoritative_rbo_key_set_sha256
state = compatible_prefixed_loader(model_path, rbo_prefix, "cpu")
if len(state) != expected_rbo_tensor_count:
raise RuntimeError(
"historical authoritative RBO tensor count differs: "
f"expected={expected_rbo_tensor_count} actual={len(state)}"
)
if not any(name.startswith("none_fabric.") for name in state):
raise RuntimeError(
"historical authoritative RBO has no model-owned NoNE Fabric"
)
assert callable(detect_intermediate)
assert callable(detect_layers)
assert callable(rbo_config_type)
assert callable(build_parent_rbo)
cfg.moe_intermediate_size = detect_intermediate(state)
rbo_cfg = rbo_config_type(
hidden_size=int(cfg.hidden_size),
n_intents=8,
n_expert_layers=detect_layers(state),
feedback_hidden_size=128,
)
intent = _FrozenWideIntent(hidden_size=int(cfg.hidden_size), carry_dim=512)
def construct_exact_rbo() -> nn.Module:
exact_rbo = cast(
nn.Module,
build_parent_rbo(cfg, rbo_cfg, intent_module=intent),
)
exact_rbo.register_buffer(
"_resynthesis_recursive_arm_exhausted",
torch.zeros((), dtype=torch.bool),
persistent=False,
)
return exact_rbo
rbo = _build_exact_historical_rbo_from_state(
construct_exact_rbo,
state,
device=device,
dtype=dtype,
)
assert callable(wire_parent_rbo)
wire_parent_rbo(model, rbo)
rbo.requires_grad_(False)
rbo.eval()
model.resynthesis_rbo = rbo
model.resynthesis_rbo_packaging = "integrated_exact_historical_source"
model.resynthesis_none_architecture_active = True
authoritative_rbo_tensor_count = len(state)
authoritative_rbo_key_set_sha256 = hashlib.sha256(
"\n".join(sorted(state)).encode("utf-8")
).hexdigest()
return rbo
def compatible_additive_loader(
model: Any,
state: Mapping[str, torch.Tensor],
*,
device: str | torch.device | None = None,
) -> None:
nonlocal additive_tensor_count_after_split
nonlocal embedded_legacy_rbo_key_set_sha256
nonlocal legacy_state
assert callable(original_additive_loader)
additive_state, captured_legacy = _split_embedded_legacy_rbo_state(state)
embedded_legacy_rbo_key_set_sha256 = _validate_embedded_legacy_rbo_split(
state,
additive_state,
captured_legacy,
expected_source_tensor_count=expected_additive_tensor_count,
)
if additive_wire_invocations != 1:
raise RuntimeError(
"historical additive checkpoint load did not follow one exact wire"
)
assert callable(additive_lm_head)
head = additive_lm_head(model)
if not isinstance(head, nn.Module):
raise RuntimeError("historical additive loader found no additive head")
_validate_historical_additive_initializer_targets(
head,
additive_state,
skipped_additive_initializer_targets,
)
_load_additive_state_into_resident_owner_boundary(
original_additive_loader,
model,
additive_state,
)
additive_tensor_count_after_split = len(additive_state)
legacy_state = captured_legacy
def compatible_wire_additive_moe(
model: Any,
cfg: Any,
n_layers: int = 2,
) -> Any:
nonlocal additive_wire_invocations
additive_wire_invocations += 1
if additive_wire_invocations != 1:
raise RuntimeError("historical additive owner was wired more than once")
assert callable(original_wire_additive_moe)
assert callable(additive_lm_head)
return _wire_exact_historical_additive_without_overwritten_initializers(
original_wire_additive_moe,
additive_lm_head,
model,
cfg,
n_layers=n_layers,
skipped_persistent_targets=skipped_additive_initializer_targets,
)
setattr(integrated_module, "_load_prefixed_tensors", compatible_prefixed_loader)
setattr(integrated_module, legacy_integrated_rbo_loader_name, exact_rbo_loader)
setattr(additive_module, "load_additive_state_dict", compatible_additive_loader)
setattr(additive_module, "wire_additive_moe", compatible_wire_additive_moe)
try:
assert callable(load_promoted)
with _relocated_parent_checkpoint_view(cfg) as checkpoint_view:
loaded = _call_without_transformers_allocator_warmup(
load_promoted,
str(checkpoint_view),
device=device,
dtype=torch.bfloat16,
)
finally:
setattr(additive_module, "wire_additive_moe", original_wire_additive_moe)
setattr(integrated_module, "_load_prefixed_tensors", original_prefixed_loader)
setattr(
integrated_module,
legacy_integrated_rbo_loader_name,
original_rbo_loader,
)
setattr(additive_module, "load_additive_state_dict", original_additive_loader)
runtime = loaded[0] if isinstance(loaded, tuple) and loaded else None
if not isinstance(runtime, nn.Module):
raise RuntimeError("historical Resynthesis loader returned no runtime module")
vocabulary_graph = loaded[1] if isinstance(loaded, tuple) and len(loaded) > 1 else None
_bind_resynthesis_parent_runtime_facade(
runtime,
vocabulary_graph=vocabulary_graph,
)
final_hidden_hotpath_id = _install_parent_final_hidden_only_forward_boundary(
runtime
)
if final_hidden_hotpath_id != PARENT_FINAL_HIDDEN_ONLY_FORWARD_ID:
raise RuntimeError("historical parent final-hidden hot-path identity differs")
if len(legacy_state) != LEGACY_RBO_CAPABILITY_TENSOR_COUNT:
raise RuntimeError(
"historical embedded legacy capability payload was not captured"
)
capability_source_sha256 = _verified_legacy_capability_provenance(cfg)
bank = LegacyRBOCapabilityBank(legacy_state)
receipt = ResynthesisBoundaryMigrationReceipt(
ladder_namespace_applied=ladder_migration_applied,
embedded_legacy_rbo_applied=True,
embedded_legacy_rbo_tensor_count=len(legacy_state),
additive_tensor_count_after_split=additive_tensor_count_after_split,
embedded_legacy_rbo_key_set_sha256=embedded_legacy_rbo_key_set_sha256,
authoritative_rbo_tensor_count=authoritative_rbo_tensor_count,
authoritative_rbo_key_set_sha256=authoritative_rbo_key_set_sha256,
exact_historical_source_loaded=True,
)
return loaded, receipt, bank, capability_source_sha256
def resynthesis_parent_info(cfg: ResynthesisConfig) -> ResynthesisParentInfo:
"""Extract typed base info without loading weights (cheap, for validation)."""
raw = load_resynthesis_parent_config(cfg)
manifest = load_resynthesis_parent_manifest(cfg)
parent_source_hashes = _verified_parent_source_hashes(cfg)
observed_parent_source_bundle_sha256 = _source_bundle_sha256(
parent_source_hashes
)
parent_source_bundle_sha256 = cfg.parent_source_bundle_sha256
capability_source_sha256 = _verified_legacy_capability_provenance(cfg)
text_cfg = raw.get("text_config", raw)
none_rbo = manifest.get("none_rbo")
additive_moe = manifest.get("additive_moe")
native_decode_confidence = manifest.get("native_decode_confidence")
model_artifact = manifest.get("model_artifact")
if (
not isinstance(none_rbo, dict)
or not isinstance(additive_moe, dict)
or not isinstance(native_decode_confidence, dict)
or not isinstance(model_artifact, dict)
):
raise RuntimeError(
"Resynthesis checkpoint lacks integrated NoNE/RBO, adaptive experts, or native confidence"
)
artifact_composition = manifest.get("composition")
artifact_checkpoint_id = manifest.get("checkpoint_id")
if (
artifact_composition != _LEGACY_PARENT_COMPOSITION
or not isinstance(artifact_checkpoint_id, str)
or not artifact_checkpoint_id
):
raise RuntimeError(
"Resynthesis parent artifact identity differs from its sealed lineage"
)
weights_path = _resolve_path(cfg.base_model_dir, cfg.base_weights)
resolved_native_root = Path(cfg.base_model_dir).expanduser().resolve()
parameter_elements = _safetensors_parameter_elements(weights_path)
native_manifest_sha256, migration_promotion_eligible = (
_verified_resynthesis_native_parent_manifest(
cfg,
resolved_native_root,
weights_path,
parameter_elements,
manifest,
)
)
return ResynthesisParentInfo(
hidden_size=int(
text_cfg.get("hidden_size", RESYNTHESIS_HIDDEN_SIZE)
),
num_hidden_layers=int(
text_cfg.get(
"num_hidden_layers", RESYNTHESIS_NUM_HIDDEN_LAYERS
)
),
vocab_size=int(
text_cfg.get("vocab_size", RESYNTHESIS_PROJECTION_VOCAB_SIZE)
),
max_position_embeddings=max(
int(
text_cfg.get(
"max_position_embeddings",
RESYNTHESIS_MAX_POSITION_EMBEDDINGS,
)
),
RESYNTHESIS_MAX_POSITION_EMBEDDINGS,
),
num_attention_heads=int(
text_cfg.get(
"num_attention_heads", RESYNTHESIS_NUM_ATTENTION_HEADS
)
),
num_key_value_heads=int(
text_cfg.get(
"num_key_value_heads",
RESYNTHESIS_NUM_KEY_VALUE_HEADS,
)
),
intermediate_size=int(
text_cfg.get(
"intermediate_size", RESYNTHESIS_INTERMEDIATE_SIZE
)
),
head_dim=int(text_cfg.get("head_dim", RESYNTHESIS_HEAD_DIM)),
weights_path=str(weights_path),
config_path=str(_resolve_path(cfg.base_model_dir, cfg.config_path)),
baseline_frozen=True,
tie_word_embeddings=bool(raw.get("tie_word_embeddings", False)),
model_type=_RESYNTHESIS_PARENT_MODEL_TYPE,
composition=_RESYNTHESIS_PARENT_COMPOSITION,
integrated_rbo_tensors=int(none_rbo.get("n_tensors", 0)),
integrated_additive_tensors=int(additive_moe.get("n_tensors", 0)),
integrated_native_decode_confidence_tensors=int(
native_decode_confidence.get("n_tensors", 0)
),
checkpoint_id=(
"resynthesis-native-parent:"
f"{str(model_artifact.get('artifact_sha256', ''))}"
),
manifest_payload_sha256=str(manifest.get("manifest_payload_sha256", "")),
model_artifact_sha256=str(model_artifact.get("artifact_sha256", "")),
parameter_elements=parameter_elements,
native_owner="Resynthesis",
native_generation=RESYNTHESIS_NATIVE_PARENT_GENERATION,
native_root=str(resolved_native_root),
native_manifest_path=str(
Path(cfg.native_parent_manifest_path).expanduser().resolve()
),
native_manifest_sha256=native_manifest_sha256,
native_migration_promotion_eligible=(migration_promotion_eligible),
parent_source_bundle_sha256=parent_source_bundle_sha256,
observed_parent_source_bundle_sha256=(
observed_parent_source_bundle_sha256
),
parent_source_bundle_matches_expected=(
observed_parent_source_bundle_sha256
== parent_source_bundle_sha256
),
legacy_capability_source_sha256=capability_source_sha256,
historical_inherited_checkpoint_id=artifact_checkpoint_id,
historical_inherited_composition=artifact_composition,
historical_inherited_model_type=str(
raw.get("model_type", text_cfg.get("model_type", ""))
),
)
def _build_native_hybrid_context_cache(runtime: nn.Module) -> Any:
"""Build the parent's bounded hot-KV plus global recurrent cache.
This is a storage-boundary adapter for the immutable parent architecture.
It does not select an answer or route: all input tokens still execute every
decoder layer. The 24 trained linear-attention layers retain their global
recurrent state, while the eight full-attention layers retain one native
Dual-Chunk local window of hot K/V.
"""
runtime_config = getattr(runtime, "config", None)
get_text_config = getattr(runtime_config, "get_text_config", None)
if not callable(get_text_config):
raise RuntimeError(
"Resynthesis runtime exposes no text configuration for native hybrid K/V"
)
text_config = get_text_config(decoder=True)
layer_types = getattr(text_config, "layer_types", None)
if (
not isinstance(layer_types, (list, tuple))
or len(layer_types) != RESYNTHESIS_NUM_HIDDEN_LAYERS
or not all(isinstance(layer_type, str) for layer_type in layer_types)
):
raise RuntimeError("Resynthesis native hybrid K/V layer geometry differs")
linear_layers = sum(layer_type == "linear_attention" for layer_type in layer_types)
full_layers = sum(layer_type == "full_attention" for layer_type in layer_types)
if (
linear_layers != RESYNTHESIS_LINEAR_ATTENTION_LAYERS
or full_layers != RESYNTHESIS_FULL_ATTENTION_LAYERS
or linear_layers + full_layers != len(layer_types)
):
raise RuntimeError(
"Resynthesis native hybrid K/V does not match the trained parent topology"
)
from transformers.cache_utils import (
Cache,
DynamicSlidingWindowLayer,
LinearAttentionLayer,
)
cache_layers: list[Any] = []
for layer_type in layer_types:
if layer_type == "linear_attention":
cache_layers.append(cast(Any, LinearAttentionLayer)())
else:
cache_layers.append(
DynamicSlidingWindowLayer(
sliding_window=RESYNTHESIS_NATIVE_HOT_KV_TOKENS
)
)
cache = Cache(layers=cache_layers)
setattr(cache, "_resynthesis_native_hybrid_context", True)
setattr(
cache,
"_resynthesis_recurrent_layer_count",
RESYNTHESIS_LINEAR_ATTENTION_LAYERS,
)
setattr(
cache,
"_resynthesis_full_attention_layer_count",
RESYNTHESIS_FULL_ATTENTION_LAYERS,
)
setattr(
cache,
"_resynthesis_hot_window_tokens",
RESYNTHESIS_NATIVE_HOT_KV_TOKENS,
)
return cache
def _expand_immutable_native_hybrid_context_cache_boundary(
master_cache: object,
batch_indices_t: torch.Tensor,
) -> Any:
"""Clone one sealed parent prefix cache and expand only its isolated copy.
The parent prefix is immutable shared evidence. ``batch_indices_t`` is a
tensor-native all-zero reorder index: each output row selects the one
authoritative prefix row. Transformers does not implement
``batch_repeat_interleave`` for ``LinearAttentionLayer``, while
``reorder_cache`` supports both that recurrent layer and
``DynamicSlidingWindowLayer``. Deep-copying before the reorder preserves
the master tensors and every cache-layer bookkeeping field.
"""
from transformers.cache_utils import (
Cache,
DynamicSlidingWindowLayer,
LinearAttentionLayer,
)
if (
not isinstance(master_cache, Cache)
or getattr(master_cache, "_resynthesis_native_hybrid_context", False)
is not True
):
raise RuntimeError(
"immutable prefix expansion requires the native hybrid context cache"
)
if (
batch_indices_t.ndim != 1
or batch_indices_t.dtype != torch.long
or batch_indices_t.numel() < 1
):
raise RuntimeError(
"immutable prefix expansion requires a nonempty rank-one long index"
)
torch._assert_async(
batch_indices_t.eq(0).all(),
"immutable prefix expansion index must select only the sealed prefix row",
)
layers = master_cache.layers
if (
not isinstance(layers, list)
or len(layers) != RESYNTHESIS_NUM_HIDDEN_LAYERS
or sum(isinstance(layer, LinearAttentionLayer) for layer in layers)
!= RESYNTHESIS_LINEAR_ATTENTION_LAYERS
or sum(isinstance(layer, DynamicSlidingWindowLayer) for layer in layers)
!= RESYNTHESIS_FULL_ATTENTION_LAYERS
):
raise RuntimeError("immutable prefix cache topology differs")
for layer in layers:
if isinstance(layer, LinearAttentionLayer):
for state_t in (layer.conv_states, layer.recurrent_states):
if state_t is not None and (
state_t.ndim < 1 or state_t.shape[0] != 1
):
raise RuntimeError(
"immutable recurrent prefix cache must contain one row"
)
elif isinstance(layer, DynamicSlidingWindowLayer):
for state_t in (layer.keys, layer.values):
if state_t is not None and (
state_t.ndim < 1 or state_t.shape[0] != 1
):
raise RuntimeError(
"immutable sliding-window prefix cache must contain one row"
)
expanded_cache = copy.deepcopy(master_cache)
expanded_cache.reorder_cache(cast(torch.LongTensor, batch_indices_t))
return expanded_cache
def _native_context_cache_telemetry_boundary(
cache: object,
*,
device: torch.device,
) -> NativeContextCacheTelemetry:
"""Observe cache storage geometry without influencing model decisions."""
packet = NativeContextCacheTelemetry(
hybrid_active=torch.empty((), device=device, dtype=torch.bool),
total_positions=torch.empty((), device=device, dtype=torch.long),
hot_resident_positions=torch.empty((), device=device, dtype=torch.long),
hot_window_tokens=torch.empty((), device=device, dtype=torch.long),
recurrent_layer_count=torch.empty((), device=device, dtype=torch.long),
full_attention_layer_count=torch.empty(
(),
device=device,
dtype=torch.long,
),
)
_update_native_context_cache_telemetry_boundary(cache, packet=packet)
return packet
def _update_native_context_cache_telemetry_boundary(
cache: object,
*,
packet: NativeContextCacheTelemetry,
) -> None:
"""Fill preallocated cache telemetry tensors without per-wave allocation."""
hybrid_active = bool(getattr(cache, "_resynthesis_native_hybrid_context", False))
recurrent_layers = int(getattr(cache, "_resynthesis_recurrent_layer_count", 0))
full_layers = int(getattr(cache, "_resynthesis_full_attention_layer_count", 0))
hot_window = int(getattr(cache, "_resynthesis_hot_window_tokens", 0))
total_positions = 0
get_seq_length = getattr(cache, "get_seq_length", None)
if callable(get_seq_length):
total_positions = int(get_seq_length())
elif isinstance(getattr(cache, "positions", None), int):
total_positions = int(getattr(cache, "positions"))
hot_resident_positions = 0
layers = getattr(cache, "layers", None)
if isinstance(layers, list):
for layer in layers:
keys = getattr(layer, "keys", None)
if isinstance(keys, torch.Tensor) and keys.ndim >= 2:
hot_resident_positions = max(
hot_resident_positions,
int(keys.shape[-2]),
)
elif hybrid_active:
hot_resident_positions = min(total_positions, hot_window)
with torch.no_grad():
packet.hybrid_active.fill_(hybrid_active)
packet.total_positions.fill_(total_positions)
packet.hot_resident_positions.fill_(hot_resident_positions)
packet.hot_window_tokens.fill_(hot_window)
packet.recurrent_layer_count.fill_(recurrent_layers)
packet.full_attention_layer_count.fill_(full_layers)
class ResynthesisNativeParent(nn.Module):
"""Frozen Resynthesis-owned parent exposing trained hidden/logit surfaces.
The base weights are loaded and frozen (requires_grad=False). The forward
pass produces the final hidden states [batch, seq, hidden_size] that the
additive science stack + RBO operate on.
For training, the base is a feature extractor: gradients flow only through
the additive science layers, not the frozen base (Pillar: additive-only,
baseline-frozen).
"""
_base_loaded_marker: torch.Tensor
_decode_cached_positions: torch.Tensor
_native_context_full_attention_layer_count: torch.Tensor
_native_context_hot_resident_positions: torch.Tensor
_native_context_hot_window_tokens: torch.Tensor
_native_context_hybrid_active: torch.Tensor
_native_context_recurrent_layer_count: torch.Tensor
_native_context_total_positions: torch.Tensor
_parent_inference_session_buffers_normalized: torch.Tensor
_training_low_rank_projection_cache: torch.Tensor | None
_training_low_rank_projection_source_id: int | None
_training_low_rank_projection_source_version: int | None
_last_frozen_backbone_final_hidden_t: torch.Tensor | None
_last_frozen_backbone_summary_hidden_t: torch.Tensor | None
_last_frozen_backbone_input_positions_t: torch.Tensor | None
def __init__(
self, cfg: ResynthesisConfig, *, device: torch.device | str = "cpu"
) -> None:
super().__init__()
self.cfg = cfg
self.device = torch.device(device)
self.info = resynthesis_parent_info(cfg)
self._weights_loaded = False
self.runtime: nn.Module | None = None
self.tokenizer: Any | None = None
self.tokenizer_backend_identity: ResynthesisTokenizerIdentity | None = None
self._last_forward: ResynthesisParentForward | None = None
self._decode_past_key_values: Any | None = None
self._decode_authority_active = False
self._parent_session_may_hold_inference_tensors = False
self._parent_session_buffers_verified_normal = False
self._training_low_rank_projection_cache = None
self._training_low_rank_projection_source_id = None
self._training_low_rank_projection_source_version = None
self._last_frozen_backbone_final_hidden_t = None
self._last_frozen_backbone_summary_hidden_t = None
self._last_frozen_backbone_input_positions_t = None
self.verified_model_artifact_sha256 = ""
self.ladder_namespace_migration_applied = False
self.embedded_legacy_rbo_migration_applied = False
self.embedded_legacy_rbo_tensor_count = 0
self.additive_tensor_count_after_legacy_split = 0
self.embedded_legacy_rbo_key_set_sha256 = ""
self.authoritative_rbo_tensor_count = 0
self.authoritative_rbo_key_set_sha256 = ""
self.exact_historical_source_loaded = False
self.historical_tensor_native_hotpath_id = ""
self.historical_recursive_arm_exhaustion_id = ""
self.legacy_capability_source_sha256 = ""
object.__setattr__(self, "_pending_legacy_capability_bank", None)
self.register_buffer(
"_base_loaded_marker", torch.tensor([0], dtype=torch.int8), persistent=False
)
self.register_buffer(
"_decode_cached_positions",
torch.zeros((), dtype=torch.long),
persistent=False,
)
self.register_buffer(
"_parent_inference_session_buffers_normalized",
torch.zeros((), dtype=torch.long),
persistent=False,
)
self.register_buffer(
"_native_context_hybrid_active",
torch.zeros((), dtype=torch.bool),
persistent=False,
)
for buffer_name in (
"_native_context_total_positions",
"_native_context_hot_resident_positions",
"_native_context_hot_window_tokens",
"_native_context_recurrent_layer_count",
"_native_context_full_attention_layer_count",
):
self.register_buffer(
buffer_name,
torch.zeros((), dtype=torch.long),
persistent=False,
)
def load_weights(self) -> "ResynthesisNativeParent":
"""Load the complete native backbone + adaptive MoE + NoNE/RBO graph.
The previous loader registered checkpoint tensors as inert buffers and
reduced the base forward to an embedding lookup. Loading now goes
through Resynthesis's integrated constructor directly in this process, making
its trained decoder, adaptive experts, NoNE Fabric, RBO traversal, VGE,
and native head children of this model graph.
"""
if self._weights_loaded:
return self
weights_path = Path(self.info.weights_path)
if not weights_path.is_file():
raise FileNotFoundError(
f"Resynthesis base weights not found: {weights_path}. "
"Set ResynthesisConfig.base_model_dir to the integrated Resynthesis checkpoint directory."
)
if self.info.composition != _RESYNTHESIS_PARENT_COMPOSITION:
raise RuntimeError(
"Resynthesis requires Resynthesis's one-model NoNE/RBO composition"
)
if (
self.info.integrated_rbo_tensors <= 0
or self.info.integrated_additive_tensors <= 0
or self.info.integrated_native_decode_confidence_tensors <= 0
):
raise RuntimeError(
"Resynthesis requires trained integrated RBO, adaptive-expert, and native-confidence tensors"
)
fastokens_dependency = activate_resynthesis_fastokens(self.cfg)
from resynthesis.none_paging import file_sha256_authority_boundary
parent_sha256_cache_root = (
Path(__file__).resolve().parents[1]
/ ".nnf-resynthesis"
/ "artifact_sha256_identity_cache"
/ RESYNTHESIS_NATIVE_PARENT_GENERATION
)
preload_artifact_sha256 = file_sha256_authority_boundary(
weights_path,
expected_sha256=self.info.model_artifact_sha256,
identity_cache_root=parent_sha256_cache_root,
)
loaded, migration, legacy_bank, capability_source_sha256 = (
_load_exact_resynthesis_parent(
self.cfg,
device=str(self.device),
expected_rbo_tensor_count=self.info.integrated_rbo_tensors,
expected_additive_tensor_count=self.info.integrated_additive_tensors,
)
)
runtime, vocabulary_graph, _bank, tokenizer, _ladder = loaded
loaded_tokenizer_identity = tokenizer_identity(
tokenizer,
dependency=fastokens_dependency,
tokenizer_path=_resolve_path(
self.cfg.base_model_dir,
self.cfg.tokenizer_path,
),
inherited_tokenizer_path=_resolve_path(
self.cfg.base_model_dir,
self.cfg.tokenizer_path,
),
)
setattr(
tokenizer,
"_resynthesis_tokenizer_identity",
loaded_tokenizer_identity,
)
if not isinstance(runtime, nn.Module):
raise RuntimeError(
"Resynthesis integrated constructor returned a non-module runtime"
)
_bind_resynthesis_parent_runtime_facade(
runtime,
vocabulary_graph=vocabulary_graph,
)
if getattr(runtime, "resynthesis_rbo", None) is None:
raise RuntimeError("Resynthesis integrated runtime has no trained in-graph RBO")
if not bool(
getattr(runtime, "resynthesis_none_architecture_active", False)
):
raise RuntimeError(
"Resynthesis integrated runtime has no active NoNE architecture"
)
native_decode_confidence = getattr(runtime, "native_decode_confidence", None)
if not isinstance(native_decode_confidence, nn.Module):
raise RuntimeError(
"Resynthesis integrated runtime has no native decode-confidence diagnostic"
)
native_confidence_tensor_count = len(native_decode_confidence.state_dict())
if (
native_confidence_tensor_count
!= self.info.integrated_native_decode_confidence_tensors
):
raise RuntimeError(
"Resynthesis native decode-confidence tensor count differs from its manifest"
)
verified_artifact_sha256 = file_sha256_authority_boundary(
weights_path,
expected_sha256=self.info.model_artifact_sha256,
identity_cache_root=parent_sha256_cache_root,
)
if verified_artifact_sha256 != preload_artifact_sha256:
raise RuntimeError(
"loaded Resynthesis model artifact SHA-256 changed during parent load"
)
runtime.requires_grad_(False)
runtime.eval()
self.runtime = runtime
self.tokenizer = tokenizer
self.tokenizer_backend_identity = loaded_tokenizer_identity
self.verified_model_artifact_sha256 = verified_artifact_sha256
self.ladder_namespace_migration_applied = migration.ladder_namespace_applied
self.embedded_legacy_rbo_migration_applied = (
migration.embedded_legacy_rbo_applied
)
self.embedded_legacy_rbo_tensor_count = (
migration.embedded_legacy_rbo_tensor_count
)
self.additive_tensor_count_after_legacy_split = (
migration.additive_tensor_count_after_split
)
self.embedded_legacy_rbo_key_set_sha256 = (
migration.embedded_legacy_rbo_key_set_sha256
)
self.authoritative_rbo_tensor_count = migration.authoritative_rbo_tensor_count
self.authoritative_rbo_key_set_sha256 = (
migration.authoritative_rbo_key_set_sha256
)
self.exact_historical_source_loaded = migration.exact_historical_source_loaded
self.historical_tensor_native_hotpath_id = (
HISTORICAL_TENSOR_NATIVE_HOTPATH_ID
if migration.exact_historical_source_loaded
else ""
)
self.historical_recursive_arm_exhaustion_id = (
HISTORICAL_RECURSIVE_ARM_EXHAUSTION_ID
if migration.exact_historical_source_loaded
else ""
)
self.legacy_capability_source_sha256 = capability_source_sha256
object.__setattr__(self, "_pending_legacy_capability_bank", legacy_bank)
self._weights_loaded = True
self._base_loaded_marker.fill_(1)
return self
def checkpoint_lineage(self) -> dict[str, Any]:
"""Return immutable parent identity at the checkpoint I/O boundary."""
if not self._weights_loaded:
self.load_weights()
if not all(
(
self.info.checkpoint_id,
self.info.manifest_payload_sha256,
self.info.model_artifact_sha256,
)
):
raise RuntimeError(
"integrated Resynthesis parent has incomplete artifact lineage"
)
return {
"schema": "nnf.resynthesis.parent_lineage.v4",
"checkpointId": self.info.public_checkpoint_id,
"manifestPayloadSha256": self.info.manifest_payload_sha256,
"modelArtifactSha256": self.info.model_artifact_sha256,
"parameterElements": self.info.parameter_elements,
"nativeOwner": self.info.native_owner,
"nativeGeneration": self.info.native_generation,
"nativeRoot": self.info.native_root,
"nativeManifestSha256": self.info.native_manifest_sha256,
"nativeMigrationPromotionEligible": (
self.info.native_migration_promotion_eligible
),
"externalProductCheckpointDependency": False,
"composition": self.info.public_composition,
"modelType": self.info.public_model_type,
"historicalInheritedCheckpointId": (
self.info.historical_inherited_checkpoint_id
),
"historicalInheritedComposition": (
self.info.historical_inherited_composition
),
"historicalInheritedModelType": (
self.info.historical_inherited_model_type
),
"rboTensorCount": self.info.integrated_rbo_tensors,
"adaptiveTensorCount": self.info.integrated_additive_tensors,
"nativeDecodeConfidenceTensorCount": (
self.info.integrated_native_decode_confidence_tensors
),
"authoritativeRboKeySetSha256": self.authoritative_rbo_key_set_sha256,
"exactHistoricalSourceLoaded": self.exact_historical_source_loaded,
"historicalTensorNativeHotpathId": (
self.historical_tensor_native_hotpath_id
),
"historicalTensorNativeExpertDispatch": True,
"historicalTensorNativeHopSelection": True,
"historicalRecursiveArmExhaustionId": (
self.historical_recursive_arm_exhaustion_id
),
"historicalRecursiveArmExhaustionPropagation": True,
"embeddedLegacyCapabilityTensorCount": self.embedded_legacy_rbo_tensor_count,
"embeddedLegacyCapabilityKeySetSha256": (
self.embedded_legacy_rbo_key_set_sha256
),
"nativeAnswerSurface": "rbo_head_argmax_mapped_through_trained_vge",
"parentDualChunkRoPECompose": True,
"parentOnlineSoftmaxLongPool": True,
"nativeAttentionPositionAperture": NATIVE_ATTENTION_POSITION_APERTURE,
"onlineSoftmaxTileTokens": RESYNTHESIS_ONLINE_SOFTMAX_TILE_TOKENS,
"dualChunkPretrainLength": DUAL_CHUNK_PRETRAIN_LENGTH,
"dualChunkLocalSize": DUAL_CHUNK_LOCAL_SIZE,
"dualChunkAtSuccessiveSeamExposed": True,
"pretrainedRoPEBandTokens": PRETRAINED_ROPE_BAND_TOKENS,
"nativePrefillTileTokens": RESYNTHESIS_NATIVE_PREFILL_TILE_TOKENS,
}
def tokenizer_backend_receipt_boundary(self) -> dict[str, object]:
"""Serialize the verified parent-BPE executor at an I/O boundary."""
if not self._weights_loaded:
self.load_weights()
if self.tokenizer is None or self.tokenizer_backend_identity is None:
raise RuntimeError("integrated parent has no verified tokenizer backend")
return tokenizer_boundary_receipt(self.tokenizer)
def take_legacy_capability_bank(self) -> nn.Module | None:
"""Transfer the preserved payload into the trainable Resynthesis graph."""
if not self._weights_loaded:
self.load_weights()
bank = getattr(self, "_pending_legacy_capability_bank", None)
if bank is not None and not isinstance(bank, nn.Module):
raise RuntimeError("pending legacy capability owner is not a module")
object.__setattr__(self, "_pending_legacy_capability_bank", None)
return bank
def active_none_fabric(self) -> nn.Module:
"""Return the exact trained NoNE fabric owned by the loaded parent graph."""
if not self._weights_loaded:
self.load_weights()
runtime = self.runtime
if runtime is None:
raise RuntimeError("Resynthesis integrated runtime was not attached")
parent_rbo = getattr(runtime, "resynthesis_rbo", None)
parent_fabric = getattr(parent_rbo, "none_fabric", None)
if not isinstance(parent_fabric, nn.Module):
raise RuntimeError("Resynthesis integrated RBO has no trained NoNE fabric module")
if not callable(getattr(parent_fabric, "step_phase", None)):
raise RuntimeError(
"Resynthesis parent NoNE fabric exposes no step_phase contract"
)
return parent_fabric
def _record_native_context_cache_telemetry_boundary(
self,
cache: object,
) -> None:
"""Update diagnostic cache geometry in model-owned tensor buffers."""
packet = NativeContextCacheTelemetry(
hybrid_active=self._native_context_hybrid_active,
total_positions=self._native_context_total_positions,
hot_resident_positions=self._native_context_hot_resident_positions,
hot_window_tokens=self._native_context_hot_window_tokens,
recurrent_layer_count=self._native_context_recurrent_layer_count,
full_attention_layer_count=(
self._native_context_full_attention_layer_count
),
)
_update_native_context_cache_telemetry_boundary(
cache,
packet=packet,
)
def native_context_cache_telemetry_boundary(
self,
) -> NativeContextCacheTelemetry:
"""Return read-only tensor evidence; this packet has no answer authority."""
return NativeContextCacheTelemetry(
hybrid_active=self._native_context_hybrid_active.detach().clone(),
total_positions=self._native_context_total_positions.detach().clone(),
hot_resident_positions=(
self._native_context_hot_resident_positions.detach().clone()
),
hot_window_tokens=(self._native_context_hot_window_tokens.detach().clone()),
recurrent_layer_count=(
self._native_context_recurrent_layer_count.detach().clone()
),
full_attention_layer_count=(
self._native_context_full_attention_layer_count.detach().clone()
),
)
def _prefill_tiled_native(
self,
*,
runtime: nn.Module,
active_input_ids: torch.Tensor,
position_ids: torch.Tensor,
attention_mask: torch.Tensor,
record_kv: object,
starting_prefix: torch.Tensor,
telemetry_required: bool,
) -> NativeTiledPrefill:
"""Run a retained-cache prefill for native autoregressive continuation."""
return self._prefill_tiled_native_impl(
runtime=runtime,
active_input_ids=active_input_ids,
position_ids=position_ids,
attention_mask=attention_mask,
record_kv=record_kv,
starting_prefix=starting_prefix,
telemetry_required=telemetry_required,
retain_final_cache=True,
)
def _prefill_tiled_training_native(
self,
*,
runtime: nn.Module,
active_input_ids: torch.Tensor,
position_ids: torch.Tensor,
attention_mask: torch.Tensor,
record_kv: object,
starting_prefix: torch.Tensor,
telemetry_required: bool,
) -> NativeTiledPrefill:
"""Run an exact masked training prefill without retaining its final cache."""
return self._prefill_tiled_native_impl(
runtime=runtime,
active_input_ids=active_input_ids,
position_ids=position_ids,
attention_mask=attention_mask,
record_kv=record_kv,
starting_prefix=starting_prefix,
telemetry_required=telemetry_required,
retain_final_cache=False,
)
def _prefill_tiled_native_impl(
self,
*,
runtime: nn.Module,
active_input_ids: torch.Tensor,
position_ids: torch.Tensor,
attention_mask: torch.Tensor,
record_kv: object,
starting_prefix: torch.Tensor,
telemetry_required: bool,
retain_final_cache: bool,
) -> NativeTiledPrefill:
"""Tile the parent prefill so 4M+ prompts fit bounded activation memory.
Each tile threads the parent cache and uses a CUMULATIVE attention mask
(``attention_mask[:, :end]``). Within the pretrained band, the parent
keeps exact full-attention K/V. Longer prompts use its trained hybrid
topology: all 24 linear-attention layers keep global recurrent state,
while eight full-attention layers keep one Dual-Chunk hot K/V window.
Folded ``position_ids`` are sliced per tile (never re-folded), and no
token is dropped or truncated.
Raises if the parent loses its KV cache before every prompt token has
been preflighted.
"""
total_prefill_tokens = int(active_input_ids.shape[1])
tile_tokens = min(
total_prefill_tokens,
RESYNTHESIS_NATIVE_PREFILL_TILE_TOKENS,
)
running_prefix = starting_prefix.clone()
tile_past: Any = None
summary_hidden: torch.Tensor | None = None
summary_index = 0
summary_capacity = (
(total_prefill_tokens + tile_tokens - 1) // tile_tokens
) * RESYNTHESIS_NATIVE_PREFILL_SUMMARIES_PER_TILE
if total_prefill_tokens > RESYNTHESIS_NATIVE_HYBRID_ACTIVATION_TOKENS:
custom_cache_builder = getattr(
runtime,
"resynthesis_build_native_hybrid_context_cache",
None,
)
if callable(custom_cache_builder):
tile_past = custom_cache_builder(
hot_window_tokens=active_input_ids.new_ones(
(), dtype=torch.long
).mul_(RESYNTHESIS_NATIVE_HOT_KV_TOKENS)
)
else:
tile_past = _build_native_hybrid_context_cache(runtime)
if not bool(
getattr(
tile_past,
"_resynthesis_native_hybrid_context",
False,
)
):
raise RuntimeError(
"Resynthesis native hybrid context cache has no model-owned marker"
)
out: Any = None
for start in range(0, total_prefill_tokens, tile_tokens):
end = min(total_prefill_tokens, start + tile_tokens)
tile_cache_required = retain_final_cache or end != total_prefill_tokens
tile_ids = active_input_ids[:, start:end]
tile_positions = position_ids[:, start:end]
# Cumulative mask keeps every prior token attendable under exact SDPA.
tile_mask = attention_mask[:, :end]
tile_new_positions = tile_ids.new_ones(
(), dtype=torch.long
).mul_(tile_ids.shape[1])
if callable(record_kv):
record_kv(
prefix_positions=running_prefix,
new_positions=tile_new_positions,
)
elif telemetry_required:
raise RuntimeError(
"Resynthesis integrated runtime lacks tensor-native KV telemetry"
)
out = runtime(
input_ids=tile_ids,
position_ids=tile_positions,
attention_mask=tile_mask,
past_key_values=tile_past,
output_hidden_states=True,
return_dict=True,
use_cache=tile_cache_required,
logits_to_keep=1,
)
tile_hidden_states = getattr(out, "hidden_states", None)
if not tile_hidden_states or not isinstance(
tile_hidden_states[-1], torch.Tensor
):
raise RuntimeError(
"Resynthesis native prefill returned no hidden state for additive attention"
)
tile_hidden = tile_hidden_states[-1]
if tile_hidden.ndim != 3 or tile_hidden.shape[1] < 1:
raise RuntimeError("Resynthesis native prefill hidden geometry differs")
if summary_hidden is None:
summary_hidden = tile_hidden.new_empty(
tile_hidden.shape[0],
summary_capacity,
tile_hidden.shape[-1],
)
segment_count = min(
RESYNTHESIS_NATIVE_PREFILL_SUMMARIES_PER_TILE,
tile_hidden.shape[1],
)
for segment_index in range(segment_count):
segment_start = segment_index * tile_hidden.shape[1] // segment_count
segment_end = (
(segment_index + 1) * tile_hidden.shape[1] // segment_count
)
segment_hidden = tile_hidden[:, segment_start:segment_end, :]
# The installed final-hidden-only parent contract returns one
# logits-aligned position for ``logits_to_keep=1``. Attention
# over a singleton has exact unit weight, so its output is the
# same tensor. Avoid launching score, softmax, ring-softmax,
# matmul, arange, and allocating index_copy for every prefill
# tile. The multi-position compatibility path retains the full
# trained attention pool without changing its semantics.
segment_context = (
segment_hidden[:, 0, :]
if segment_hidden.shape[1] == 1
else _parent_hidden_context(segment_hidden)
)
summary_hidden[:, summary_index, :].copy_(segment_context)
summary_index += 1
returned_past = getattr(out, "past_key_values", None)
if tile_cache_required and returned_past is not None:
tile_past = returned_past
running_prefix = running_prefix + tile_new_positions
elif end != total_prefill_tokens:
raise RuntimeError(
"Resynthesis native 4M prefill lost KV cache before all tokens"
)
elif not tile_cache_required:
# A pre-existing multi-tile cache is passed into the final tile
# so attention remains exact. Some parent implementations return
# that object even with ``use_cache=False``; sever it explicitly
# after extracting hidden/logits so one training wave cannot
# retain parent KV/recurrent state into the next wave.
tile_past = None
setattr(out, "past_key_values", None)
else:
tile_past = None
if out is None:
raise RuntimeError("Resynthesis native prefill received an empty prompt")
if summary_hidden is None or summary_index < 1:
raise RuntimeError("Resynthesis native prefill produced no attended summaries")
return NativeTiledPrefill(
runtime_output=out,
summary_hidden=summary_hidden.narrow(1, 0, summary_index),
input_positions=active_input_ids.new_ones(
(), dtype=torch.long
).mul_(total_prefill_tokens),
)
def forward_hidden_logits(
self,
input_ids: torch.Tensor,
attention_mask: torch.Tensor | None = None,
) -> ResynthesisParentForward:
"""Execute one cache-retaining causal parent prefill or continuation."""
return self._forward_hidden_logits_impl(
input_ids,
attention_mask=attention_mask,
masked_training_prefill=False,
)
def forward_training_hidden_logits(
self,
input_ids: torch.Tensor,
*,
attention_mask: torch.Tensor,
) -> ResynthesisParentForward:
"""Execute one fresh masked training prefill with no final parent cache."""
return self._forward_hidden_logits_impl(
input_ids,
attention_mask=attention_mask,
masked_training_prefill=True,
)
def forward_training_hidden_logits_shared_prefix(
self,
packet: NativeSharedPrefixTrainingPacket,
) -> ResynthesisParentForward:
"""Run one packed batch from an immutable frozen-parent prefix cache.
Prefix construction enters only the frozen backbone. The integrated
parent RBO/Fabric therefore still executes exactly once, on the suffix
batch, matching an ordinary final-position training prefill while
avoiding one identical backbone prefix execution per row.
"""
if not self._weights_loaded:
self.load_weights()
runtime = self.runtime
if runtime is None:
raise RuntimeError("Resynthesis integrated runtime was not attached")
prefix_ids = packet.prefix_ids.to(device=self.device, dtype=torch.long)
suffix_ids = packet.suffix_ids.to(device=self.device, dtype=torch.long)
suffix_mask = packet.suffix_mask.to(
device=self.device,
dtype=torch.long,
)
batch_indices_t = packet.batch_indices_t.to(
device=self.device,
dtype=torch.long,
)
if (
prefix_ids.ndim != 2
or prefix_ids.shape[0] != 1
or prefix_ids.shape[1] < 1
or suffix_ids.ndim != 2
or suffix_ids.shape[0] < 1
or suffix_ids.shape[1] < 1
or suffix_mask.shape != suffix_ids.shape
or batch_indices_t.shape != (suffix_ids.shape[0],)
):
raise RuntimeError("shared parent-prefix training geometry differs")
torch._assert_async(
suffix_mask.ge(0).logical_and(suffix_mask.le(1)).all(),
"shared parent-prefix suffix mask is not binary",
)
torch._assert_async(
suffix_mask[:, -1].eq(1).all(),
"shared parent-prefix suffix must end in a visible token",
)
if (
suffix_ids.shape[0] < 2
or prefix_ids.shape[1] + suffix_ids.shape[1]
> RESYNTHESIS_NATIVE_PREFILL_TILE_TOKENS
):
# A post-materialization CUDA replan may isolate one row, while a
# packed full-payload window may cross the native prefill tile.
# Neither condition invalidates the prompt. Reconstruct the exact
# target-free masked batch and use the ordinary tiled parent path;
# the live additive RBO/Fabric/expert/page graph still runs after
# this frozen-parent boundary. This is a semantic fallback, not a
# token cap: every visible suffix token remains in the forward.
prefix_batch_ids = prefix_ids.expand(suffix_ids.shape[0], -1)
full_ids = torch.cat((prefix_batch_ids, suffix_ids), dim=1)
full_mask = torch.cat(
(
torch.ones_like(prefix_batch_ids, dtype=torch.long),
suffix_mask,
),
dim=1,
)
return self.forward_training_hidden_logits(
full_ids,
attention_mask=full_mask,
)
if self._decode_past_key_values is not None:
raise RuntimeError(
"shared parent-prefix training requires a fresh decode arm"
)
torch._assert_async(
self._decode_cached_positions.eq(0),
"shared parent-prefix training requires a fresh decode arm",
)
backbone = getattr(runtime, "backbone", None)
if not isinstance(backbone, nn.Module) or getattr(
backbone,
"_resynthesis_final_hidden_only_forward_id",
"",
) != PARENT_FINAL_HIDDEN_ONLY_FORWARD_ID:
raise RuntimeError(
"shared parent-prefix training requires the exact frozen backbone"
)
decoder_core = getattr(backbone, "model", None)
if not isinstance(decoder_core, nn.Module):
raise RuntimeError(
"shared parent-prefix training requires the frozen decoder core"
)
if not self._decode_authority_active:
begin_decode = getattr(runtime, "resynthesis_begin_authority_decode", None)
if callable(begin_decode):
authority_forward_id = begin_decode()
if not isinstance(authority_forward_id, torch.Tensor):
raise RuntimeError(
"Resynthesis authority transaction returned no tensor forward ID"
)
elif not self.exact_historical_source_loaded:
raise RuntimeError(
"Resynthesis integrated runtime lacks its authority transaction"
)
self._decode_authority_active = True
prefix_mask = torch.ones_like(prefix_ids, dtype=torch.long)
prefix_positions_t = (
prefix_mask.cumsum(dim=1).sub(1).clamp_min(0)
)
validate_native_context_admission(prefix_positions_t)
prefix_position_stack = build_long_context_position_stack(
prefix_positions_t
)
master_cache = _build_native_hybrid_context_cache(runtime)
prefix_count_t = prefix_ids.new_ones((), dtype=torch.long).mul_(
prefix_ids.shape[1]
)
with torch.no_grad():
# Enter the decoder core, not the conditional-generation wrapper.
# Its LM head owns additive experts and the embedded parent RBO;
# invoking that head here would execute RBO/Fabric once on the
# prefix and again on the suffix. The immutable prefix needs only
# the decoder's exact K/V and recurrent state.
prefix_output = decoder_core(
input_ids=prefix_ids,
position_ids=prefix_position_stack.rope_position_ids,
attention_mask=prefix_mask,
past_key_values=master_cache,
output_hidden_states=False,
return_dict=True,
use_cache=True,
)
returned_master = getattr(prefix_output, "past_key_values", None)
if (
returned_master is not master_cache
or getattr(
returned_master,
"_resynthesis_native_hybrid_context",
False,
)
is not True
):
raise RuntimeError(
"frozen backbone did not return its immutable hybrid prefix cache"
)
expanded_cache = _expand_immutable_native_hybrid_context_cache_boundary(
master_cache,
batch_indices_t,
)
setattr(prefix_output, "past_key_values", None)
prefix_batch_mask = prefix_mask.expand(suffix_ids.shape[0], -1)
full_input_positions_t = prefix_count_t + suffix_ids.new_ones(
(), dtype=torch.long
).mul_(suffix_ids.shape[1])
return self._forward_hidden_logits_impl(
suffix_ids,
attention_mask=suffix_mask,
masked_training_prefill=True,
shared_prefix_cache=expanded_cache,
shared_prefix_mask=prefix_batch_mask,
shared_full_input_positions_t=full_input_positions_t,
)
def frozen_backbone_position_policy_sha256_boundary(self) -> str:
"""Return the immutable decoder position/tile policy identity."""
return frozen_backbone_position_policy_sha256_boundary(
final_hidden_adapter_id=PARENT_FINAL_HIDDEN_ONLY_FORWARD_ID,
dual_chunk_pretrain_length=DUAL_CHUNK_PRETRAIN_LENGTH,
dual_chunk_local_size=DUAL_CHUNK_LOCAL_SIZE,
pretrained_rope_band_tokens=PRETRAINED_ROPE_BAND_TOKENS,
prefill_tile_tokens=RESYNTHESIS_NATIVE_PREFILL_TILE_TOKENS,
prefill_summaries_per_tile=(
RESYNTHESIS_NATIVE_PREFILL_SUMMARIES_PER_TILE
),
)
def frozen_backbone_feature_dtype_boundary(self) -> torch.dtype:
"""Return the exact frozen decoder feature dtype before sidecar lookup.
A durable feature object is keyed before the expensive parent forward.
The dtype therefore has to come from the loaded decoder itself, not
from a host configuration guess. Captured features are authoritative
when this process has already run a prefill; otherwise the first
floating decoder parameter supplies the output dtype used by the
frozen final-hidden adapter.
"""
captured = self._last_frozen_backbone_final_hidden_t
if isinstance(captured, torch.Tensor):
return captured.dtype
if not self._weights_loaded:
self.load_weights()
runtime = self.runtime
backbone = getattr(runtime, "backbone", None)
decoder_core = getattr(backbone, "model", None)
if not isinstance(decoder_core, nn.Module):
raise RuntimeError(
"frozen-backbone feature dtype requires the frozen decoder core"
)
for parameter in decoder_core.parameters():
if parameter.is_floating_point():
return parameter.dtype
for buffer in decoder_core.buffers():
if buffer.is_floating_point():
return buffer.dtype
raise RuntimeError(
"frozen-backbone decoder has no floating feature dtype authority"
)
def frozen_backbone_cache_authority_boundary(
self,
*,
qualified_work_id: str,
prompt_sha256: str,
prompt_mask_sha256: str,
token_dtype: torch.dtype,
feature_dtype: torch.dtype,
summary_positions: int,
) -> FrozenBackboneCacheAuthority:
"""Bind one external packed row to this exact immutable parent."""
token_dtype_name = str(token_dtype).removeprefix("torch.")
feature_dtype_name = str(feature_dtype).removeprefix("torch.")
authority = FrozenBackboneCacheAuthority(
qualified_work_id=qualified_work_id,
prompt_sha256=prompt_sha256,
prompt_mask_sha256=prompt_mask_sha256,
parent_checkpoint_id=self.info.checkpoint_id,
parent_manifest_payload_sha256=(
self.info.manifest_payload_sha256
),
parent_model_artifact_sha256=self.info.model_artifact_sha256,
# Retained only as a direct-path hint for a historical source-bound
# sidecar. It is excluded from executable cache identity; a
# read-only normalized index finds older source observations.
parent_source_bundle_sha256=(
self.info.parent_source_bundle_sha256
),
position_policy_sha256=(
self.frozen_backbone_position_policy_sha256_boundary()
),
token_dtype=token_dtype_name,
feature_dtype=feature_dtype_name,
hidden_size=self.info.hidden_size,
summary_positions=summary_positions,
)
authority.record_boundary()
return authority
def precompute_frozen_backbone_prefill_boundary(
self,
input_ids: torch.Tensor,
*,
attention_mask: torch.Tensor,
authorities: tuple[FrozenBackboneCacheAuthority, ...],
) -> FrozenBackbonePrefillPacket:
"""Compute raw frozen-decoder features without entering live heads.
This is the one-shot packed-sidecar producer boundary. It deliberately
calls only the immutable decoder core: the conditional-generation
wrapper, additive head, historical RBO, Fabric, experts, page runtime,
answer targets, and task-intent targets are outside this interface.
The returned typed packet can therefore accelerate a later training
forward without caching any trainable or routing-owned result.
"""
if not self._weights_loaded:
self.load_weights()
runtime = self.runtime
if runtime is None:
raise RuntimeError("Resynthesis integrated runtime was not attached")
if (
input_ids.ndim != 2
or input_ids.shape[0] < 1
or input_ids.shape[1] < 1
or attention_mask.shape != input_ids.shape
or len(authorities) != input_ids.shape[0]
):
raise ValueError(
"frozen-backbone precompute prompt geometry differs"
)
if self._decode_past_key_values is not None:
raise RuntimeError(
"frozen-backbone precompute requires a fresh decode arm"
)
active_input_ids = input_ids.to(
device=self.device,
dtype=torch.long,
)
active_attention_mask = attention_mask.to(
device=self.device,
dtype=torch.long,
)
torch._assert_async(
active_attention_mask.ge(0).logical_and(
active_attention_mask.le(1)
).all(),
"frozen-backbone precompute attention mask is not binary",
)
torch._assert_async(
active_attention_mask[:, -1].eq(1).all(),
"frozen-backbone precompute prompt must end in a visible token",
)
torch._assert_async(
self._decode_cached_positions.eq(0),
"frozen-backbone precompute requires a fresh decode arm",
)
backbone = getattr(runtime, "backbone", None)
if not isinstance(backbone, nn.Module) or getattr(
backbone,
"_resynthesis_final_hidden_only_forward_id",
"",
) != PARENT_FINAL_HIDDEN_ONLY_FORWARD_ID:
raise RuntimeError(
"frozen-backbone precompute requires the exact parent adapter"
)
decoder_core = getattr(backbone, "model", None)
if not isinstance(decoder_core, nn.Module):
raise RuntimeError(
"frozen-backbone precompute requires the frozen decoder core"
)
absolute_positions = (
active_attention_mask.cumsum(dim=1).sub(1).clamp_min(0)
)
validate_native_context_admission(absolute_positions)
position_ids = build_long_context_position_stack(
absolute_positions
).rope_position_ids
total_prefill_tokens = active_input_ids.shape[1]
tile_tokens = min(
total_prefill_tokens,
RESYNTHESIS_NATIVE_PREFILL_TILE_TOKENS,
)
summary_positions = (
total_prefill_tokens + tile_tokens - 1
) // tile_tokens
feature_dtype = self.frozen_backbone_feature_dtype_boundary()
if any(
authority.parent_checkpoint_id != self.info.checkpoint_id
or authority.parent_manifest_payload_sha256
!= self.info.manifest_payload_sha256
or authority.parent_model_artifact_sha256
!= self.info.model_artifact_sha256
or authority.position_policy_sha256
!= self.frozen_backbone_position_policy_sha256_boundary()
or authority.token_dtype
!= str(input_ids.dtype).removeprefix("torch.")
or authority.feature_dtype
!= str(feature_dtype).removeprefix("torch.")
or authority.hidden_size != self.info.hidden_size
or authority.summary_positions != summary_positions
for authority in authorities
):
raise RuntimeError(
"frozen-backbone precompute authority differs from the parent"
)
tile_past: Any = None
if (
total_prefill_tokens
> RESYNTHESIS_NATIVE_HYBRID_ACTIVATION_TOKENS
):
custom_cache_builder = getattr(
runtime,
"resynthesis_build_native_hybrid_context_cache",
None,
)
if callable(custom_cache_builder):
tile_past = custom_cache_builder(
hot_window_tokens=active_input_ids.new_ones(
(),
dtype=torch.long,
).mul_(RESYNTHESIS_NATIVE_HOT_KV_TOKENS)
)
else:
tile_past = _build_native_hybrid_context_cache(runtime)
if not bool(
getattr(
tile_past,
"_resynthesis_native_hybrid_context",
False,
)
):
raise RuntimeError(
"frozen-backbone precompute hybrid cache is not native"
)
summary_hidden_t: torch.Tensor | None = None
final_hidden_t: torch.Tensor | None = None
summary_index = 0
with torch.no_grad():
for start in range(0, total_prefill_tokens, tile_tokens):
end = min(total_prefill_tokens, start + tile_tokens)
tile_cache_required = end != total_prefill_tokens
output = decoder_core(
input_ids=active_input_ids[:, start:end],
position_ids=position_ids[:, start:end],
attention_mask=active_attention_mask[:, :end],
past_key_values=tile_past,
output_hidden_states=False,
return_dict=True,
use_cache=tile_cache_required,
)
tile_hidden_t = getattr(
output,
"last_hidden_state",
None,
)
if (
not isinstance(tile_hidden_t, torch.Tensor)
or tile_hidden_t.ndim != 3
or tile_hidden_t.shape[0] != input_ids.shape[0]
or tile_hidden_t.shape[1] < 1
or tile_hidden_t.shape[2] != self.info.hidden_size
or tile_hidden_t.dtype != feature_dtype
):
raise RuntimeError(
"frozen-backbone decoder feature geometry differs"
)
# The installed parent adapter exposes the one logits-aligned
# final position from each tile. Selecting the same position
# here makes the offline packet byte-equivalent without
# invoking the additive LM head that records it online.
final_hidden_t = tile_hidden_t[:, -1:, :]
if summary_hidden_t is None:
summary_hidden_t = final_hidden_t.new_empty(
input_ids.shape[0],
summary_positions,
self.info.hidden_size,
)
summary_hidden_t[:, summary_index : summary_index + 1, :].copy_(
final_hidden_t
)
summary_index += 1
returned_past = getattr(output, "past_key_values", None)
if tile_cache_required and returned_past is not None:
tile_past = returned_past
elif tile_cache_required:
raise RuntimeError(
"frozen-backbone precompute lost its decoder cache"
)
else:
tile_past = None
setattr(output, "past_key_values", None)
if (
final_hidden_t is None
or summary_hidden_t is None
or summary_index != summary_positions
):
raise RuntimeError(
"frozen-backbone precompute produced incomplete features"
)
input_positions_t = active_input_ids.new_ones(
(input_ids.shape[0],),
dtype=torch.long,
).mul_(total_prefill_tokens)
row_packets = tuple(
FrozenBackbonePrefillPacket.from_features_boundary(
authority=authority,
final_hidden_t=final_hidden_t[row_index : row_index + 1],
summary_hidden_t=summary_hidden_t[
row_index : row_index + 1
],
input_positions_t=input_positions_t[
row_index : row_index + 1
],
)
for row_index, authority in enumerate(authorities)
)
return FrozenBackbonePrefillPacket.stack_boundary(row_packets)
def frozen_backbone_prefill_row_packet_boundary(
self,
*,
authority: FrozenBackboneCacheAuthority,
batch_index: int,
) -> FrozenBackbonePrefillPacket:
"""Select one just-computed raw decoder row for durable external I/O."""
final_hidden_t = self._last_frozen_backbone_final_hidden_t
summary_hidden_t = self._last_frozen_backbone_summary_hidden_t
input_positions_t = self._last_frozen_backbone_input_positions_t
if (
not isinstance(final_hidden_t, torch.Tensor)
or not isinstance(summary_hidden_t, torch.Tensor)
or not isinstance(input_positions_t, torch.Tensor)
):
raise RuntimeError(
"frozen-backbone row requested before one masked parent prefill"
)
if batch_index < 0 or batch_index >= final_hidden_t.shape[0]:
raise IndexError("frozen-backbone batch row is unavailable")
if (
authority.parent_checkpoint_id != self.info.checkpoint_id
or authority.parent_manifest_payload_sha256
!= self.info.manifest_payload_sha256
or authority.parent_model_artifact_sha256
!= self.info.model_artifact_sha256
or authority.position_policy_sha256
!= self.frozen_backbone_position_policy_sha256_boundary()
or authority.hidden_size != self.info.hidden_size
or authority.summary_positions != summary_hidden_t.shape[1]
or authority.feature_dtype
!= str(final_hidden_t.dtype).removeprefix("torch.")
):
raise RuntimeError(
"frozen-backbone row authority differs from the loaded parent"
)
row = slice(batch_index, batch_index + 1)
return FrozenBackbonePrefillPacket.from_features_boundary(
authority=authority,
final_hidden_t=final_hidden_t[row],
summary_hidden_t=summary_hidden_t[row],
input_positions_t=input_positions_t[row],
)
def forward_training_from_frozen_backbone_packet(
self,
input_ids: torch.Tensor,
*,
attention_mask: torch.Tensor,
packet: FrozenBackbonePrefillPacket,
) -> ResynthesisParentForward:
"""Replay decoder-only features through the complete live parent graph.
The decoder core is replaced for this one synchronous call only. The
surrounding conditional-generation wrapper, additive LM head,
historical RBO, and historical Fabric all execute normally for every
original prefill tile. This is deliberately narrower than replaying a
cached ``ResynthesisParentForward``, which would bypass those trained surfaces.
"""
if not self._weights_loaded:
self.load_weights()
runtime = self.runtime
if runtime is None:
raise RuntimeError("Resynthesis integrated runtime was not attached")
packet.validate_boundary()
active_packet = packet.to_device_boundary(self.device)
if (
input_ids.ndim != 2
or input_ids.shape[1] < 1
or attention_mask.shape != input_ids.shape
or attention_mask.dtype != torch.bool
or active_packet.final_hidden_t.shape[0] != input_ids.shape[0]
or active_packet.final_hidden_t.shape[2] != self.info.hidden_size
):
raise RuntimeError(
"frozen-backbone replay prompt/feature geometry differs"
)
expected_summary_positions = (
input_ids.shape[1] + RESYNTHESIS_NATIVE_PREFILL_TILE_TOKENS - 1
) // RESYNTHESIS_NATIVE_PREFILL_TILE_TOKENS
if (
active_packet.summary_hidden_t.shape[1]
!= expected_summary_positions
):
raise RuntimeError(
"frozen-backbone replay summary/tile geometry differs"
)
expected_input_positions_t = attention_mask.sum(
dim=1,
dtype=torch.long,
)
torch._assert_async(
active_packet.input_positions_t.eq(
expected_input_positions_t
).all(),
"frozen-backbone replay input positions differ",
)
expected_left_padding_mask_t = torch.arange(
input_ids.shape[1],
device=attention_mask.device,
dtype=torch.long,
).unsqueeze(0).ge(
input_ids.shape[1] - expected_input_positions_t.unsqueeze(1)
)
torch._assert_async(
attention_mask.eq(expected_left_padding_mask_t).all(),
"frozen-backbone replay mask is not canonical left padding",
)
torch._assert_async(
input_ids.masked_select(~attention_mask).eq(0).all(),
"frozen-backbone replay padding tokens must be zero",
)
torch._assert_async(
active_packet.final_hidden_t.eq(
active_packet.summary_hidden_t[:, -1:, :]
).all(),
"frozen-backbone final hidden differs from its final tile summary",
)
backbone = getattr(runtime, "backbone", None)
if not isinstance(backbone, nn.Module) or getattr(
backbone,
"_resynthesis_final_hidden_only_forward_id",
"",
) != PARENT_FINAL_HIDDEN_ONLY_FORWARD_ID:
raise RuntimeError(
"frozen-backbone replay requires the exact parent adapter"
)
decoder_core = getattr(backbone, "model", None)
if not isinstance(decoder_core, nn.Module):
raise RuntimeError(
"frozen-backbone replay requires the frozen decoder core"
)
original_forward = decoder_core.forward
replay_call_index = 0
replay_cache = object()
def replay_decoder_forward(
*args: Any,
**kwargs: Any,
) -> _FrozenBackboneDecoderOutput:
nonlocal replay_call_index
input_value = kwargs.get("input_ids")
if input_value is None and args:
input_value = args[0]
if (
not isinstance(input_value, torch.Tensor)
or input_value.ndim != 2
or input_value.shape[0]
!= active_packet.final_hidden_t.shape[0]
or replay_call_index
>= active_packet.summary_hidden_t.shape[1]
):
raise RuntimeError(
"frozen-backbone decoder replay invocation differs"
)
hidden_t = active_packet.summary_hidden_t[
:,
replay_call_index : replay_call_index + 1,
:,
]
replay_call_index += 1
past_key_values = kwargs.get("past_key_values")
if kwargs.get("use_cache") is True and past_key_values is None:
past_key_values = replay_cache
return _FrozenBackboneDecoderOutput(
last_hidden_state=hidden_t,
past_key_values=past_key_values,
)
setattr(decoder_core, "forward", replay_decoder_forward)
try:
result = self._forward_hidden_logits_impl(
input_ids,
attention_mask=attention_mask,
masked_training_prefill=True,
)
finally:
setattr(decoder_core, "forward", original_forward)
if replay_call_index != active_packet.summary_hidden_t.shape[1]:
raise RuntimeError(
"frozen-backbone replay did not consume every sealed tile"
)
# The live tiled wrapper reports the padded batch width because every
# row shares its decoder call geometry. Durable sidecars retain the
# exact unpadded row positions, which are the prefill participation
# evidence consumed by the live RBO/Fabric graph. Preserve that vector
# after the decoder-only replay instead of manufacturing the wider
# cohort position for shorter rows.
return replace(
result,
parent_prefill_input_positions=(
active_packet.input_positions_t.detach()
),
)
def _forward_hidden_logits_impl(
self,
input_ids: torch.Tensor,
*,
attention_mask: torch.Tensor | None,
masked_training_prefill: bool,
shared_prefix_cache: object | None = None,
shared_prefix_mask: torch.Tensor | None = None,
shared_full_input_positions_t: torch.Tensor | None = None,
) -> ResynthesisParentForward:
"""Execute one causal parent prefill or native KV continuation.
A decode arm runs the complete prompt exactly once. Later emissions
enter the same integrated Resynthesis/RBO graph as one new token with
its native KV state. If the parent supplies no cache, this falls back
to complete-prefix execution without changing stop or route authority.
"""
if not self._weights_loaded:
self.load_weights()
runtime = self.runtime
if runtime is None:
raise RuntimeError("Resynthesis integrated runtime was not attached")
self._last_frozen_backbone_final_hidden_t = None
self._last_frozen_backbone_summary_hidden_t = None
self._last_frozen_backbone_input_positions_t = None
if input_ids.ndim != 2 or input_ids.shape[1] < 1:
raise ValueError("Resynthesis parent input IDs must have shape [batch, sequence]")
active_input_ids = input_ids.to(device=self.device, dtype=torch.long)
explicit_attention_mask = attention_mask is not None
if attention_mask is None:
active_attention_mask = torch.ones(
input_ids.shape,
device=self.device,
dtype=torch.long,
)
else:
if attention_mask.shape != input_ids.shape:
raise ValueError("Resynthesis parent attention mask geometry differs")
active_attention_mask = attention_mask.to(
device=self.device,
dtype=torch.long,
)
torch._assert_async(
active_attention_mask.ge(0).logical_and(
active_attention_mask.le(1)
).all(),
"Resynthesis parent attention mask is not binary",
)
torch._assert_async(
active_attention_mask[:, -1].eq(1).all(),
"Resynthesis left-padded prompt must end in a visible token",
)
shared_prefix_active = shared_prefix_cache is not None
if shared_prefix_active != (
shared_prefix_mask is not None
and shared_full_input_positions_t is not None
):
raise RuntimeError("shared parent-prefix cache boundary is incomplete")
if shared_prefix_active and (
not masked_training_prefill
or not explicit_attention_mask
or shared_prefix_mask is None
or shared_prefix_mask.ndim != 2
or shared_prefix_mask.shape[0] != active_input_ids.shape[0]
or shared_prefix_mask.shape[1] < 1
or shared_prefix_mask.device != active_input_ids.device
or self._decode_past_key_values is not None
):
raise RuntimeError("shared parent-prefix continuation geometry differs")
use_past = (
shared_prefix_active
or self._decode_past_key_values is not None
)
if not self._decode_authority_active:
begin_decode = getattr(runtime, "resynthesis_begin_authority_decode", None)
if callable(begin_decode):
authority_forward_id = begin_decode()
if not isinstance(authority_forward_id, torch.Tensor):
raise RuntimeError(
"Resynthesis authority transaction returned no tensor forward ID"
)
elif not self.exact_historical_source_loaded:
raise RuntimeError(
"Resynthesis integrated runtime lacks its authority transaction"
)
self._decode_authority_active = True
if use_past:
if attention_mask is not None and not shared_prefix_active:
raise RuntimeError(
"explicit batched attention masks require a fresh decode arm"
)
if not shared_prefix_active:
active_input_ids = active_input_ids[:, -1:]
torch._assert_async(
self._decode_cached_positions.eq(input_ids.shape[1] - 1),
"Resynthesis KV continuation does not match the visible prefix",
)
prefix_positions = (
shared_prefix_mask.new_ones((), dtype=torch.long).mul_(
shared_prefix_mask.shape[1]
)
if shared_prefix_active and shared_prefix_mask is not None
else self._decode_cached_positions.to(
device=active_input_ids.device,
dtype=torch.long,
).clone()
)
runtime_new_positions = active_input_ids.new_ones(
(), dtype=torch.long
).mul_(active_input_ids.shape[1])
result_prefix_positions = (
prefix_positions.new_zeros(())
if shared_prefix_active
else prefix_positions
)
result_new_positions = (
shared_full_input_positions_t
if shared_prefix_active
and shared_full_input_positions_t is not None
else runtime_new_positions
)
record_kv = getattr(runtime, "resynthesis_record_kv_cache_reuse", None)
position_attention_mask = (
active_attention_mask
if shared_prefix_active
else active_attention_mask[:, -active_input_ids.shape[1] :]
if use_past
else active_attention_mask
)
absolute_positions = (
position_attention_mask.cumsum(dim=1)
.sub(1)
.clamp_min(0)
.add(prefix_positions.to(device=self.device))
)
validate_native_context_admission(absolute_positions)
# STACK+COMPOSE parent decode: absolute KV cache + Dual Chunk → parent RoPE.
# ``position_ids`` feed the parent's live rotary/YaRN; KV slots stay absolute.
# The full-prompt folded positions are computed once and then SLICED per
# prefill tile, so each tile sees the correct in-distribution RoPE phase
# for its absolute position (never re-folded per tile).
position_stack = build_long_context_position_stack(absolute_positions)
position_ids = position_stack.rope_position_ids
runtime_attention_mask = (
torch.cat(
(
shared_prefix_mask.to(dtype=active_attention_mask.dtype),
active_attention_mask,
),
dim=1,
)
if shared_prefix_active and shared_prefix_mask is not None
else active_attention_mask
)
prefill: NativeTiledPrefill | None = None
parent_outputs_consumed_by_autograd = torch.is_grad_enabled()
parent_forward_inference_boundary = torch.is_inference_mode_enabled()
# The parent is frozen. ``no_grad`` avoids retaining its activation
# tape while preserving ordinary tensor identity for historical
# recurrent/session buffers. ``inference_mode`` was marginally cheaper
# inside the parent, but forced a full module-buffer traversal plus
# output clones after every training wave.
with torch.no_grad():
if use_past:
# Decode continuation: one new token, one KV telemetry record.
if callable(record_kv):
record_kv(
prefix_positions=result_prefix_positions,
new_positions=result_new_positions,
)
elif not self.exact_historical_source_loaded:
raise RuntimeError(
"Resynthesis integrated runtime lacks tensor-native KV telemetry"
)
output = runtime(
input_ids=active_input_ids,
position_ids=position_ids,
attention_mask=runtime_attention_mask,
past_key_values=(
shared_prefix_cache
if shared_prefix_active
else self._decode_past_key_values
),
output_hidden_states=True,
return_dict=True,
use_cache=not shared_prefix_active,
logits_to_keep=1,
)
else:
# Native 4M+ tiled prefill: every token executes the parent.
# The trained linear-attention state remains global; the full-
# attention K/V becomes a bounded hot window only beyond the
# parent's pretrained band. No token is dropped or truncated.
transient_training_prefill = masked_training_prefill
if transient_training_prefill and not explicit_attention_mask:
raise RuntimeError(
"masked training prefill requires an explicit attention mask"
)
prefill_boundary = (
self._prefill_tiled_training_native
if transient_training_prefill
else self._prefill_tiled_native
)
prefill = prefill_boundary(
runtime=runtime,
active_input_ids=active_input_ids,
position_ids=position_ids,
attention_mask=runtime_attention_mask,
record_kv=record_kv,
starting_prefix=prefix_positions,
telemetry_required=not self.exact_historical_source_loaded,
)
output = prefill.runtime_output
# The external parent owns recurrent/session buffers that may be
# updated in its inference-mode forward. Normalize those buffers
# before the next RBO session reset, while the parent runtime is still
# attached and before any trainable graph consumes the result.
parent_rbo = getattr(runtime, "resynthesis_rbo", None)
if (
isinstance(parent_rbo, nn.Module)
and (
self._parent_session_may_hold_inference_tensors
or parent_forward_inference_boundary
)
):
normalized = self._normalize_inference_session_buffers(parent_rbo)
with torch.no_grad():
self._parent_inference_session_buffers_normalized.add_(
normalized.to(
device=self._parent_inference_session_buffers_normalized.device
)
)
# The traversal above is exhaustive over registered historical
# session buffers. Its completion proves ordinary tensor identity;
# extracting ``normalized`` as a Python scalar here only adds a
# host boundary to every inference-to-training transition.
self._parent_session_may_hold_inference_tensors = False
self._parent_session_buffers_verified_normal = True
returned_cache = getattr(output, "past_key_values", None)
if shared_prefix_active:
if shared_prefix_cache is None:
raise RuntimeError("shared parent-prefix cache was lost")
self._record_native_context_cache_telemetry_boundary(
shared_prefix_cache
)
setattr(output, "past_key_values", None)
self._decode_past_key_values = None
else:
self._decode_past_key_values = returned_cache
if self._decode_past_key_values is not None:
self._record_native_context_cache_telemetry_boundary(
self._decode_past_key_values
)
with torch.no_grad():
if self._decode_past_key_values is None:
self._decode_cached_positions.zero_()
else:
self._decode_cached_positions.add_(
runtime_new_positions.to(
device=self._decode_cached_positions.device
)
)
logits = getattr(output, "logits", None)
hidden_states = getattr(output, "hidden_states", None)
if not isinstance(logits, torch.Tensor) or not hidden_states:
raise RuntimeError(
"Resynthesis integrated forward returned no hidden/logit tensors"
)
parent_result = _resynthesis_parent_last_rbo_result(runtime)
shaped_hidden = getattr(parent_result, "shaped_hidden", None)
hidden = (
shaped_hidden
if isinstance(shaped_hidden, torch.Tensor)
else hidden_states[-1]
)
if hidden.shape[:-1] != logits.shape[:-1]:
raise RuntimeError(
"Resynthesis integrated hidden/logit geometry differs: "
f"hidden={tuple(hidden.shape)} logits={tuple(logits.shape)}"
)
parent_context_hidden = _parent_hidden_context(hidden)
if shared_prefix_active:
suffix_hidden = hidden_states[-1]
if (
not isinstance(suffix_hidden, torch.Tensor)
or suffix_hidden.ndim != 3
or suffix_hidden.shape[1] != 1
):
raise RuntimeError(
"shared parent-prefix final hidden geometry differs"
)
parent_prefill_hidden = suffix_hidden
assert shared_full_input_positions_t is not None
parent_prefill_input_positions = shared_full_input_positions_t
elif prefill is None:
parent_prefill_hidden = hidden.new_empty(
hidden.shape[0],
0,
hidden.shape[-1],
)
parent_prefill_input_positions = runtime_new_positions.new_zeros(())
else:
parent_prefill_hidden = prefill.summary_hidden.to(
device=hidden.device,
dtype=hidden.dtype,
)
parent_prefill_input_positions = prefill.input_positions.to(
device=hidden.device,
dtype=torch.long,
)
if masked_training_prefill and not use_past:
raw_final_hidden_t = hidden_states[-1]
if (
not isinstance(raw_final_hidden_t, torch.Tensor)
or raw_final_hidden_t.ndim != 3
or raw_final_hidden_t.shape[1] != 1
or parent_prefill_hidden.ndim != 3
or parent_prefill_hidden.shape[0]
!= raw_final_hidden_t.shape[0]
or parent_prefill_hidden.shape[2]
!= raw_final_hidden_t.shape[2]
):
raise RuntimeError(
"frozen-backbone captured feature geometry differs"
)
input_positions_t = parent_prefill_input_positions.reshape(-1)
if input_positions_t.numel() == 1:
input_positions_t = input_positions_t.expand(
raw_final_hidden_t.shape[0]
)
if input_positions_t.shape != (raw_final_hidden_t.shape[0],):
raise RuntimeError(
"frozen-backbone captured input positions differ"
)
self._last_frozen_backbone_final_hidden_t = (
raw_final_hidden_t[:, -1:, :].detach()
)
self._last_frozen_backbone_summary_hidden_t = (
parent_prefill_hidden.detach()
)
self._last_frozen_backbone_input_positions_t = (
input_positions_t.detach().clone()
)
# Resynthesis has already causally contextualized the complete prefill.
# The appended autoregressive RBO stack consumes the current position,
# just as it does for every cached continuation. This preserves the
# whole parent context without re-running four 4096-wide GRU/attention
# expert layers over historical positions on every next-token decision.
hidden = hidden[:, -1:, :]
logits = logits[:, -1:, :]
parent_rbo = getattr(runtime, "resynthesis_rbo", None)
expert_routes = getattr(parent_rbo, "_last_none_actual_expert_routes", None)
layer_routes = getattr(parent_rbo, "_last_none_actual_layer_routes", None)
if not isinstance(expert_routes, torch.Tensor):
expert_routes = hidden.new_empty(0)
if not isinstance(layer_routes, torch.Tensor):
layer_routes = hidden.new_empty(0)
result = ResynthesisParentForward(
hidden=(
_regularize_parent_tensor_for_autograd_boundary(hidden)
if parent_outputs_consumed_by_autograd
else hidden.detach()
),
logits=(
_regularize_parent_tensor_for_autograd_boundary(logits)
if parent_outputs_consumed_by_autograd
else logits.detach()
),
parent_context_hidden=(
_regularize_parent_tensor_for_autograd_boundary(
parent_context_hidden
)
if parent_outputs_consumed_by_autograd
else parent_context_hidden.detach()
),
parent_expert_routes=(
_regularize_parent_tensor_for_autograd_boundary(expert_routes)
if parent_outputs_consumed_by_autograd
else expert_routes.detach()
),
parent_layer_routes=(
_regularize_parent_tensor_for_autograd_boundary(layer_routes)
if parent_outputs_consumed_by_autograd
else layer_routes.detach()
),
kv_prefix_positions=(
_regularize_parent_tensor_for_autograd_boundary(
result_prefix_positions
)
if parent_outputs_consumed_by_autograd
else result_prefix_positions.detach()
),
kv_new_positions=(
_regularize_parent_tensor_for_autograd_boundary(
result_new_positions
)
if parent_outputs_consumed_by_autograd
else result_new_positions.detach()
),
parent_prefill_hidden=(
_regularize_parent_tensor_for_autograd_boundary(
parent_prefill_hidden
)
if parent_outputs_consumed_by_autograd
else parent_prefill_hidden.detach()
),
parent_prefill_input_positions=(
_regularize_parent_tensor_for_autograd_boundary(
parent_prefill_input_positions
)
if parent_outputs_consumed_by_autograd
else parent_prefill_input_positions.detach()
),
)
self._last_forward = result
return result
def begin_decode(self) -> None:
"""Reset frozen feature/KV plumbing without granting parent authority."""
self._decode_past_key_values = None
self._decode_authority_active = False
self._last_forward = None
with torch.no_grad():
self._decode_cached_positions.zero_()
self._native_context_hybrid_active.zero_()
self._native_context_total_positions.zero_()
self._native_context_hot_resident_positions.zero_()
self._native_context_hot_window_tokens.zero_()
self._native_context_recurrent_layer_count.zero_()
self._native_context_full_attention_layer_count.zero_()
@staticmethod
def _normalize_inference_session_buffers(parent_rbo: nn.Module) -> torch.Tensor:
"""Restore ordinary tensor identity at the inference/training boundary.
The verified historical parent predates its later recurrent-buffer
repair. An inference-mode decode can therefore replace a registered
session buffer with an inference tensor. Such a tensor cannot be reset
by the subsequent training transaction outside inference mode. This is
a caller-session compatibility boundary: it clones only registered
buffers that actually carry inference identity and leaves parameters,
trained values, routing, and immutable parent source untouched.
"""
normalized = torch.zeros((), dtype=torch.long)
with torch.inference_mode(False):
for module in parent_rbo.modules():
for name, buffer in module.named_buffers(recurse=False):
if buffer.is_inference():
setattr(module, name, buffer.detach().clone())
normalized.add_(torch.ones_like(normalized))
return normalized
def begin_session(self) -> None:
"""Reset caller-owned parent state without changing trained tensors."""
if not self._weights_loaded:
self.load_weights()
runtime = self.runtime
parent_rbo = getattr(runtime, "resynthesis_rbo", None)
reset_session = getattr(parent_rbo, "reset_session_state", None)
if not isinstance(parent_rbo, nn.Module) or not callable(reset_session):
raise RuntimeError("Resynthesis integrated RBO lacks caller-session reset")
inference_mode = torch.is_inference_mode_enabled()
if not inference_mode and (
self._parent_session_may_hold_inference_tensors
or not self._parent_session_buffers_verified_normal
):
normalized = self._normalize_inference_session_buffers(parent_rbo)
with torch.no_grad():
self._parent_inference_session_buffers_normalized.add_(
normalized.to(
device=self._parent_inference_session_buffers_normalized.device
)
)
self._parent_session_may_hold_inference_tensors = False
self._parent_session_buffers_verified_normal = True
reset_session()
if inference_mode:
self._parent_session_may_hold_inference_tensors = True
self._parent_session_buffers_verified_normal = False
self.begin_decode()
def session_transition_receipt_boundary(self) -> dict[str, object]:
"""Serialize inference-to-training compatibility only at log boundary."""
return {
"schema": "nnf.resynthesis.parent_session_transition.v1",
"normalizedInferenceSessionBuffers": int(
self._parent_inference_session_buffers_normalized.detach()
.to(device="cpu", dtype=torch.long)
.reshape(())
),
"parentSessionMayHoldInferenceTensors": (
self._parent_session_may_hold_inference_tensors
),
"parentSessionBuffersVerifiedNormal": (
self._parent_session_buffers_verified_normal
),
"trainedParametersChangedByTransition": False,
"immutableParentSourceChanged": False,
"baseForwardCacheOwner": "ResynthesisRBO",
}
def forward_hidden(self, input_ids: torch.Tensor) -> torch.Tensor:
"""Compatibility boundary for callers that only consume parent hidden."""
return self.forward_hidden_logits(input_ids).hidden
def forward_logits(self, hidden: torch.Tensor) -> torch.Tensor:
"""Project through Resynthesis's frozen native RBO-owned generation head.
Parent parameters remain frozen, while autograd is retained with respect
to ``hidden`` so the appended Resynthesis experts receive real CE signal.
"""
if not self._weights_loaded:
self.load_weights()
runtime = self.runtime
if runtime is None:
raise RuntimeError("Resynthesis integrated runtime was not attached")
head = getattr(getattr(runtime, "backbone", None), "lm_head", None)
if not isinstance(head, nn.Module):
raise RuntimeError("Resynthesis integrated runtime has no native generation head")
logits = head(hidden)
if not isinstance(logits, torch.Tensor):
raise RuntimeError(
"Resynthesis native generation head returned a non-tensor value"
)
return logits
def forward_low_rank_logits_projection(
self,
hidden_projection: torch.Tensor,
) -> torch.Tensor:
"""Compose the frozen native head with a trainable hidden projection.
For ``U: [hidden, rank]`` this returns ``W_head @ U`` in FP32. The
full-vocabulary result and gradient are exact; only the associative
order of the two linear projections changes. The detached forward value
is reused across waves until the optimizer mutates ``U``. Each wave
attaches an exact tiled ``W_head.T @ grad`` backward, so reuse never
severs the trainable correction path.
"""
if not self._weights_loaded:
self.load_weights()
runtime = self.runtime
if runtime is None:
raise RuntimeError("Resynthesis integrated runtime was not attached")
head = getattr(getattr(runtime, "backbone", None), "lm_head", None)
native_rbo = getattr(head, "_resynthesis_rbo", None)
frozen_weight = getattr(native_rbo, "_frozen_lm_weight", None)
if (
not isinstance(head, nn.Module)
or getattr(head, "training_rbo_active", False) is not True
or not isinstance(frozen_weight, torch.Tensor)
):
raise RuntimeError(
"Resynthesis native linear head is unavailable for low-rank completion"
)
if (
hidden_projection.ndim != 2
or hidden_projection.shape[0] != frozen_weight.shape[1]
or hidden_projection.device != frozen_weight.device
):
raise ValueError("low-rank completion projection geometry differs")
source_id = id(hidden_projection)
source_version = hidden_projection._version
projection = self._training_low_rank_projection_cache
if (
projection is None
or self._training_low_rank_projection_source_id != source_id
or self._training_low_rank_projection_source_version != source_version
or projection.device != hidden_projection.device
or projection.shape
!= (frozen_weight.shape[0], hidden_projection.shape[1])
):
projection = hidden_projection.new_empty(
(frozen_weight.shape[0], hidden_projection.shape[1]),
dtype=torch.float32,
)
projection_fp32 = hidden_projection.detach().float()
with torch.no_grad():
for start in range(
0,
frozen_weight.shape[0],
_FROZEN_HEAD_PROJECTION_TILE_ROWS,
):
end = min(
start + _FROZEN_HEAD_PROJECTION_TILE_ROWS,
frozen_weight.shape[0],
)
projection[start:end] = torch.mm(
frozen_weight[start:end].float(),
projection_fp32,
)
self._training_low_rank_projection_cache = projection
self._training_low_rank_projection_source_id = source_id
self._training_low_rank_projection_source_version = source_version
return cast(
torch.Tensor,
_CachedFrozenHeadLowRankProjection.apply( # type: ignore[no-untyped-call]
hidden_projection,
frozen_weight,
projection,
),
)
def native_decode_stop(
self,
hidden: torch.Tensor,
logits: torch.Tensor,
generated_ids: torch.Tensor,
) -> ResynthesisNativeDecodeStop:
"""Observe the exact parent's legacy completion surface diagnostically.
Only the active final hidden/logit position is scored. ``generated_ids``
contains model-emitted continuation tokens, never the immutable prompt.
No EOS, host token count, or RBO traversal heuristic participates. The
returned tensors have no answer, stop, retention, or veto authority;
the additive Resynthesis completion graph owns those decisions.
"""
if not self._weights_loaded:
self.load_weights()
runtime = self.runtime
if runtime is None:
raise RuntimeError("Resynthesis integrated runtime was not attached")
surface = getattr(runtime, "native_decode_confidence", None)
if not isinstance(surface, nn.Module):
raise RuntimeError(
"Resynthesis parent has no trained native decode-confidence surface"
)
if (
hidden.ndim != 3
or logits.ndim != 3
or hidden.shape[:2] != logits.shape[:2]
or generated_ids.ndim != 2
or hidden.shape[0] != generated_ids.shape[0]
or generated_ids.shape[1] < 1
):
raise RuntimeError("Resynthesis native decode-confidence input geometry differs")
with torch.no_grad():
raw_score = surface(
hidden=hidden[:, -1:, :],
logits=logits[:, -1:, :],
generated_ids=generated_ids,
)
if not isinstance(raw_score, torch.Tensor):
raise RuntimeError(
"Resynthesis native decode-confidence surface returned no tensor"
)
score = raw_score.reshape(hidden.shape[0], -1)[:, -1]
probability = score.sigmoid()
decision = probability.ge(0.5)
return ResynthesisNativeDecodeStop(
score=score,
probability=probability,
decision=decision,
)
def apply_native_answer_surface(
self,
hidden: torch.Tensor,
logits: torch.Tensor,
) -> ResynthesisNativeAnswer:
"""Encode additive logits with the parent's immutable vocabulary map.
``hidden`` is accepted only to preserve the tensor-native historical
interface. It cannot enter answer selection. In particular, this
boundary deliberately does not call the frozen parent's active RBO or
answer head: Resynthesis has already produced the authoritative logits,
and the parent supplies only token-to-bit/glyph vocabulary plumbing.
"""
del hidden
runtime = self.runtime
if runtime is None:
raise RuntimeError("Resynthesis integrated runtime was not attached")
vge = getattr(runtime, "resynthesis_vge", None)
token_to_glyph = getattr(vge, "token_ids_to_glyph", None)
token_to_bits = getattr(vge, "token_ids_to_bit_ids", None)
if not callable(token_to_glyph) or not callable(token_to_bits):
raise RuntimeError("frozen parent VGE lacks its immutable bit mapping")
if logits.ndim != 3 or logits.shape[1] < 1:
raise RuntimeError("additive vocabulary logits geometry differs")
token_ids = logits[:, -1:, :].argmax(dim=-1)
glyph = token_to_glyph(token_ids)
bit_ids = token_to_bits(token_ids)
if not all(
isinstance(value, torch.Tensor)
for value in (token_ids, glyph, bit_ids)
):
raise RuntimeError("frozen parent VGE returned a non-tensor packet")
return ResynthesisNativeAnswer(
logits=logits,
token_ids=token_ids,
bit_ids=bit_ids,
glyph=glyph,
)
def native_token_glyphs_loss_boundary(
self,
token_ids_t: torch.Tensor,
) -> torch.Tensor:
"""Map token identities through immutable vocabulary plumbing only.
Object-native knowledge absorption already owns the verified token
identities at its explicit loss boundary. Re-running the frozen 12B
decoder merely to recover their vocabulary coordinates would make that
historical parent the training bottleneck again. This boundary calls
only the parent's immutable VGE lookup: it performs no parent forward,
routing, answer selection, teacher projection, or parameter update.
The same interface maps target-free context IDs and verified value IDs;
callers remain responsible for keeping value glyphs on the loss side
so ground-truth-derived tensors never enter ``forward_thinking``.
"""
if (
token_ids_t.ndim < 1
or token_ids_t.dtype not in (torch.int32, torch.int64)
):
raise ValueError(
"native vocabulary glyph lookup requires integer token tensors"
)
runtime = self.runtime
if runtime is None:
raise RuntimeError("Resynthesis integrated runtime was not attached")
vge = getattr(runtime, "resynthesis_vge", None)
token_to_glyph = getattr(vge, "token_ids_to_glyph", None)
if not callable(token_to_glyph):
raise RuntimeError(
"frozen parent VGE lacks its immutable glyph mapping"
)
glyph_t = token_to_glyph(token_ids_t)
if (
not isinstance(glyph_t, torch.Tensor)
or glyph_t.ndim != token_ids_t.ndim + 1
or glyph_t.shape[:-1] != token_ids_t.shape
or glyph_t.shape[-1] < 1
or not glyph_t.is_floating_point()
or glyph_t.device != token_ids_t.device
):
raise RuntimeError(
"frozen parent VGE returned an invalid glyph tensor"
)
torch._assert_async(
torch.isfinite(glyph_t).all(),
"frozen parent VGE returned non-finite glyph coordinates",
)
return glyph_t
def apply_execution_outcome(
self,
outcome_features: torch.Tensor,
) -> ResynthesisOutcomeApplication:
"""Apply one persisted verifier/tool outcome to the exact parent route.
The eight Resynthesis fields are observations only. They are translated
to Resynthesis's fixed execution-memory geometry without fabricating unknown
code-specific evidence such as patch application or timeout status.
"""
if outcome_features.shape != (1, 8):
raise ValueError("one parent execution outcome must have shape [1, 8]")
runtime = self.runtime
if runtime is None:
raise RuntimeError("Resynthesis integrated runtime was not attached")
apply_outcome = getattr(
runtime,
"resynthesis_apply_current_turn_execution_outcome",
None,
)
observed = outcome_features.detach().to(device=self.device, dtype=torch.float32)
passed = observed[0, 0].clamp(0.0, 1.0)
failed = observed[0, 1].clamp(0.0, 1.0)
score = observed[0, 2].clamp(0.0, 1.0)
coverage = observed[0, 3].clamp(0.0, 1.0)
verified = observed[0, 4].clamp(0.0, 1.0)
infrastructure_failure = observed[0, 6].clamp(0.0, 1.0)
zero = observed.new_zeros(())
observation = torch.stack(
(
zero,
verified,
passed,
torch.maximum(failed, infrastructure_failure),
zero,
failed * verified,
zero,
zero,
score,
coverage,
)
)
acquisition = torch.stack(
(
failed * verified,
zero,
zero,
passed * verified,
)
)
if not callable(apply_outcome):
if not self.exact_historical_source_loaded:
raise RuntimeError(
"Resynthesis integrated runtime lacks its trained outcome boundary"
)
parent_rbo = getattr(runtime, "resynthesis_rbo", None)
capture = getattr(parent_rbo, "capture_session_state", None)
apply_historical = getattr(
parent_rbo,
"apply_execution_observation_from_boundary",
None,
)
if not callable(capture) or not callable(apply_historical):
raise RuntimeError(
"historical Resynthesis RBO lacks outcome-memory boundaries"
)
parent_result = _resynthesis_parent_last_rbo_result(runtime)
domain_idx = int(getattr(parent_result, "domain_idx", 0))
subdomain_idx = int(getattr(parent_result, "subdomain_idx", 0))
before = capture()
apply_historical(
observation,
domain_idx=domain_idx,
subdomain_idx=subdomain_idx,
resolved=bool(passed.eq(1.0).detach().to(device="cpu")),
)
after = capture()
before_values = vars(before)
after_values = vars(after)
if set(before_values) != set(after_values):
raise RuntimeError("historical Resynthesis outcome-state geometry changed")
squared = observation.new_zeros((), dtype=torch.float32)
for name in sorted(before_values):
before_value = before_values[name]
after_value = after_values[name]
if isinstance(before_value, torch.Tensor):
if not isinstance(after_value, torch.Tensor) or (
before_value.shape != after_value.shape
):
raise RuntimeError(
"historical Resynthesis outcome-state tensor geometry differs"
)
delta = after_value.detach().to(
device=observation.device,
dtype=torch.float32,
) - before_value.detach().to(
device=observation.device,
dtype=torch.float32,
)
squared = squared + delta.square().sum()
rbo_delta = squared.sqrt()
route_changed = rbo_delta.gt(0)
if not bool(route_changed.detach().to(device="cpu")):
raise RuntimeError(
"persisted outcome did not change historical RBO state"
)
zero_delta = rbo_delta.new_zeros(())
return ResynthesisOutcomeApplication(
applied=route_changed,
rbo_state_delta_l2=rbo_delta,
arm_state_delta_l1=zero_delta,
legacy_state_delta_l2=zero_delta,
route_state_changed=route_changed,
)
packet = apply_outcome(
observation,
acquisition=acquisition,
resolved=passed.eq(1.0),
)
packet_applied = getattr(packet, "applied", None)
packet_rbo_delta = getattr(packet, "rbo_state_delta_l2", None)
packet_arm_delta = getattr(packet, "arm_state_delta_l1", None)
packet_legacy_delta = getattr(packet, "legacy_state_delta_l2", None)
packet_route_changed = getattr(packet, "route_state_changed", None)
if not all(
isinstance(value, torch.Tensor)
for value in (
packet_applied,
packet_rbo_delta,
packet_arm_delta,
packet_legacy_delta,
packet_route_changed,
)
):
raise RuntimeError(
"Resynthesis outcome boundary returned an invalid tensor packet"
)
assert isinstance(packet_applied, torch.Tensor)
assert isinstance(packet_rbo_delta, torch.Tensor)
assert isinstance(packet_arm_delta, torch.Tensor)
assert isinstance(packet_legacy_delta, torch.Tensor)
assert isinstance(packet_route_changed, torch.Tensor)
return ResynthesisOutcomeApplication(
applied=packet_applied,
rbo_state_delta_l2=packet_rbo_delta,
arm_state_delta_l1=packet_arm_delta,
legacy_state_delta_l2=packet_legacy_delta,
route_state_changed=packet_route_changed,
)
def current_acquisition_policy(self) -> ResynthesisAcquisitionPolicy:
"""Read the trained parent action after a real outcome was ingested."""
runtime = self.runtime
if runtime is None:
raise RuntimeError("Resynthesis integrated runtime was not attached")
active_rbo_fn = getattr(runtime, "_active_resynthesis_rbo", None)
if not callable(active_rbo_fn):
raise RuntimeError("Resynthesis integrated runtime has no active RBO boundary")
active_rbo = active_rbo_fn(required=True)
execution_grounding = getattr(active_rbo, "execution_grounding", None)
policy_fn = getattr(execution_grounding, "acquisition_policy", None)
authority = getattr(execution_grounding, "acquisition_policy_authority", None)
if not callable(policy_fn) or not isinstance(authority, torch.Tensor):
raise RuntimeError("Resynthesis RBO has no trained evidence-acquisition policy")
policy = policy_fn(device=self.device, dtype=torch.float32)
action_probs = getattr(policy, "action_probs", None)
action_index = getattr(policy, "action_index", None)
confidence = getattr(policy, "confidence", None)
observation_count = getattr(policy, "observation_count", None)
acquisition_count = getattr(policy, "acquisition_count", None)
values = (
action_probs,
action_index,
confidence,
observation_count,
acquisition_count,
)
if not all(isinstance(value, torch.Tensor) for value in values):
raise RuntimeError(
"Resynthesis acquisition policy returned an invalid tensor packet"
)
assert isinstance(action_probs, torch.Tensor)
assert isinstance(action_index, torch.Tensor)
assert isinstance(confidence, torch.Tensor)
assert isinstance(observation_count, torch.Tensor)
assert isinstance(acquisition_count, torch.Tensor)
if action_probs.shape != (4,):
raise RuntimeError("Resynthesis acquisition action geometry differs")
return ResynthesisAcquisitionPolicy(
action_probs=action_probs.detach(),
action_index=action_index.detach().reshape(()),
confidence=confidence.detach().reshape(()),
observation_count=observation_count.detach().reshape(()),
acquisition_count=acquisition_count.detach().reshape(()),
authority=authority.detach().reshape(()),
)
def freeze(self) -> "ResynthesisNativeParent":
"""Ensure all base parameters are frozen (requires_grad=False)."""
for p in self.parameters():
p.requires_grad_(False)
for b in self.buffers():
b.requires_grad_(False)
return self
# Historical names remain import-compatible, but all factories construct the
# Resynthesis-native owner. They are aliases only and never redirect loading.
FrozenResynthesisParent = ResynthesisNativeParent
FrozenNexumBase = ResynthesisNativeParent
def build_frozen_base(
cfg: ResynthesisConfig | None = None,
*,
device: torch.device | str = "cpu",
lazy: bool = True,
) -> ResynthesisNativeParent:
"""Factory: build the frozen Resynthesis native graph parent.
Args:
cfg: Resynthesis config (defaults to ResynthesisConfig()).
device: torch device.
lazy: if True, weights are loaded on first forward_hidden() call.
if False, weights are loaded immediately.
"""
cfg = cfg or ResynthesisConfig()
base = ResynthesisNativeParent(cfg, device=device)
if not lazy:
base.load_weights()
base.freeze()
return base