File size: 5,013 Bytes
767a051 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 | """Generic byte materialization from ordered TOMAGI EMIT execution records."""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Mapping, Sequence
from .core import (
FLAG_EMIT_BIG_ENDIAN,
FLAG_EMIT_COUNT_MASK,
FLAG_EMIT_COUNT_SHIFT,
Opcode,
Program,
State,
PROGRAM_FLAG_EMIT_BYTES,
PROGRAM_FLAG_SEEDED_PROFILE,
run,
u32,
)
from .format import load
TRACE_FIELDS = (
"step", "cell_before", "opcode", "branch", "cell_after", "key_hi", "key_lo",
"rho", "theta", "tick", "phi", "orientation", "sheet", "residual", "output",
"lineage", "status",
)
@dataclass(frozen=True, slots=True)
class EmitRecord:
sequence: int
step: int
cell_index: int
flags: int
payload: int
byte_count: int
byte_order: str
data: bytes
lineage: int
def as_record(self) -> dict[str, Any]:
return {
"sequence": self.sequence,
"step": self.step,
"cell_index": self.cell_index,
"flags": self.flags,
"payload": self.payload,
"byte_count": self.byte_count,
"byte_order": self.byte_order,
"hex": self.data.hex(),
"lineage": self.lineage,
}
def decode_emit_payload(flags: int, payload: int) -> tuple[bytes, int, str]:
flags = u32(flags)
payload = u32(payload)
count = (flags & FLAG_EMIT_COUNT_MASK) >> FLAG_EMIT_COUNT_SHIFT
if not 1 <= count <= 4:
raise ValueError(f"EMIT byte count must be in 1..4, found {count}")
if flags & FLAG_EMIT_BIG_ENDIAN:
return payload.to_bytes(4, "big")[-count:], count, "big"
return payload.to_bytes(4, "little")[:count], count, "little"
def _validate_trace(
program: Program,
trace: Sequence[Mapping[str, Any]],
) -> list[Mapping[str, Any]]:
"""Authenticate a canonical-initial-state trace prefix by deterministic replay."""
rows = list(trace)
_, expected = run(program, ticks=len(rows), trace=True)
if len(expected) != len(rows):
raise ValueError("trace continues after the program halted")
for index, (row, reference) in enumerate(zip(rows, expected)):
if not isinstance(row, Mapping):
raise ValueError(f"trace row {index} must be an object")
unknown = sorted(
(key for key in row if key not in TRACE_FIELDS), key=repr
)
if unknown:
raise ValueError(
f"trace row {index} has unknown fields: "
+ ", ".join(repr(key) for key in unknown)
)
for field in TRACE_FIELDS:
if field not in row:
raise ValueError(f"trace row {index} is missing field {field}")
value = row[field]
if isinstance(value, bool) or not isinstance(value, int):
raise ValueError(f"trace row {index} field {field} must be an integer")
if value != reference[field]:
raise ValueError(
f"trace row {index} field {field} does not match deterministic replay"
)
return rows
def materialize_trace(
program: Program,
trace: Sequence[Mapping[str, Any]],
) -> tuple[bytes, list[EmitRecord]]:
if not (program.flags & PROGRAM_FLAG_SEEDED_PROFILE):
raise ValueError("byte materialization requires a seeded-profile program")
if not (program.flags & PROGRAM_FLAG_EMIT_BYTES):
raise ValueError("program does not declare emitted-byte materialization")
validated_trace = _validate_trace(program, trace)
output = bytearray()
records: list[EmitRecord] = []
for row in validated_trace:
index = int(row["cell_before"])
if not 0 <= index < len(program.cells):
raise ValueError(f"trace cell index outside program: {index}")
cell = program.cells[index]
if cell.opcode != int(Opcode.EMIT):
continue
data, count, byte_order = decode_emit_payload(cell.flags, cell.payload)
records.append(EmitRecord(
sequence=len(records),
step=int(row["step"]),
cell_index=index,
flags=cell.flags,
payload=cell.payload,
byte_count=count,
byte_order=byte_order,
data=data,
lineage=int(row.get("lineage", 0)),
))
output.extend(data)
if not records:
raise ValueError("execution trace contains no EMIT records")
return bytes(output), records
def materialize_file(
program_path: str | Path,
destination: str | Path,
*,
ticks: int | None = None,
) -> tuple[bytes, State, list[dict[str, int]], list[EmitRecord]]:
program = load(program_path)
state, trace = run(program, ticks=ticks, trace=True)
data, records = materialize_trace(program, trace)
out = Path(destination)
out.parent.mkdir(parents=True, exist_ok=True)
out.write_bytes(data)
return data, state, trace, records
|