Check speculative decoding correctness and its cost

from __future__ import annotations

import json
import math

Vector = list[float]


def validate_distribution(values: Vector) -> None:
    if (
        not values
        or any(not math.isfinite(v) or v < 0 for v in values)
        or not math.isclose(math.fsum(values), 1.0, abs_tol=1e-8)
    ):
        raise ValueError("Expected a finite normalized probability distribution")


def residual_distribution(p: Vector, q: Vector) -> Vector:
    """Compute normalized [p(x) - q(x)]_+ required for lossless speculative rejection sampling."""
    validate_distribution(p)
    validate_distribution(q)
    if len(p) != len(q):
        raise ValueError("Vocabulary sizes differ")
    pos = [max(0.0, pv - qv) for pv, qv in zip(p, q)]
    total = sum(pos)
    if total <= 1e-15:
        return list(p)
    return [val / total for val in pos]


def marginal_output_distribution(
    p: Vector, q: Vector, use_residual: bool = True
) -> Vector:
    """Compute the exact marginal probability P(Y = y) of one speculative verification step."""
    validate_distribution(p)
    validate_distribution(q)
    if len(p) != len(q):
        raise ValueError("Vocabulary sizes differ")
    vocab = len(p)
    accepted_mass = [q[y] * min(1.0, p[y] / max(q[y], 1e-15)) for y in range(vocab)]
    reject_prob = 1.0 - sum(accepted_mass)
    resample_dist = residual_distribution(p, q) if use_residual else p
    return [accepted_mass[y] + reject_prob * resample_dist[y] for y in range(vocab)]


def total_variation_distance(p: Vector, q: Vector) -> float:
    return round(0.5 * sum(abs(pv - qv) for pv, qv in zip(p, q)), 8)


def _sample_from_cdf(dist: Vector, u: float) -> int:
    acc = 0.0
    for idx, prob in enumerate(dist):
        acc += prob
        if u < acc:
            return idx
    return len(dist) - 1


def verify_speculative_draft(
    draft_tokens: list[int],
    draft_probs: list[Vector],
    target_probs: list[Vector],
    uniform_draws: list[float],
    resample_draws: list[float],
    use_residual: bool = True,
) -> dict[str, object]:
    """Verify gamma draft tokens in one target pass using lossless rejection sampling.

    - target_probs has length gamma + 1 (positions 0..gamma-1 plus the bonus position gamma).
    - Returns accepted count, emitted token IDs, and step trace.
    """
    gamma = len(draft_tokens)
    if len(draft_probs) != gamma or len(target_probs) != gamma + 1:
        raise ValueError(
            "target_probs must have length gamma + 1 and draft_probs length gamma"
        )

    if len(uniform_draws) != gamma or len(resample_draws) != gamma + 1:
        raise ValueError("Draw counts do not match draft length")
    if any(not 0 <= draw < 1 for draw in uniform_draws + resample_draws):
        raise ValueError("Draws must be in [0,1)")
    for vector in draft_probs + target_probs:
        validate_distribution(vector)
    width = len(target_probs[0])
    if any(len(vector) != width for vector in draft_probs + target_probs):
        raise ValueError("Vocabulary sizes differ")
    if any(
        not 0 <= token < width or draft_probs[i][token] <= 0
        for i, token in enumerate(draft_tokens)
    ):
        raise ValueError("Draft token must have positive proposal probability")
    emitted: list[int] = []
    trace: list[dict[str, object]] = []
    accepted_draft_tokens = 0

    for i in range(gamma):
        tok = draft_tokens[i]
        p_tok = target_probs[i][tok]
        q_tok = max(draft_probs[i][tok], 1e-12)
        accept_ratio = min(1.0, p_tok / q_tok)
        u = uniform_draws[i]

        if u < accept_ratio:
            emitted.append(tok)
            accepted_draft_tokens += 1
            trace.append(
                {
                    "position": i,
                    "draft_token": tok,
                    "p_target": round(p_tok, 4),
                    "q_draft": round(q_tok, 4),
                    "accept_prob": round(accept_ratio, 4),
                    "u_draw": round(u, 4),
                    "outcome": "accepted",
                    "emitted_token": tok,
                }
            )
        else:
            fallback_dist = (
                residual_distribution(target_probs[i], draft_probs[i])
                if use_residual
                else target_probs[i]
            )
            corrected_tok = _sample_from_cdf(fallback_dist, resample_draws[i])
            emitted.append(corrected_tok)
            trace.append(
                {
                    "position": i,
                    "draft_token": tok,
                    "p_target": round(p_tok, 4),
                    "q_draft": round(q_tok, 4),
                    "accept_prob": round(accept_ratio, 4),
                    "u_draw": round(u, 4),
                    "outcome": "rejected_and_resampled",
                    "emitted_token": corrected_tok,
                }
            )
            return {
                "gamma": gamma,
                "accepted_draft_tokens": accepted_draft_tokens,
                "emitted_tokens": emitted,
                "total_emitted": len(emitted),
                "trace": trace,
            }

    # All gamma draft tokens were accepted; sample one bonus token from target_probs[gamma]
    bonus_tok = _sample_from_cdf(target_probs[gamma], resample_draws[-1])
    emitted.append(bonus_tok)
    trace.append(
        {
            "position": gamma,
            "draft_token": None,
            "outcome": "bonus_token_from_target",
            "emitted_token": bonus_tok,
        }
    )
    return {
        "gamma": gamma,
        "accepted_draft_tokens": accepted_draft_tokens,
        "emitted_tokens": emitted,
        "total_emitted": len(emitted),
        "trace": trace,
    }


def expected_tokens_per_step(alpha: float, gamma: int) -> float:
    """Closed-form expected tokens generated per verification step for acceptance rate alpha."""
    if not (0.0 <= alpha <= 1.0):
        raise ValueError("alpha must be in [0, 1]")
    if gamma <= 0:
        raise ValueError("gamma must be positive")
    if abs(1.0 - alpha) < 1e-9:
        return float(gamma + 1)
    return round((1.0 - (alpha ** (gamma + 1))) / (1.0 - alpha), 4)


def estimate_forward_pass_ms(
    num_active_tokens: int,
    base_weight_read_ms: float,
    compute_ms_per_token: float,
    ridge_tokens: int = 48,
) -> float:
    """Piecewise roofline step time for a target or draft forward pass."""
    extra_tokens = max(0, num_active_tokens - ridge_tokens)
    return round(base_weight_read_ms + extra_tokens * compute_ms_per_token, 4)


def speculative_speedup_by_batch(
    gamma: int = 4,
    alphas: tuple[float, ...] = (0.80, 0.45),
    batch_sizes: tuple[int, ...] = (1, 2, 4, 8, 16, 32, 64),
) -> list[dict[str, object]]:
    """Compare standard target decode vs speculative decoding across batch sizes and acceptance rates."""
    # Target 70B FP8 on 2xH100: 22.0 ms weight load, 0.28 ms/token compute above ridge=48
    # Draft 8B FP8 on 2xH100: 2.6 ms weight load, 0.032 ms/token compute above ridge=48
    rows: list[dict[str, object]] = []
    for b in batch_sizes:
        std_step_ms = estimate_forward_pass_ms(
            num_active_tokens=b,
            base_weight_read_ms=22.0,
            compute_ms_per_token=0.42,
            ridge_tokens=32,
        )
        draft_pass_ms = estimate_forward_pass_ms(
            num_active_tokens=b,
            base_weight_read_ms=2.6,
            compute_ms_per_token=0.045,
            ridge_tokens=32,
        )
        verify_tokens = b * (gamma + 1)
        target_verify_ms = estimate_forward_pass_ms(
            num_active_tokens=verify_tokens,
            base_weight_read_ms=22.0,
            compute_ms_per_token=0.42,
            ridge_tokens=32,
        )
        spec_cycle_ms = round(gamma * draft_pass_ms + target_verify_ms, 2)

        row: dict[str, object] = {
            "batch_size": b,
            "verify_batched_tokens": verify_tokens,
            "standard_decode_ms_per_token": round(std_step_ms, 2),
            "spec_cycle_ms": spec_cycle_ms,
        }
        for alpha in alphas:
            exp_tokens = expected_tokens_per_step(alpha, gamma)
            eff_ms_per_tok = round(spec_cycle_ms / exp_tokens, 2)
            speedup = round(std_step_ms / eff_ms_per_tok, 2)
            tag = f"alpha_{int(alpha * 100)}"
            row[f"{tag}_expected_tokens"] = exp_tokens
            row[f"{tag}_ms_per_token"] = eff_ms_per_tok
            row[f"{tag}_speedup"] = speedup
        rows.append(row)
    return rows


def run_verification_demo() -> dict[str, object]:
    # 4-token vocabulary example over gamma = 4 draft positions
    draft_tokens = [0, 1, 2, 1]
    draft_probs = [
        [0.65, 0.20, 0.10, 0.05],
        [0.10, 0.70, 0.15, 0.05],
        [0.05, 0.15, 0.60, 0.20],
        [0.15, 0.55, 0.20, 0.10],
    ]
    target_probs = [
        [0.70, 0.18, 0.08, 0.04],  # pos 0: p(0)=0.70 >= q(0)=0.65 -> accept_prob = 1.0
        [0.12, 0.68, 0.14, 0.06],  # pos 1: p(1)=0.68, q(1)=0.70 -> accept_prob = 0.9714
        [
            0.10,
            0.48,
            0.18,
            0.24,
        ],  # pos 2: p(2)=0.18 < q(2)=0.60 -> accept_prob = 0.30 (rejected!)
        [0.20, 0.50, 0.20, 0.10],  # pos 3: discarded after pos 2 rejection
        [0.25, 0.25, 0.25, 0.25],  # bonus pos 4
    ]
    uniform_draws = [0.42, 0.51, 0.74, 0.19]
    resample_draws = [0.30, 0.30, 0.45, 0.30, 0.30]

    step_result = verify_speculative_draft(
        draft_tokens=draft_tokens,
        draft_probs=draft_probs,
        target_probs=target_probs,
        uniform_draws=uniform_draws,
        resample_draws=resample_draws,
        use_residual=True,
    )

    p_pos2 = target_probs[2]
    q_pos2 = draft_probs[2]
    exact_marginal = marginal_output_distribution(p_pos2, q_pos2, use_residual=True)
    buggy_marginal = marginal_output_distribution(p_pos2, q_pos2, use_residual=False)

    return {
        "verification_step": step_result,
        "distribution_check_pos2": {
            "target_p": p_pos2,
            "draft_q": q_pos2,
            "residual_p_minus_q_norm": [
                round(x, 4) for x in residual_distribution(p_pos2, q_pos2)
            ],
            "tv_distance_with_residual": total_variation_distance(
                p_pos2, exact_marginal
            ),
            "tv_distance_without_residual": total_variation_distance(
                p_pos2, buggy_marginal
            ),
        },
        "batch_crossover": speculative_speedup_by_batch(gamma=4),
    }


def main() -> None:
    report = run_verification_demo()
    print(
        json.dumps(
            {
                "execution_mode": "CPU algorithm with controlled inputs; any latency/capacity coefficients are assumptions",
                **report,
            },
            indent=2,
        )
    )


if __name__ == "__main__":
    main()

from __future__ import annotations

import unittest

from lab import (
    expected_tokens_per_step,
    marginal_output_distribution,
    run_verification_demo,
    speculative_speedup_by_batch,
    total_variation_distance,
)


class TestSpeculativeDecodingLab(unittest.TestCase):
    def test_greedy_threshold_without_residual_breaks_target_distribution(self) -> None:
        p = [0.10, 0.48, 0.18, 0.24]
        q = [0.05, 0.15, 0.60, 0.20]
        exact_marginal = marginal_output_distribution(p, q, use_residual=True)
        buggy_marginal = marginal_output_distribution(p, q, use_residual=False)
        self.assertLessEqual(total_variation_distance(p, exact_marginal), 1e-7)
        self.assertGreater(total_variation_distance(p, buggy_marginal), 0.12)

    def test_high_batch_speculative_verification_slows_down_step_time(self) -> None:
        rows = {row["batch_size"]: row for row in speculative_speedup_by_batch(gamma=4)}
        self.assertGreater(rows[1]["alpha_80_speedup"], 2.0)
        self.assertLess(rows[32]["alpha_80_speedup"], 1.0)
        self.assertLessEqual(rows[64]["alpha_80_speedup"], 0.75)

    def test_rejection_truncates_remaining_draft_and_resamples_residual(self) -> None:
        demo = run_verification_demo()["verification_step"]
        self.assertEqual(demo["accepted_draft_tokens"], 2)
        self.assertEqual(demo["total_emitted"], 3)
        self.assertEqual(demo["emitted_tokens"], [0, 1, 1])

    def test_expected_tokens_formula_matches_boundaries(self) -> None:
        self.assertEqual(expected_tokens_per_step(0.0, 4), 1.0)
        self.assertEqual(expected_tokens_per_step(1.0, 4), 5.0)
        self.assertAlmostEqual(expected_tokens_per_step(0.80, 4), 3.3616, places=4)


class ProbabilityValidationTests(unittest.TestCase):
    def test_invalid_probability_distribution_is_rejected(self):
        from lab import residual_distribution

        for vector in [[0.5, -0.5], [float("nan"), 1], [0.2, 0.2]]:
            with self.assertRaises(ValueError):
                residual_distribution(vector, [0.5, 0.5])

    def test_zero_draw_cannot_select_zero_probability_token(self):
        from lab import _sample_from_cdf

        self.assertEqual(_sample_from_cdf([0, 1], 0), 1)


if __name__ == "__main__":
    unittest.main()


class GuardBoundaryTests(unittest.TestCase):
    def test_rejects_invalid_boundary_inputs_with_specific_reason(self):
        import lab
        cases = [('residual_distribution([1],[0.5,0.5])', 'Vocabulary sizes'), ('marginal_output_distribution([1],[0.5,0.5], use_residual=False)', 'Vocabulary sizes'), ('verify_speculative_draft([0],[],[[1],[1]],[0],[0,0])', 'target_probs must have length'), ('verify_speculative_draft([0],[[1]],[[1],[1]],[],[0,0])', 'Draw counts'), ('verify_speculative_draft([0],[[1]],[[1],[1]],[1],[0,0])', 'Draws must'), ('verify_speculative_draft([0],[[1]],[[0.5,0.5],[1]],[0],[0,0])', 'Vocabulary sizes'), ('verify_speculative_draft([0],[[0,1]],[[0.5,0.5],[0.5,0.5]],[0],[0,0])', 'positive proposal'), ('expected_tokens_per_step(-1,1)', 'alpha must'), ('expected_tokens_per_step(0.5,0)', 'gamma must')]
        for code, reason in cases:
            with self.subTest(code=code):
                with self.assertRaisesRegex((ValueError, MemoryError, KeyError, RuntimeError), reason):
                    exec(code, vars(lab).copy())