Blame for src/memory/replay.py

"""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()