Spaces:
Paused
Paused
| """Rebuild the v2 reference component from explicit, hash-bound raw events. | |
| This is not an aggregate-v1 migration or a neural/runtime checkpoint restore. | |
| Bounded episode exports cannot substitute for a complete input/control history. | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import hashlib | |
| import json | |
| import math | |
| from pathlib import Path | |
| from typing import Any | |
| from src.embodiment.models import ActionCommand, EnvironmentObservation, SensorFrame | |
| from src.memory.layer import MemoryWorldModel, MemoryWorldModelError | |
| from src.memory.store import MemoryStore | |
| from src.memory.world_model import TransitionWorldModel | |
| MAX_INPUT_BYTES = 8 * 1024 * 1024 | |
| MAX_EVENTS = 100_000 | |
| FLAGS = ("read_enabled", "write_enabled", "prediction_enabled", "learning_enabled") | |
| CAPACITIES = ( | |
| "episode_capacity", | |
| "working_capacity", | |
| "prediction_capacity", | |
| "retention_ticks", | |
| "max_contexts", | |
| ) | |
| class ReplayError(ValueError): | |
| """The supplied history cannot safely rebuild a reference state.""" | |
| def _object(value: Any, keys: set[str], name: str) -> dict[str, Any]: | |
| if not isinstance(value, dict) or set(value) != keys: | |
| raise ReplayError(f"{name} requires exactly {sorted(keys)}") | |
| return value | |
| def _text(value: Any, name: str) -> str: | |
| if not isinstance(value, str) or not value.strip(): | |
| raise ReplayError(f"{name} must be a non-empty string") | |
| return value | |
| def _integer(value: Any, name: str, minimum: int = 0) -> int: | |
| if type(value) is not int or not minimum <= value <= MAX_EVENTS: | |
| raise ReplayError(f"{name} must be an integer in [{minimum}, {MAX_EVENTS}]") | |
| return value | |
| def _flags(value: Any) -> dict[str, bool]: | |
| raw = _object(value, set(FLAGS), "controls") | |
| if any(type(item) is not bool for item in raw.values()): | |
| raise ReplayError("control values must be booleans") | |
| return dict(raw) | |
| def _unique(pairs: list[tuple[str, Any]]) -> dict[str, Any]: | |
| result: dict[str, Any] = {} | |
| for key, value in pairs: | |
| if key in result: | |
| raise ReplayError(f"duplicate JSON key: {key}") | |
| result[key] = value | |
| return result | |
| def _invalid_constant(value: str) -> None: | |
| raise ReplayError(f"non-finite JSON constant: {value}") | |
| def rebuild_reference(source: Path, destination: Path, expected_sha256: str) -> Path: | |
| """Replay from reset into a NEW directory; never rewrite a source or snapshot. | |
| The caller supplies the expected hash independently. The completeness flag is | |
| a source declaration, not independent evidence of the original run's history. | |
| """ | |
| if destination.exists() or destination.is_symlink(): | |
| raise FileExistsError("rebuild requires a new destination directory") | |
| with source.open("rb") as stream: | |
| data = stream.read(MAX_INPUT_BYTES + 1) | |
| if len(data) > MAX_INPUT_BYTES: | |
| raise ReplayError("replay exceeds the input byte budget") | |
| digest = hashlib.sha256(data).hexdigest() | |
| if len(expected_sha256) != 64 or digest != expected_sha256.lower(): | |
| raise ReplayError("replay source SHA-256 mismatch") | |
| try: | |
| raw = json.loads( | |
| data, object_pairs_hook=_unique, parse_constant=_invalid_constant | |
| ) | |
| # Also reject overflow such as 1e999, including nested sensor payloads. | |
| json.dumps(raw, allow_nan=False) | |
| packet = _object( | |
| raw, | |
| { | |
| "kind", | |
| "schema_version", | |
| "run_id", | |
| "complete_from_reset", | |
| "configuration", | |
| "events", | |
| }, | |
| "replay", | |
| ) | |
| if ( | |
| packet["kind"] != "mhrn.cognition.replay" | |
| or type(packet["schema_version"]) is not int | |
| or packet["schema_version"] != 1 | |
| ): | |
| raise ReplayError("unsupported replay protocol") | |
| if packet["complete_from_reset"] is not True: | |
| raise ReplayError("a complete history from reset must be declared") | |
| run_id = _text(packet["run_id"], "run_id") | |
| config = _object(packet["configuration"], set(CAPACITIES), "configuration") | |
| sizes = {key: _integer(config[key], key, 1) for key in CAPACITIES} | |
| store = MemoryStore( | |
| run_id=run_id, | |
| episode_capacity=sizes["episode_capacity"], | |
| working_capacity=sizes["working_capacity"], | |
| prediction_capacity=sizes["prediction_capacity"], | |
| retention_ticks=sizes["retention_ticks"], | |
| ) | |
| model = MemoryWorldModel( | |
| store, TransitionWorldModel(max_contexts=sizes["max_contexts"]), run_id | |
| ) | |
| events = packet["events"] | |
| if not isinstance(events, list) or not 1 <= len(events) <= MAX_EVENTS: | |
| raise ReplayError("replay requires a bounded, non-empty event list") | |
| seen: set[str] = set() | |
| episode: str | None = None | |
| last_tick = -1 | |
| ended = False | |
| for raw_event in events: | |
| event = _object( | |
| raw_event, | |
| {"episode_id", "tick", "frame", "action", "observation", "controls"}, | |
| "event", | |
| ) | |
| current = _text(event["episode_id"], "episode_id") | |
| tick = _integer(event["tick"], "tick") | |
| if current != episode: | |
| if current in seen: | |
| raise ReplayError("episode history is reordered") | |
| seen.add(current) | |
| episode, last_tick, ended = current, -1, False | |
| model.reset_episode(current) | |
| if tick != last_tick + 1 or ended: | |
| raise ReplayError( | |
| "ticks must be contiguous from zero within each episode" | |
| ) | |
| controls = _flags(event["controls"]) | |
| store.set_controls( | |
| read_enabled=controls["read_enabled"], | |
| write_enabled=controls["write_enabled"], | |
| ) | |
| model.prediction_enabled = controls["prediction_enabled"] | |
| model.learning_enabled = controls["learning_enabled"] | |
| frame_data = _object( | |
| event["frame"], {"sensor_id", "tick", "modality", "payload"}, "frame" | |
| ) | |
| if _integer(frame_data["tick"], "frame.tick") != tick: | |
| raise ReplayError("sensor and event ticks differ") | |
| frame = SensorFrame( | |
| _text(frame_data["sensor_id"], "sensor_id"), | |
| tick, | |
| _text(frame_data["modality"], "modality"), | |
| frame_data["payload"], | |
| ) | |
| action = None | |
| if event["action"] is not None: | |
| command = _object( | |
| event["action"], | |
| {"actuator_id", "tick", "action", "payload"}, | |
| "action", | |
| ) | |
| if _integer(command["tick"], "action.tick") != tick: | |
| raise ReplayError("action and event ticks differ") | |
| action = ActionCommand( | |
| _text(command["actuator_id"], "actuator_id"), | |
| tick, | |
| _text(command["action"], "action"), | |
| command["payload"], | |
| ) | |
| observation = None | |
| if event["observation"] is not None: | |
| observed = _object( | |
| event["observation"], | |
| {"tick", "state", "reward", "terminated", "truncated"}, | |
| "observation", | |
| ) | |
| if _integer( | |
| observed["tick"], "observation.tick" | |
| ) != tick + 1 or not isinstance(observed["state"], dict): | |
| raise ReplayError( | |
| "expected a next-tick observation with object state" | |
| ) | |
| reward = observed["reward"] | |
| if type(reward) not in (int, float) or not math.isfinite(reward): | |
| raise ReplayError("reward must be finite and numeric") | |
| if ( | |
| type(observed["terminated"]) is not bool | |
| or type(observed["truncated"]) is not bool | |
| ): | |
| raise ReplayError("episode termination flags must be boolean") | |
| observation = EnvironmentObservation( | |
| tick + 1, | |
| observed["state"], | |
| reward, | |
| observed["terminated"], | |
| observed["truncated"], | |
| ) | |
| ended = observation.terminated or observation.truncated | |
| prediction = model.predict(frame, action, tick) | |
| model.complete(frame, action, observation, tick, prediction) | |
| last_tick = tick | |
| except (KeyError, TypeError, ValueError, OverflowError, RecursionError) as error: | |
| if isinstance(error, ReplayError): | |
| raise | |
| raise ReplayError(f"invalid replay history: {error}") from error | |
| # Nothing has been written before the entire input has been validated/replayed. | |
| destination.mkdir(parents=False, exist_ok=False) | |
| state_path = model.save(destination / "state.json") | |
| sources = { | |
| name: hashlib.sha256(Path(__file__).with_name(name).read_bytes()).hexdigest() | |
| for name in ( | |
| "replay.py", | |
| "layer.py", | |
| "store.py", | |
| "world_model.py", | |
| "prediction_metrics.py", | |
| ) | |
| } | |
| provenance = { | |
| "kind": "reference_rebuild_not_runtime_migration", | |
| "schema_version": 1, | |
| "source_sha256": digest, | |
| "event_count": len(events), | |
| "episode_count": len(seen), | |
| "source_completeness": "declared_not_independently_verified", | |
| "replay_protocol": "mhrn.cognition.replay.v1", | |
| "run_id": run_id, | |
| "state_sha256": hashlib.sha256(state_path.read_bytes()).hexdigest(), | |
| "implementation_sha256": sources, | |
| "scientific_evidence": False, | |
| "neural_memory_claim": False, | |
| "canonical_runtime_checkpoint": False, | |
| } | |
| with (destination / "provenance.json").open("x", encoding="utf-8") as stream: | |
| stream.write(json.dumps(provenance, indent=2, sort_keys=True) + "\n") | |
| return state_path | |
| def main() -> None: | |
| """Command-line rebuild requiring explicit source and a new destination.""" | |
| parser = argparse.ArgumentParser(description=__doc__) | |
| parser.add_argument("source", type=Path) | |
| parser.add_argument("destination", type=Path) | |
| parser.add_argument("--sha256", required=True) | |
| args = parser.parse_args() | |
| try: | |
| print(rebuild_reference(args.source, args.destination, args.sha256)) | |
| except (OSError, ReplayError, MemoryWorldModelError) as error: | |
| parser.exit(2, f"Rebuild rejected: {error}\n") | |
| if __name__ == "__main__": | |
| main() | |