"""Finite real-scene engineering smoke for offline dynamic training kernels.""" from __future__ import annotations import json import math import platform import sys from dataclasses import asdict, dataclass, field from pathlib import Path import torch from torch import Tensor from mcss.dynamic.carrier import DynamicSceneCarrier from mcss.dynamic.checkpoint import load_dynamic_checkpoint, save_dynamic_checkpoint from mcss.dynamic.config import CarrierConfig, WriteConfig from mcss.dynamic.types import Action, OnlineObservation, hash_value from mcss.dynamic.write_rule import DirectWriteRule from mcss.measurements import FixedMeasurementRenderer from mcss.training.grounded import grounded_measurement_loss, paired_context_measurement_loss from mcss.training.supervision import TrainingSupervision from mcss.training.write_unroll import unroll_observed_episode @dataclass(frozen=True) class DynamicTrainingSmokeConfig: carrier: CarrierConfig = field(default_factory=CarrierConfig) write: WriteConfig = field(default_factory=WriteConfig) learning_rate: float = 1e-3 renderer_samples: int = 8 ray_chunk_size: int = 2048 def __post_init__(self) -> None: if not math.isfinite(self.learning_rate) or self.learning_rate <= 0: raise ValueError("learning_rate must be finite and positive") if self.renderer_samples < 2 or self.ray_chunk_size < 1: raise ValueError("renderer configuration is invalid") def run_dynamic_training_smoke( warmup: tuple[OnlineObservation, ...], stream: tuple[OnlineObservation, ...], supervision: TrainingSupervision, output_dir: str | Path, *, device: str | torch.device, config: DynamicTrainingSmokeConfig | None = None, resource_adaptation: str | None = None, ) -> dict[str, object]: """Run one grounded step and two ALL-write steps; this is not convergence training.""" warmup = tuple(warmup) stream = tuple(stream) config = config or DynamicTrainingSmokeConfig() _validate_smoke_inputs(warmup, stream, supervision) target_device = torch.device(device) carrier = DynamicSceneCarrier(config.carrier).to(target_device) write_rule = DirectWriteRule(config.carrier, config.write).to(target_device) renderer = FixedMeasurementRenderer( n_samples=config.renderer_samples, ray_chunk_size=config.ray_chunk_size, ).to(target_device) optimizer = torch.optim.Adam( [*carrier.parameters(), *write_rule.parameters()], lr=config.learning_rate ) records: list[dict[str, object]] = [] optimizer.zero_grad(set_to_none=True) paired = paired_context_measurement_loss( carrier, warmup[:2], warmup[2:], supervision.frame_ids, supervision.query_cameras, supervision.query_rgb, supervision.query_depth, renderer, ) records.append( _backward_and_step( name="grounded_context_pair", loss=paired.loss, carrier=carrier, write_rule=write_rule, optimizer=optimizer, ) ) for index in range(2): optimizer.zero_grad(set_to_none=True) state, fast, _, anchor_c2w = unroll_observed_episode( carrier, write_rule, warmup, stream, (Action.ALL, Action.ALL), f"{supervision.episode_id}:write-smoke-{index}", ) loss, _ = grounded_measurement_loss( state, supervision.query_cameras, supervision.query_rgb, supervision.query_depth, anchor_c2w, renderer, ) record = _backward_and_step( name=f"write_unroll_all_{index + 1}", loss=loss, carrier=carrier, write_rule=write_rule, optimizer=optimizer, ) if ( torch.count_nonzero(fast.delta_fuse) == 0 or torch.count_nonzero(fast.delta_complete) == 0 ): raise RuntimeError("ALL unroll did not produce nonzero fast writes") if float(record["write_gradient_norm"]) <= 0: raise RuntimeError("ALL unroll did not produce a write-rule gradient") if record["write_rule_hash_before"] == record["write_rule_hash_after"]: raise RuntimeError("ALL unroll optimizer step did not change write-rule parameters") record["fast_fuse_norm"] = float(fast.delta_fuse.norm().detach().cpu()) record["fast_complete_norm"] = float(fast.delta_complete.norm().detach().cpu()) records.append(record) output = Path(output_dir) output.mkdir(parents=True, exist_ok=True) checkpoint_path = output / "dynamic_training_smoke.pt" checkpoint_hash = save_dynamic_checkpoint( checkpoint_path, carrier, write_rule, phase="engineering-training-smoke-untrained", provenance={ "episode_id": supervision.episode_id, "scene_id": supervision.scene_id, "split_id": "train", "query_vault_id": supervision.query_vault_id, "warmup_frame_ids": [observation.frame_id for observation in warmup], "stream_frame_ids": [observation.frame_id for observation in stream], "query_frame_ids": list(supervision.frame_ids), "purpose": "finite engineering smoke; not convergence training", }, ) restored_carrier, restored_rule, metadata = load_dynamic_checkpoint( checkpoint_path, target_device ) original_state_hash = _unroll_state_hash( carrier, write_rule, warmup, stream, supervision.episode_id ) restored_state_hash = _unroll_state_hash( restored_carrier, restored_rule, warmup, stream, supervision.episode_id ) if original_state_hash != restored_state_hash: raise RuntimeError("strict checkpoint roundtrip changed the unroll output state") report = { "status": "untrained-engineering-smoke", "converged": False, "steps_completed": len(records), "config": asdict(config), "resource_adaptation": resource_adaptation, "episode": { "episode_id": supervision.episode_id, "scene_id": supervision.scene_id, "split_id": "train", "warmup_frame_ids": [observation.frame_id for observation in warmup], "stream_frame_ids": [observation.frame_id for observation in stream], "query_frame_ids": list(supervision.frame_ids), "image_size": list(warmup[0].camera.image_size), }, "steps": records, "checkpoint": { "path": str(checkpoint_path.resolve()), "sha256": checkpoint_hash, "roundtrip_metadata": metadata, "original_unroll_state_hash": original_state_hash, "restored_unroll_state_hash": restored_state_hash, }, "environment": _environment(target_device), } _write_json(output / "config.json", asdict(config)) _write_json(output / "report.json", report) return report def _validate_smoke_inputs( warmup: tuple[OnlineObservation, ...], stream: tuple[OnlineObservation, ...], supervision: TrainingSupervision, ) -> None: if len(warmup) != 4 or len(stream) != 2: raise ValueError("training smoke requires exactly four warmup and two stream observations") if len(supervision.frame_ids) != 4: raise ValueError("training smoke requires exactly four query frames") if warmup[0].scene_id != supervision.scene_id: raise ValueError("training supervision scene does not match the online observations") observed_ids = {observation.frame_id for observation in (*warmup, *stream)} if observed_ids & set(supervision.frame_ids): raise ValueError("training query frames overlap observed frames") if supervision.query_cameras.device != warmup[0].rgb.device: raise ValueError("training supervision and online observations must share a device") def _backward_and_step(*, name, loss: Tensor, carrier, write_rule, optimizer) -> dict[str, object]: if loss.ndim != 0 or not torch.isfinite(loss): raise RuntimeError(f"{name} produced a nonfinite scalar loss") pre_hashes = _parameter_hashes(carrier, write_rule) loss.backward() carrier_gradient_norm = _gradient_norm(carrier.parameters(), f"{name} carrier") write_gradient_norm = _gradient_norm(write_rule.parameters(), f"{name} write rule") optimizer.step() post_hashes = _parameter_hashes(carrier, write_rule) if pre_hashes == post_hashes: raise RuntimeError(f"{name} optimizer step did not change slow parameters") return { "name": name, "loss": float(loss.detach().cpu()), "carrier_gradient_norm": carrier_gradient_norm, "write_gradient_norm": write_gradient_norm, "carrier_hash_before": pre_hashes["carrier"], "carrier_hash_after": post_hashes["carrier"], "write_rule_hash_before": pre_hashes["write_rule"], "write_rule_hash_after": post_hashes["write_rule"], } def _gradient_norm(parameters, name: str) -> float: gradients = [parameter.grad for parameter in parameters if parameter.grad is not None] if not gradients: return 0.0 if not all(torch.isfinite(gradient).all() for gradient in gradients): raise RuntimeError(f"{name} has nonfinite gradients") squared_norm = sum(gradient.detach().float().square().sum() for gradient in gradients) return float(torch.sqrt(squared_norm).cpu()) def _parameter_hashes(carrier, write_rule) -> dict[str, str]: return { "carrier": hash_value(carrier.state_dict()), "write_rule": hash_value(write_rule.state_dict()), } def _unroll_state_hash(carrier, write_rule, warmup, stream, episode_id: str) -> str: with torch.no_grad(): state, _, _, _ = unroll_observed_episode( carrier, write_rule, warmup, stream, (Action.ALL, Action.ALL), f"{episode_id}:roundtrip", ) return hash_value(state) def _environment(device: torch.device) -> dict[str, object]: return { "device": str(device), "python": sys.version, "platform": platform.platform(), "torch": torch.__version__, "cuda_available": torch.cuda.is_available(), "cuda_version": torch.version.cuda, "cuda_device_name": torch.cuda.get_device_name(device) if device.type == "cuda" else None, } def _write_json(path: Path, value: object) -> None: path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8")