File size: 6,533 Bytes
e1acc15 | 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 | """Round6 budget: admission proves instrumentation; measurement prices no profiles."""
from __future__ import annotations
import math
from pathlib import Path
from typing import cast
from offload.launch.windows import pf1_window as pf
from offload.launch.windows import up2_checks as ck
from offload.prefill_maxlen.evidence import Json, number, record
PATH = pf.ROOT/'offload/dspark/probabilistic/pp4/round6-budget.json'
MAX_HOLD_S, ABORT_S = 85*60, 73*60
ALL_CELLS = ('short-b1', '32k-b1', '32k-b4')
B1_CELLS = ALL_CELLS[:2]
RESTORED = 'RESTORED: K1 START VERIFIED, greedy 6/6 IDENTICAL, PRODUCTION HOLD OK, window closed'
def pair_reserve(admission_path: Path, expected: str) -> dict[str, Json]:
"""Use only the successful admission HOLD's own hash-bound physical step.
There is no pointer, solo receipt, estimate override or old pair-only fallback
on measurement admission. The600s bound remains planning/admission-only.
"""
data = ck.read(admission_path)
ck.require(data.get('schema') == 'pp4-live-admission-v1' and data.get('source_commit') == expected
and data.get('verdict') == 'QUALIFIED' and not data.get('stop')
and record(data['restore']).get('status') == RESTORED
and record(data['restore']).get('restore_deadline_met') is True
and record(data['post_close_watch']).get('status') == 'PASS', 'successful restored/watched admission HOLD required for measured pair duration')
evidence = record(data['physical_pair_evidence'])
admission_run = data.get('admission_run')
ck.require(isinstance(admission_run, str) and bool(admission_run.strip())
and evidence.get('run') == admission_run, 'physical pair must match parent admission run')
path = Path(str(evidence['evidence_path']))
ck.require(path.is_absolute() and path.parent == admission_path.resolve().parent
and path.resolve().parent == admission_path.resolve().parent
and path.name == 'physical-pair-qualification.json' and path.is_file() and pf.digest(path) == evidence['sha256'],
'admission physical-pair primary evidence absent/changed')
physical = ck.read(path)
from offload.dspark.probabilistic.pp4.pair_contract import validate
from offload.dspark.probabilistic.pp4.pair_qualify import SOURCES
validate(physical, expected, cast(str, admission_run))
ck.require(physical['source_sha256'] == {name: pf.digest(pf.ROOT/name) for name in SOURCES},
'admission pair executed different sources')
ck.require(record(data['artifact_sha256']).get(path.name) == evidence['sha256'],
'physical pair is not an artifact of this admission HOLD')
seconds = number(physical['seconds'])
ck.require(math.isfinite(seconds) and seconds > 0, 'missing MEASURED admission pair duration')
return {'seconds': math.ceil(max(180., seconds*1.5)), 'measured_driver_s': seconds,
'classification': 'ESTIMATED reserve = max(180s, MEASURED admission pair driver +50%) rounded up',
'source': str(path), 'sha256': pf.digest(path),
'admission_receipt': str(admission_path), 'admission_receipt_sha256': pf.digest(admission_path)}
def estimate(*, pair: dict[str, Json] | None = None) -> dict[str, Json]:
data = ck.read(PATH)
ck.require(data.get('schema') == 'pp4-round6-budget-v1', 'registered budget schema differs')
plan = record(data['cells'])
ck.require(set(plan) == set(ALL_CELLS), 'registered budget cell set differs')
prices: dict[str, float] = {}
for name, value in plan.items():
row = record(value)
observations = ck.rows(row['observations'])
scale = 4/3 if name == '32k-b4' else 1.
ttft = max(number(r['max_ttft_s']) for r in observations)*scale
span = max(number(r['max_decode_span_s']) for r in observations)*scale
price = math.ceil((ttft+span*511/1499+7)/5)*5
ck.require(row.get('allowance_per_group_s') == price and row.get('batch_scaling') == scale,
'budget must preserve cold TTFT and registered512-gap scaling')
groups = 2 if name == '32k-b4' else 4
ck.require(row.get('groups_per_start') == groups and row.get('profile_groups_per_start') == 0,
'Round6 retains headline groups and omits measurement profile companions')
prices[name] = price*groups
reserve: dict[str, Json] = {'seconds': 600, 'classification': 'ESTIMATED bound for dry planning/admission only; measured admission receipt still required'} if pair is None else pair
pair_s = number(reserve['seconds'])
ck.require(math.isfinite(pair_s) and pair_s > 0, 'invalid pair reserve')
common = (number(data['worker_startup_s'])+max(number(x) for x in cast(list[Json], data['startup_observations_s']))
+number(data['warm_per_start_s'])+number(data['transition_per_start_s']))
def price_schedule(names: tuple[str, ...]) -> tuple[dict[str, Json], float]:
block = common+sum(prices[name] for name in names)
blocks: dict[str, Json] = {'s1-A': block+number(data['cuts_s']),
's2-B': block+number(data['greedy_on_s']), 's3-A': block}
total = number(data['setup_s'])+pair_s+sum(number(v) for v in blocks.values())+number(data['restore_s'])
return blocks, total
blocks, full_total = price_schedule(ALL_CELLS)
names = ALL_CELLS if full_total <= MAX_HOLD_S else B1_CELLS
blocks, total = price_schedule(names)
return {'classification': data['classification'], 'registration': 'pp4-round6-budget-v1',
'source_sha256': pf.digest(PATH), 'pair': reserve, 'block_budget_s': blocks,
'cells': list(names), 'omitted_cells': [] if names == ALL_CELLS else ['32k-b4'],
'all_cells_total_s': full_total, 'headline_per_start_s': sum(prices[name] for name in names),
'profile_per_start_s': 0, 'full_round_p95_guard': 'omitted in measurement; instrumentation qualified in admission',
'total_s': total, 'total_minutes': total/60, 'fits': total <= MAX_HOLD_S,
'restore_s': data['restore_s'], 'abort_s': ABORT_S, 'restore_deadline_s': MAX_HOLD_S}
def admitted(admission_path: Path, expected: str) -> dict[str, Json]:
result = estimate(pair=pair_reserve(admission_path, expected))
ck.require(result['fits'] is True,
f"registered measurement ESTIMATED {number(result['total_minutes']):.2f}min exceeds85min after B4 omission; refuses before host contact")
return result
|