shared/tosp_protocol.py

from __future__ import annotations

import hashlib
from typing import Any

from constitutional_constants import HEADER_SIZE, LAMBDA, OMEGA, PHI, QBEC_VERSION, SIGMA
from shared.serialization import safe_json_dumps


def build_header(phase: str, rdod_value: float, frac: float, node_id: str, state: dict[str, Any] | None = None) -> bytes:
    digest_source = state if state is not None else {
        "phase": phase,
        "rdod": round(rdod_value, 8),
        "frac": round(frac, 8),
        "node_id": node_id,
    }
    digest = hashlib.sha256(safe_json_dumps(digest_source, sort_keys=True).encode("utf-8")).hexdigest()[:32]
    raw = (
        f"{QBEC_VERSION:<10}"
        f"{SIGMA:4.1f}"
        f"{LAMBDA:<16}"
        f"{phase:<16}"
        f"{rdod_value:8.4f}"
        f"{frac:8.4f}"
        f"{node_id:<16}"
        f"{OMEGA:10.2f}"
        f"{PHI:8.5f}"
        f"{digest:<32}"
    )
    return raw[:128].ljust(HEADER_SIZE).encode("utf-8")


def verify_header(header: bytes) -> tuple[bool, dict[str, Any] | str]:
    if len(header) != HEADER_SIZE:
        return False, "L3_REJECT: Header size != 144"
    text = header.decode("utf-8")
    if text[14:30].strip() != LAMBDA:
        return False, "L1_REJECT: Lattice lock mismatch"
    try:
        sigma_value = float(text[10:14].strip())
    except ValueError:
        return False, "L2_REJECT: Invalid sigma"
    if sigma_value != SIGMA:
        return False, "L2_REJECT: Sovereignty violation"
    return True, {
        "version": text[0:10].strip(),
        "sigma": sigma_value,
        "lambda": text[14:30].strip(),
        "phase": text[30:46].strip(),
        "rdod": float(text[46:54].strip()),
        "frac": float(text[54:62].strip()),
        "node_id": text[62:78].strip(),
        "omega": float(text[78:88].strip()),
        "phi": float(text[88:96].strip()),
        "merkle": text[96:128].strip()
    }