arborist/bench/scripts/t3_bound_calculator.py

"""T3 per-window covert-channel budget calculator (#000036 §11).

Implements the closed-form bound from
`docs/soft-hash-channel-t3-bound.md` §6:

    I_window ≤ C_B1 · g · W · log₂(SNR_grad + 1)
             + C_B2 · ⌈W / K⌉ · log₂(R)
             + C_B3 · ⌈W / E⌉ · log₂(N_b · σ_grad / ‖∇L_max‖) / 2

Inputs are operator-measurable on a deployment:

    g            T2 gradient-fraction the adversary controls
    ‖∇L_max‖     gradient-norm cap (clipping ceiling)
    σ_grad       per-step gradient stddev (SGD noise floor)
    K            LR-decision interval (steps between LR changes)
    R            LR-grid size
    W            nonce-window length (steps before re-anchor)
    N_b          batches per epoch
    E            steps per epoch (typically equals N_b)

Conservative defaults: C_B1 = C_B2 = C_B3 = 1 (each by data-
processing inequality; see §6 + §10). Operators with deployment-
specific empirical measurements of the constants pass them via
``--c-b1`` / ``--c-b2`` / ``--c-b3`` flags.

Pure stdlib. No numpy / scipy dependency — the math is
arithmetic + math.log2.

Usage:

    python -m bench.scripts.t3_bound_calculator \\
        --gradient-fraction 0.05 \\
        --gradient-norm-max 1.0 \\
        --gradient-noise-stddev 0.1 \\
        --lr-decision-interval 100 \\
        --lr-grid-size 8 \\
        --window-length 10000 \\
        --batches-per-epoch 1024 \\
        --steps-per-epoch 1024
"""

from __future__ import annotations

import argparse
import json
import math
import sys


CALCULATOR_VERSION = "t3-bound-v1-bottou-refinement"


def t3_bound_bits(
    *,
    gradient_fraction: float,        # g
    gradient_norm_max: float,        # ‖∇L_max‖
    gradient_noise_stddev: float,    # σ_grad
    lr_decision_interval: int,       # K
    lr_grid_size: int,               # R
    window_length: int,              # W
    batches_per_epoch: int,          # N_b
    steps_per_epoch: int,            # E
    c_b1: float = 1.0,
    c_b2: float = 1.0,
    c_b3: float = 1.0,
) -> dict:
    """Compute the per-window mutual-information bound + per-bandwidth
    contributions per #000036 §6.

    Returns a dict with the total bound, three per-bandwidth
    contributions, the constants in use, and a sentence of operator
    guidance based on the bound vs the SHA-256 (256-bit) baseline.
    """
    # Validation — the bound is meaningful only on positive inputs.
    if not 0 < gradient_fraction <= 1:
        raise ValueError(
            f"gradient_fraction must be in (0, 1]; got {gradient_fraction}"
        )
    if gradient_norm_max <= 0:
        raise ValueError(
            f"gradient_norm_max must be > 0; got {gradient_norm_max}"
        )
    if gradient_noise_stddev <= 0:
        raise ValueError(
            f"gradient_noise_stddev must be > 0; got {gradient_noise_stddev}"
        )
    for name, v in [
        ("lr_decision_interval", lr_decision_interval),
        ("lr_grid_size", lr_grid_size),
        ("window_length", window_length),
        ("batches_per_epoch", batches_per_epoch),
        ("steps_per_epoch", steps_per_epoch),
    ]:
        if not isinstance(v, int) or v <= 0:
            raise ValueError(f"{name} must be a positive int; got {v!r}")
    for name, c in [("c_b1", c_b1), ("c_b2", c_b2), ("c_b3", c_b3)]:
        if not 0 <= c <= 1:
            raise ValueError(f"{name} must be in [0, 1]; got {c}")

    # B1 — gradient bias. Per-step Fano bound on adversary-controlled
    # parameter shift; SNR is g · ‖∇L_max‖ / σ_grad (#000036 §3).
    snr_grad = gradient_fraction * gradient_norm_max / gradient_noise_stddev
    b1_per_step = math.log2(snr_grad + 1)
    b1 = c_b1 * gradient_fraction * window_length * b1_per_step

    # B2 — LR selection. R-symbol categorical channel at every K-step
    # decision point (#000036 §4).
    decisions_in_window = -(-window_length // lr_decision_interval)  # ceil
    b2 = c_b2 * decisions_in_window * math.log2(lr_grid_size)

    # B3 — Batch order. Bottou-Bousquet refinement: per-epoch contribution
    # bounded by 0.5 · log₂(N_b · σ_grad / ‖∇L_max‖) (#000036 §5).
    epochs_in_window = -(-window_length // steps_per_epoch)
    b3_per_epoch_factor = batches_per_epoch * gradient_noise_stddev / gradient_norm_max
    if b3_per_epoch_factor <= 1:
        # log₂ of a value ≤ 1 is ≤ 0; the bound becomes trivial (the
        # adversarial order can't move the parameter trajectory beyond
        # the noise floor). Floor at 0 — adversary can't do worse than
        # random shuffle in this regime.
        b3_per_epoch = 0.0
    else:
        b3_per_epoch = math.log2(b3_per_epoch_factor) / 2.0
    b3 = c_b3 * epochs_in_window * b3_per_epoch

    total = b1 + b2 + b3

    # Operator guidance — translate the bound to "windows needed to
    # brute-force a 256-bit target".
    sha256_bits = 256
    if total <= 0:
        recommendation = (
            "I_window ≤ 0 (degenerate inputs); bound is vacuously safe."
        )
    elif total >= sha256_bits:
        # Adversary can in principle steer the full SHA-256 output
        # within one window — i.e. M2's nonce-window discipline is
        # too loose at this W. Operators must reduce W (or g, K, R).
        recommendation = (
            f"I_window ≈ {total:.1f} bits/window EXCEEDS the "
            f"SHA-256 ({sha256_bits} bit) output size. M2's "
            f"single-window guarantee is broken at this W. "
            f"Reduce W (or reduce g / R / increase K) until "
            f"I_window < 256 bits/window."
        )
    else:
        windows_to_brute_force = 2 ** (sha256_bits - total)
        recommendation = (
            f"I_window ≈ {total:.1f} bits/window. To steer C(M) "
            f"to a specific 256-bit target, adversary needs "
            f"≥ 2^{sha256_bits - total:.1f} ≈ {windows_to_brute_force:.2e} "
            f"windows."
        )

    return {
        "calculator_version": CALCULATOR_VERSION,
        "I_window_bits_upper_bound": round(total, 4),
        "B1_contribution": round(b1, 4),
        "B2_contribution": round(b2, 4),
        "B3_contribution": round(b3, 4),
        "snr_grad": round(snr_grad, 4),
        "decisions_in_window": decisions_in_window,
        "epochs_in_window": epochs_in_window,
        "constants": {"C_B1": c_b1, "C_B2": c_b2, "C_B3": c_b3},
        "inputs": {
            "gradient_fraction": gradient_fraction,
            "gradient_norm_max": gradient_norm_max,
            "gradient_noise_stddev": gradient_noise_stddev,
            "lr_decision_interval": lr_decision_interval,
            "lr_grid_size": lr_grid_size,
            "window_length": window_length,
            "batches_per_epoch": batches_per_epoch,
            "steps_per_epoch": steps_per_epoch,
        },
        "recommendation": recommendation,
    }


def main(argv: list[str] | None = None) -> int:
    p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    p.add_argument(
        "--gradient-fraction", type=float, required=True,
        help="g — fraction of gradients the T2 adversary controls; in (0, 1]",
    )
    p.add_argument(
        "--gradient-norm-max", type=float, required=True,
        help="‖∇L_max‖ — gradient-norm clipping ceiling (deployment param)",
    )
    p.add_argument(
        "--gradient-noise-stddev", type=float, required=True,
        help="σ_grad — per-step gradient noise stddev (SGD noise floor)",
    )
    p.add_argument(
        "--lr-decision-interval", type=int, required=True,
        help="K — steps between LR-schedule decision points",
    )
    p.add_argument(
        "--lr-grid-size", type=int, required=True,
        help="R — number of LR levels in the schedule grid",
    )
    p.add_argument(
        "--window-length", type=int, required=True,
        help="W — steps per nonce-window (M2 anchor re-randomization period)",
    )
    p.add_argument(
        "--batches-per-epoch", type=int, required=True,
        help="N_b — batch count per training epoch",
    )
    p.add_argument(
        "--steps-per-epoch", type=int, required=True,
        help="E — SGD steps per epoch (typically equals batches-per-epoch)",
    )
    p.add_argument(
        "--c-b1", type=float, default=1.0,
        help="C_B1 channel-efficiency constant; default 1.0 (data-processing-inequality ceiling)",
    )
    p.add_argument(
        "--c-b2", type=float, default=1.0,
        help="C_B2 constant; default 1.0",
    )
    p.add_argument(
        "--c-b3", type=float, default=1.0,
        help="C_B3 constant; default 1.0",
    )

    args = p.parse_args(argv)
    try:
        result = t3_bound_bits(
            gradient_fraction=args.gradient_fraction,
            gradient_norm_max=args.gradient_norm_max,
            gradient_noise_stddev=args.gradient_noise_stddev,
            lr_decision_interval=args.lr_decision_interval,
            lr_grid_size=args.lr_grid_size,
            window_length=args.window_length,
            batches_per_epoch=args.batches_per_epoch,
            steps_per_epoch=args.steps_per_epoch,
            c_b1=args.c_b1,
            c_b2=args.c_b2,
            c_b3=args.c_b3,
        )
    except ValueError as e:
        print(f"input error: {e}", file=sys.stderr)
        return 2

    print(json.dumps(result, indent=2, ensure_ascii=False))
    return 0


if __name__ == "__main__":
    sys.exit(main())