evidence/source/derive_no_fim.py

"""Attest run 1 shards and deterministically normalize them into E2 no-FIM data.

The E2 arm must not reopen upstream datasets. Run 1 did not preserve an ordered
raw-row manifest, and its source loader used different parser dispatch from the
corrected loader. The strongest available input is therefore the exact token
stream already used by run 1. Literal control strings created observationally
ambiguous EOS boundaries, so the receipt records a deterministic grammar
recovery and its limits rather than claiming unobservable original boundaries.

This tool has two CPU-only commands:

    python scripts/derive_no_fim.py attest \
      --index data/shards/index.json \
      --tokenizer tokenizer/code32k.json \
      --out config/run1_shard_integrity_receipt.json

    python scripts/derive_no_fim.py build \
      --config config/run2_no_fim.json
"""

import argparse
from datetime import datetime, timezone
import hashlib
import json
import os
import struct
import sys

import numpy as np
from tokenizers import Tokenizer

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from data import canonical_json_sha256, file_sha256, stable_file_sha256
from scripts.hf_metadata import write_json_atomic
from scripts.prepare_data import require_fresh_output_dir


ALGORITHM = "run1_deterministic_no_fim_normalization_v1"
INDEX_SCHEMA_VERSION = 3
RECEIPT_SCHEMA_VERSION = 1
DEFAULT_CHUNK_BYTES = 8 * 1024 * 1024
SHARD_TOKENS = 100_000_000
KIND_CODES = {"l2r": 0, "fim_psm": 1, "fim_spm": 2}
UNIT_RECORD_DTYPE = np.dtype(
    [
        ("length", "<u4"),
        ("kind", "u1"),
        ("reserved", "u1", (3,)),
    ]
)


def load_json(path):
    with open(path, encoding="utf-8") as f:
        value = json.load(f)
    if not isinstance(value, dict):
        raise ValueError(f"{path}: top-level JSON must be an object")
    return value


def special_token_ids(tokenizer_path):
    tokenizer = Tokenizer.from_file(tokenizer_path)
    names = {
        "eos": "<|endoftext|>",
        "prefix": "<|fim_prefix|>",
        "middle": "<|fim_middle|>",
        "suffix": "<|fim_suffix|>",
    }
    values = {key: tokenizer.token_to_id(value) for key, value in names.items()}
    if any(value is None for value in values.values()):
        raise ValueError("tokenizer is missing an EOS or FIM special token")
    if len(set(values.values())) != len(values):
        raise ValueError("EOS and FIM special-token ids must be distinct")
    return values, tokenizer.get_vocab_size()


def _as_uint16_le(ids):
    value = np.asarray(ids, dtype="<u2")
    if value.ndim != 1:
        raise ValueError("token unit must be one-dimensional")
    return value


def fim_scan_state(ids, special_ids):
    """Classify a segment as L2R, complete FIM, or visibly incomplete FIM."""
    ids = _as_uint16_le(ids)
    prefix_id = special_ids["prefix"]
    middle_id = special_ids["middle"]
    suffix_id = special_ids["suffix"]
    positions = {
        "prefix": np.flatnonzero(ids == prefix_id),
        "middle": np.flatnonzero(ids == middle_id),
        "suffix": np.flatnonzero(ids == suffix_id),
    }
    counts = {key: int(value.size) for key, value in positions.items()}
    if counts == {"prefix": 0, "middle": 0, "suffix": 0}:
        return "l2r"
    if any(value > 1 for value in counts.values()):
        raise ValueError(f"malformed FIM sentinel counts: {counts}")
    if counts["prefix"] != 1:
        raise ValueError("FIM sentinel fragment has no prefix sentinel")

    prefix = int(positions["prefix"][0])
    if prefix != 0:
        raise ValueError("malformed FIM unit does not begin with prefix sentinel")
    if counts["suffix"] == 0:
        if counts["middle"]:
            raise ValueError("FIM middle sentinel appears before suffix sentinel")
        return "incomplete"

    suffix = int(positions["suffix"][0])
    if suffix == 1:
        if counts["middle"] == 0:
            return "incomplete"
        middle = int(positions["middle"][0])
        if middle <= suffix:
            raise ValueError("SPM middle sentinel precedes its suffix payload")
        if middle < 4 or ids.size - middle - 1 < 2:
            return "incomplete"
        return "complete"

    if suffix < 2:
        raise ValueError("malformed PSM unit has an empty prefix")
    if counts["middle"] == 0:
        return "incomplete"
    middle = int(positions["middle"][0])
    if middle <= suffix:
        raise ValueError("PSM middle sentinel precedes its suffix payload")
    if middle < suffix + 3 or middle >= ids.size - 1:
        return "incomplete"
    return "complete"


def decode_fim_unit(ids, special_ids):
    """Return `(kind, raw_ids)` for one complete recovered run 1 unit."""
    ids = _as_uint16_le(ids)
    state = fim_scan_state(ids, special_ids)
    if state == "l2r":
        return "l2r", ids
    if state == "incomplete":
        raise ValueError("incomplete FIM unit")
    prefix_id = special_ids["prefix"]
    middle_id = special_ids["middle"]
    suffix_id = special_ids["suffix"]
    prefix = int(np.flatnonzero(ids == prefix_id)[0])
    middle = int(np.flatnonzero(ids == middle_id)[0])
    suffix = int(np.flatnonzero(ids == suffix_id)[0])
    if prefix != 0:
        raise AssertionError("complete FIM unit lost its prefix position")

    if suffix == 1:
        raw = _as_uint16_le(
            np.concatenate((ids[middle + 1:], ids[2:middle]))
        )
        if any(
            np.any(raw == special_ids[key])
            for key in ("prefix", "middle", "suffix")
        ):
            raise ValueError("decoded SPM raw unit retains a FIM sentinel")
        if not 16 <= raw.size <= 1024:
            raise ValueError("decoded SPM raw length is outside 16 to 1024")
        return "fim_spm", raw

    raw = _as_uint16_le(
        np.concatenate(
            (
                ids[1:suffix],
                ids[middle + 1:],
                ids[suffix + 1:middle],
            )
        )
    )
    if any(
        np.any(raw == special_ids[key])
        for key in ("prefix", "middle", "suffix")
    ):
        raise ValueError("decoded PSM raw unit retains a FIM sentinel")
    if not 16 <= raw.size <= 1024:
        raise ValueError("decoded PSM raw length is outside 16 to 1024")
    return "fim_psm", raw


class UnitEvidence:
    """Streaming hashes and counts for recovered ordered raw units."""

    def __init__(self, eos_token_id, domain="unspecified"):
        self.eos_token_id = eos_token_id
        self.domain = domain.encode("utf-8")
        self.ordered = hashlib.sha256()
        self.destination = hashlib.sha256()
        self.source_wire = hashlib.sha256()
        for digest, label in (
            (self.ordered, b"RECOVERED_RAW"),
            (self.destination, b"DESTINATION"),
            (self.source_wire, b"SOURCE_WIRE"),
        ):
            digest.update(b"WISP_NO_FIM_V1\0" + label + b"\0")
            digest.update(struct.pack("<Q", len(self.domain)))
            digest.update(self.domain)
        self.units = 0
        self.source_tokens = 0
        self.derived_tokens = 0
        self.raw_tokens = 0
        self.empty_units = 0
        self.internal_eos_tokens = 0
        self.counts = {"l2r": 0, "fim_psm": 0, "fim_spm": 0}
        self._eos_bytes = np.asarray([eos_token_id], dtype="<u2").tobytes()
        self._source_wire_complete = True

    def add(
        self,
        source_ids,
        kind,
        raw,
        internal_eos_tokens=0,
    ):
        raw = _as_uint16_le(raw)
        if kind not in KIND_CODES:
            raise ValueError(f"unknown recovered unit kind: {kind}")
        if kind == "l2r":
            if not 1 <= raw.size <= 1024:
                raise ValueError("plain recovered raw length is outside 1 to 1024")
        elif not 16 <= raw.size <= 1024:
            raise ValueError("FIM recovered raw length is outside 16 to 1024")
        raw_bytes = raw.tobytes()
        ordinal = self.units
        kind_code = KIND_CODES[kind]
        self.ordered.update(
            b"U"
            + struct.pack("<QBQ", ordinal, kind_code, int(raw.size))
        )
        self.ordered.update(raw_bytes)
        destination_bytes = raw_bytes + self._eos_bytes
        self.destination.update(
            b"U"
            + struct.pack(
                "<QBQ",
                ordinal,
                kind_code,
                int(raw.size) + 1,
            )
        )
        self.destination.update(destination_bytes)
        if source_ids is None:
            source_length = int(raw.size) + (3 if kind != "l2r" else 0)
            self._source_wire_complete = False
        else:
            source_ids = _as_uint16_le(source_ids)
            source_length = int(source_ids.size)
            self.source_wire.update(
                b"U"
                + struct.pack(
                    "<QBQ",
                    ordinal,
                    kind_code,
                    source_length + 1,
                )
            )
            self.source_wire.update(source_ids.tobytes())
            self.source_wire.update(self._eos_bytes)
        self.units += 1
        self.source_tokens += int(source_length) + 1
        self.raw_tokens += int(raw.size)
        self.derived_tokens += int(raw.size) + 1
        self.empty_units += int(raw.size == 0)
        self.internal_eos_tokens += int(internal_eos_tokens)
        self.counts[kind] += 1

    def summary(self):
        removed = self.source_tokens - self.derived_tokens
        fim_units = self.counts["fim_psm"] + self.counts["fim_spm"]
        if removed != 3 * fim_units:
            raise RuntimeError(
                "recovered stream did not remove exactly three tokens per "
                "FIM unit"
            )
        footer = b"END" + struct.pack(
            "<QQQQQQ",
            self.units,
            self.source_tokens,
            self.raw_tokens,
            self.derived_tokens,
            self.counts["l2r"],
            fim_units,
        )
        ordered = self.ordered.copy()
        ordered.update(footer)
        destination = self.destination.copy()
        destination.update(footer)
        source_wire = self.source_wire.copy()
        source_wire.update(footer)
        result = {
            "units": self.units,
            "source_tokens": self.source_tokens,
            "raw_tokens": self.raw_tokens,
            "derived_tokens": self.derived_tokens,
            "removed_fim_tokens": removed,
            "empty_eos_segments": self.empty_units,
            "reassembled_internal_eos_tokens": self.internal_eos_tokens,
            "l2r_units": self.counts["l2r"],
            "fim_psm_units": self.counts["fim_psm"],
            "fim_spm_units": self.counts["fim_spm"],
            "ordered_raw_units_sha256": ordered.hexdigest(),
            "destination_units_sha256": destination.hexdigest(),
        }
        if self._source_wire_complete:
            result["source_wire_units_sha256"] = source_wire.hexdigest()
        return result


def _same_summary(left, right):
    return canonical_json_sha256(left) == canonical_json_sha256(right)


def _normalization_evidence(value):
    keys = (
        "units",
        "source_tokens",
        "raw_tokens",
        "derived_tokens",
        "removed_fim_tokens",
        "l2r_units",
        "fim_psm_units",
        "fim_spm_units",
        "ordered_raw_units_sha256",
        "destination_units_sha256",
    )
    return {key: value.get(key) for key in keys}


def scan_shard(
    path,
    special_ids,
    expected_tokens=None,
    emit=None,
    split_evidence=None,
    evidence_domain="shard:unspecified",
    vocab_size=None,
    chunk_bytes=DEFAULT_CHUNK_BYTES,
):
    """Scan one shard once, hashing bytes and decoding every EOS segment."""
    if chunk_bytes < 2 or chunk_bytes % 2:
        raise ValueError("scan chunk size must be a positive even byte count")
    if os.path.islink(path):
        raise ValueError(f"{path}: shard cannot be a symbolic link")
    if not os.path.isfile(path):
        raise FileNotFoundError(path)
    before = os.stat(path, follow_symlinks=False)
    size = before.st_size
    if size % 2:
        raise ValueError(f"{path}: uint16 shard has an odd byte length")
    tokens = size // 2
    if expected_tokens is not None and tokens != expected_tokens:
        raise ValueError(
            f"{path}: {tokens} tokens != declared {expected_tokens}"
        )

    local = UnitEvidence(special_ids["eos"], evidence_domain)
    split_evidence = split_evidence or UnitEvidence(
        special_ids["eos"],
        "split:unspecified",
    )
    shard_digest = hashlib.sha256()
    carry = np.empty(0, dtype="<u2")
    pending = None
    pending_internal_eos = 0
    pending_start_segment = None
    pending_start_token = None
    eos_segment_ordinal = 0
    reassembly_groups = []
    token_offset = 0
    unit_ordinal = 0
    with open(path, "rb") as f:
        while True:
            block = f.read(chunk_bytes)
            if not block:
                break
            shard_digest.update(block)
            if len(block) % 2:
                raise ValueError(f"{path}: partial uint16 token in scan chunk")
            current = np.frombuffer(block, dtype="<u2")
            if (
                vocab_size is not None
                and current.size
                and int(current.max()) >= vocab_size
            ):
                raise ValueError(f"{path}: token id is outside tokenizer vocab")
            if carry.size:
                current = np.concatenate((carry, current))
            starts_at = token_offset - int(carry.size)
            boundaries = np.flatnonzero(current == special_ids["eos"])
            start = 0
            for boundary_value in boundaries:
                boundary = int(boundary_value)
                segment = current[start:boundary]
                if pending is not None:
                    segment = np.concatenate((pending, segment))
                try:
                    state = fim_scan_state(segment, special_ids)
                except ValueError as exc:
                    absolute = starts_at + start
                    raise ValueError(
                        f"{path}: malformed unit {unit_ordinal} at token "
                        f"{absolute}: {exc}"
                    ) from exc
                if state == "incomplete":
                    if segment.size >= 1027:
                        raise ValueError(
                            f"{path}: incomplete FIM unit {unit_ordinal} "
                            "exceeds the maximum framed length"
                        )
                    pending = np.concatenate(
                        (
                            segment,
                            np.asarray(
                                [special_ids["eos"]],
                                dtype="<u2",
                            ),
                        )
                    )
                    if pending_start_segment is None:
                        pending_start_segment = eos_segment_ordinal
                        pending_start_token = starts_at + start
                    pending_internal_eos += 1
                    start = boundary + 1
                    eos_segment_ordinal += 1
                    continue
                kind, raw = decode_fim_unit(segment, special_ids)
                local.add(
                    segment,
                    kind,
                    raw,
                    internal_eos_tokens=pending_internal_eos,
                )
                split_evidence.add(
                    segment,
                    kind,
                    raw,
                    internal_eos_tokens=pending_internal_eos,
                )
                if emit is not None:
                    emit(raw, special_ids["eos"], kind)
                if pending_internal_eos:
                    reassembly_groups.append(
                        {
                            "unit_ordinal": unit_ordinal,
                            "first_eos_segment": pending_start_segment,
                            "last_eos_segment": eos_segment_ordinal,
                            "internal_eos_tokens": pending_internal_eos,
                            "source_token_start": pending_start_token,
                            "writer_eos_token": starts_at + boundary,
                        }
                    )
                unit_ordinal += 1
                pending = None
                pending_internal_eos = 0
                pending_start_segment = None
                pending_start_token = None
                start = boundary + 1
                eos_segment_ordinal += 1
            carry = current[start:].copy()
            if carry.size > 1027:
                raise ValueError(
                    f"{path}: unterminated unit exceeds maximum wire length"
                )
            token_offset += len(block) // 2
    if pending is not None:
        raise ValueError(f"{path}: incomplete FIM unit at shard boundary")
    if carry.size:
        raise ValueError(
            f"{path}: shard does not end at an EOS-delimited unit boundary"
        )
    after = os.stat(path, follow_symlinks=False)
    if (
        before.st_dev,
        before.st_ino,
        before.st_size,
        before.st_mtime_ns,
    ) != (
        after.st_dev,
        after.st_ino,
        after.st_size,
        after.st_mtime_ns,
    ):
        raise RuntimeError(f"{path}: shard changed during integrity scan")
    summary = local.summary()
    return {
        "bytes": size,
        "tokens": tokens,
        "sha256": shard_digest.hexdigest(),
        "reassembly_groups": reassembly_groups,
        **summary,
    }


def scan_index(index_path, tokenizer_path):
    index = load_json(index_path)
    special_ids, vocab_size = special_token_ids(tokenizer_path)
    tokenizer_sha256 = file_sha256(tokenizer_path)
    if index.get("vocab_size") != vocab_size:
        raise ValueError("source index vocab size differs from tokenizer")
    root = os.path.dirname(os.path.abspath(index_path))
    split_receipts = {}
    for split in ("train", "val"):
        entries = index.get("splits", {}).get(split)
        if not isinstance(entries, list) or not entries:
            raise ValueError(f"source index split {split} is empty")
        paths = [entry.get("path") for entry in entries]
        if len(set(paths)) != len(paths):
            raise ValueError(f"source index split {split} repeats a shard path")
        split_evidence = UnitEvidence(
            special_ids["eos"],
            f"split:{split}:tokenizer:{tokenizer_sha256}",
        )
        shard_receipts = []
        for entry in entries:
            relative = entry.get("path")
            if (
                not isinstance(relative, str)
                or not relative
                or os.path.isabs(relative)
                or os.path.normpath(relative) != relative
                or relative == ".."
                or relative.startswith(".." + os.sep)
            ):
                raise ValueError(
                    f"source index split {split} has unsafe shard path"
                )
            path = os.path.join(root, relative)
            shard = scan_shard(
                path,
                special_ids,
                expected_tokens=entry.get("tokens"),
                split_evidence=split_evidence,
                evidence_domain=(
                    f"shard:{split}:{relative}:tokenizer:{tokenizer_sha256}"
                ),
                vocab_size=vocab_size,
            )
            shard_receipts.append({"path": relative, **shard})
        summary = split_evidence.summary()
        declared_total = entries[0].get("total_tokens")
        if summary["source_tokens"] != declared_total:
            raise ValueError(
                f"source split {split} scan total differs from index"
            )
        split_receipts[split] = {
            **summary,
            "shards": shard_receipts,
            "shard_manifest_sha256": canonical_json_sha256(
                [
                    {
                        key: shard[key]
                        for key in ("path", "tokens", "bytes", "sha256")
                    }
                    for shard in shard_receipts
                ]
            ),
        }
    return index, special_ids, split_receipts


def build_attestation(index_path, tokenizer_path):
    index, special_ids, splits = scan_index(index_path, tokenizer_path)
    reassembly_groups = sum(
        len(shard["reassembly_groups"])
        for shard in splits["train"]["shards"]
        + splits["val"]["shards"]
    )
    restored_internal_eos = sum(
        split["reassembled_internal_eos_tokens"]
        for split in splits.values()
    )
    return {
        "schema_version": RECEIPT_SCHEMA_VERSION,
        "created_at": datetime.now(timezone.utc).isoformat(),
        "status": "complete",
        "algorithm": ALGORITHM,
        "dtype": "uint16_le",
        "source_index": {
            "path": index_path,
            "sha256": file_sha256(index_path),
            "schema_version": index.get("schema_version", 1),
        },
        "tokenizer": {
            "path": tokenizer_path,
            "sha256": file_sha256(tokenizer_path),
            "vocab_size": index["vocab_size"],
        },
        "special_token_ids": special_ids,
        "attestation_script": {
            "path": os.path.relpath(
                os.path.abspath(__file__),
                os.getcwd(),
            ),
            "sha256": file_sha256(__file__),
        },
        "boundary_recovery": {
            "exact_original_units_proven": False,
            "mode": "deterministic_visible_grammar_normalization",
            "rule": (
                "Start only at a partial prefix-bearing EOS segment, append "
                "following same-shard segments, restore each intervening EOS "
                "as payload, and stop at the first complete valid FIM grammar."
            ),
            "detectable_reassembly_groups": reassembly_groups,
            "restored_internal_eos_tokens": restored_internal_eos,
            "cross_shard_reassembly_allowed": False,
        },
        "splits": splits,
        "limitations": [
            (
                "Literal EOS tokens in L2R content and after the final FIM "
                "marker can be observationally indistinguishable from writer "
                "boundaries."
            ),
            (
                "A raw untransformed segment that exactly mimics one valid FIM "
                "frame is structurally indistinguishable from a generated frame."
            ),
            (
                "This post-build receipt binds current token bytes and visible "
                "grammar, not raw source rows, build-time shard hashes, or exact "
                "original unit boundaries."
            ),
        ],
    }


class DerivedShardWriter:
    """Write one destination shard plus a fixed-width unit-boundary sidecar."""

    def __init__(self, out_dir, split, name):
        self.out_dir = out_dir
        self.split = split
        self.name = name
        self.buf = []
        self.records = []
        self.total = 0

    def add_raw(self, raw, eos_token_id, kind):
        raw = _as_uint16_le(raw)
        unit = np.empty(raw.size + 1, dtype="<u2")
        unit[:-1] = raw
        unit[-1] = eos_token_id
        self.buf.append(unit)
        self.records.append((int(unit.size), KIND_CODES[kind]))
        self.total += int(unit.size)

    def flush(self):
        if not self.buf:
            raise ValueError(f"derived shard {self.name} has no units")
        values = np.concatenate(self.buf).astype("<u2", copy=False)
        path = os.path.join(self.out_dir, self.name)
        if os.path.lexists(path):
            raise FileExistsError(f"refusing to replace derived shard: {path}")
        digest = hashlib.sha256()
        digest.update(memoryview(values).cast("B"))
        with open(path, "xb") as f:
            values.tofile(f)
            f.flush()
            os.fsync(f.fileno())
        sidecar_name = self.name + ".units.bin"
        sidecar_path = os.path.join(self.out_dir, sidecar_name)
        if os.path.lexists(sidecar_path):
            raise FileExistsError(
                f"refusing to replace unit sidecar: {sidecar_path}"
            )
        records = np.zeros(len(self.records), dtype=UNIT_RECORD_DTYPE)
        records["length"] = [value[0] for value in self.records]
        records["kind"] = [value[1] for value in self.records]
        sidecar_digest = hashlib.sha256()
        sidecar_digest.update(memoryview(records).cast("B"))
        with open(sidecar_path, "xb") as f:
            records.tofile(f)
            f.flush()
            os.fsync(f.fileno())
        entry = {
            "path": self.name,
            "tokens": int(values.size),
            "bytes": int(values.nbytes),
            "sha256": digest.hexdigest(),
            "unit_sidecar": {
                "path": sidecar_name,
                "records": int(records.size),
                "bytes": int(records.nbytes),
                "sha256": sidecar_digest.hexdigest(),
                "record_format": "uint32_length_uint8_kind_3_zero_bytes",
            },
        }
        self.buf = []
        self.records = []
        return entry


def verify_derived_shard(
    out_dir,
    split,
    entry,
    special_ids,
    tokenizer_sha256,
    split_evidence=None,
):
    """Independently reread one destination shard and its unit sidecar."""
    path = os.path.join(out_dir, entry["path"])
    sidecar = entry["unit_sidecar"]
    sidecar_path = os.path.join(out_dir, sidecar["path"])
    if stable_file_sha256(path) != entry["sha256"]:
        raise ValueError(f"derived shard hash differs: {entry['path']}")
    if stable_file_sha256(sidecar_path) != sidecar["sha256"]:
        raise ValueError(
            f"derived unit-sidecar hash differs: {sidecar['path']}"
        )
    values = np.memmap(path, dtype="<u2", mode="r")
    records = np.memmap(
        sidecar_path,
        dtype=UNIT_RECORD_DTYPE,
        mode="r",
    )
    if values.size != entry["tokens"] or values.nbytes != entry["bytes"]:
        raise ValueError(f"derived shard size differs: {entry['path']}")
    if (
        records.size != sidecar["records"]
        or records.nbytes != sidecar["bytes"]
        or np.any(records["reserved"] != 0)
    ):
        raise ValueError(
            f"derived unit-sidecar structure differs: {sidecar['path']}"
        )
    if int(records["length"].sum(dtype=np.uint64)) != int(values.size):
        raise ValueError(
            f"derived unit-sidecar lengths differ: {sidecar['path']}"
        )
    if np.any(records["kind"] > max(KIND_CODES.values())):
        raise ValueError(
            f"derived unit-sidecar kind differs: {sidecar['path']}"
        )
    for key in ("prefix", "middle", "suffix"):
        if np.any(values == special_ids[key]):
            raise ValueError(
                f"derived shard retains FIM sentinel: {entry['path']}"
            )

    domain = (
        f"shard:{split}:{entry['path']}:tokenizer:{tokenizer_sha256}"
    )
    evidence = UnitEvidence(special_ids["eos"], domain)
    split_evidence = split_evidence or UnitEvidence(
        special_ids["eos"],
        f"split:{split}:tokenizer:{tokenizer_sha256}",
    )
    kinds = {value: key for key, value in KIND_CODES.items()}
    offset = 0
    for record in records:
        length = int(record["length"])
        end = offset + length
        if length < 2 or end > values.size:
            raise ValueError(
                f"derived unit-sidecar boundary differs: {sidecar['path']}"
            )
        unit = np.asarray(values[offset:end])
        if int(unit[-1]) != special_ids["eos"]:
            raise ValueError(
                f"derived unit lacks writer EOS: {sidecar['path']}"
            )
        kind = kinds[int(record["kind"])]
        raw = unit[:-1]
        evidence.add(None, kind, raw)
        split_evidence.add(None, kind, raw)
        offset = end
    if offset != values.size:
        raise ValueError(
            f"derived unit-sidecar does not cover shard: {sidecar['path']}"
        )
    return evidence.summary()


def validate_derivation_contract(config, receipt):
    contract = config.get("data_build_contract")
    expected = {
        "schema_version": 2,
        "strategy": ALGORITHM,
        "source_index": receipt["source_index"],
        "source_integrity_receipt": config.get("data_integrity", {}).get(
            "receipt"
        ),
        "source_integrity_receipt_sha256": config.get(
            "data_integrity", {}
        ).get("sha256"),
        "require_fresh_output_dir": True,
    }
    if contract != expected:
        raise ValueError(
            "no-FIM derivation request differs from config contract:\n"
            f"expected {contract!r}\n"
            f"actual {expected!r}"
        )
    if config.get("fim_rate") != 0.0:
        raise ValueError("no-FIM derivation config fim_rate must be 0.0")
    if "sources" in config:
        raise ValueError(
            "no-FIM derivation config must not contain executable sources"
        )
    return contract


def build_no_fim(config_path):
    config = load_json(config_path)
    integrity = config.get("data_integrity", {})
    if integrity.get("role") != "source":
        raise ValueError("run 2 data_integrity role must be source")
    receipt_path = integrity.get("receipt")
    receipt = load_json(receipt_path)
    if file_sha256(receipt_path) != integrity.get("sha256"):
        raise ValueError("run 1 source integrity receipt hash differs")
    if receipt.get("status") != "complete" or receipt.get(
        "algorithm"
    ) != ALGORITHM:
        raise ValueError("run 1 source integrity receipt is incomplete")
    contract = validate_derivation_contract(config, receipt)

    tokenizer_path = config["tokenizer_path"]
    special_ids, vocab_size = special_token_ids(tokenizer_path)
    if receipt.get("special_token_ids") != special_ids:
        raise ValueError("source receipt special-token ids differ")
    if receipt.get("tokenizer", {}).get("sha256") != file_sha256(
        tokenizer_path
    ):
        raise ValueError("source receipt tokenizer differs")

    out_dir = require_fresh_output_dir(config["data_dir"])
    os.makedirs(out_dir, exist_ok=True)
    source_index_path = receipt["source_index"]["path"]
    if file_sha256(source_index_path) != receipt["source_index"]["sha256"]:
        raise ValueError("run 1 source index hash differs")
    source_index = load_json(source_index_path)
    source_root = os.path.dirname(os.path.abspath(source_index_path))
    tokenizer_sha256 = file_sha256(tokenizer_path)
    output_splits = {}
    split_evidence_out = {}
    for split in ("train", "val"):
        split_evidence = UnitEvidence(
            special_ids["eos"],
            f"split:{split}:tokenizer:{tokenizer_sha256}",
        )
        verified_split_evidence = UnitEvidence(
            special_ids["eos"],
            f"split:{split}:tokenizer:{tokenizer_sha256}",
        )
        expected_split = receipt["splits"][split]
        source_entries = source_index["splits"][split]
        if [entry["path"] for entry in source_entries] != [
            entry["path"] for entry in expected_split["shards"]
        ]:
            raise ValueError(f"run 1 split {split} shard order differs")
        output_entries = []
        for source_entry, expected_shard in zip(
            source_entries,
            expected_split["shards"],
        ):
            path = os.path.join(source_root, source_entry["path"])
            writer = DerivedShardWriter(
                out_dir,
                split,
                source_entry["path"],
            )
            actual_shard = {
                "path": source_entry["path"],
                **scan_shard(
                    path,
                    special_ids,
                    expected_tokens=source_entry["tokens"],
                    emit=writer.add_raw,
                    split_evidence=split_evidence,
                    evidence_domain=(
                        f"shard:{split}:{source_entry['path']}:"
                        f"tokenizer:{tokenizer_sha256}"
                    ),
                    vocab_size=vocab_size,
                ),
            }
            if not _same_summary(actual_shard, expected_shard):
                raise ValueError(
                    f"run 1 source shard changed: {source_entry['path']}"
                )
            output_entry = writer.flush()
            verified = verify_derived_shard(
                out_dir,
                split,
                output_entry,
                special_ids,
                tokenizer_sha256,
                split_evidence=verified_split_evidence,
            )
            if _normalization_evidence(verified) != (
                _normalization_evidence(actual_shard)
            ):
                raise ValueError(
                    f"derived shard evidence differs: "
                    f"{source_entry['path']}"
                )
            output_entry["source"] = {
                "path": source_entry["path"],
                "tokens": actual_shard["tokens"],
                "sha256": actual_shard["sha256"],
                "source_wire_units_sha256": actual_shard[
                    "source_wire_units_sha256"
                ],
                "ordered_raw_units_sha256": actual_shard[
                    "ordered_raw_units_sha256"
                ],
            }
            output_entry["normalization"] = _normalization_evidence(
                verified
            )
            output_entries.append(output_entry)
        actual_split = split_evidence.summary()
        verified_split = verified_split_evidence.summary()
        expected_summary = {
            key: value
            for key, value in expected_split.items()
            if key not in ("shards", "shard_manifest_sha256")
        }
        if not _same_summary(actual_split, expected_summary):
            raise ValueError(f"run 1 split {split} evidence changed")
        if _normalization_evidence(verified_split) != (
            _normalization_evidence(actual_split)
        ):
            raise ValueError(
                f"independent derived split {split} evidence differs"
            )
        output_total = sum(entry["tokens"] for entry in output_entries)
        if output_total != actual_split["derived_tokens"]:
            raise ValueError(f"derived split {split} token total differs")
        for entry in output_entries:
            entry["total_tokens"] = output_total
        output_splits[split] = output_entries
        split_evidence_out[split] = actual_split

    index = {
        "schema_version": INDEX_SCHEMA_VERSION,
        "vocab_size": vocab_size,
        "fim_rate": 0.0,
        "fim_chunk": config["fim_chunk"],
        "splits": output_splits,
        "build": {
            "completed": True,
            "strategy": ALGORITHM,
            "fresh_output_directory": True,
            "config_path": config_path,
            "config_canonical_sha256": canonical_json_sha256(config),
            "tokenizer_path": tokenizer_path,
            "tokenizer_sha256": file_sha256(tokenizer_path),
            "source_index": receipt["source_index"],
            "source_integrity_receipt": {
                "path": receipt_path,
                "sha256": integrity["sha256"],
            },
            "special_token_ids": special_ids,
            "derivation_script": {
                "path": os.path.relpath(
                    os.path.abspath(__file__),
                    os.getcwd(),
                ),
                "sha256": file_sha256(__file__),
            },
            "split_evidence": split_evidence_out,
            "contract": contract,
        },
    }
    if (
        file_sha256(receipt_path) != integrity["sha256"]
        or file_sha256(source_index_path)
        != receipt["source_index"]["sha256"]
        or file_sha256(tokenizer_path) != tokenizer_sha256
    ):
        raise RuntimeError("no-FIM derivation input changed before publication")
    index_path = os.path.join(out_dir, "index.json")
    write_json_atomic(index_path, index)
    return index_path, index


def main():
    parser = argparse.ArgumentParser()
    subparsers = parser.add_subparsers(dest="command", required=True)
    attest = subparsers.add_parser("attest")
    attest.add_argument("--index", required=True)
    attest.add_argument("--tokenizer", required=True)
    attest.add_argument("--out", required=True)
    build = subparsers.add_parser("build")
    build.add_argument("--config", required=True)
    cli = parser.parse_args()

    if cli.command == "attest":
        if os.path.lexists(cli.out):
            raise FileExistsError(
                f"refusing to replace integrity receipt: {cli.out}"
            )
        receipt = build_attestation(cli.index, cli.tokenizer)
        write_json_atomic(cli.out, receipt)
        print(
            f"wrote {cli.out}: "
            f"{receipt['splits']['train']['source_tokens']:,} train source "
            "tokens"
        )
        return

    index_path, index = build_no_fim(cli.config)
    print(
        f"wrote {index_path}: "
        f"{index['splits']['train'][0]['total_tokens']:,} train / "
        f"{index['splits']['val'][0]['total_tokens']:,} val tokens"
    )


if __name__ == "__main__":
    main()