"""Pre-window API rehearsal: every admission API probe against PRODUCTION, tiny. Coordinator-run between windows (a queue line); never inside a HOLD and never by a lane. It sends the admission's own request bodies (the pp4_admission_requests builders) to production with generation capped at MAX_TOKENS and the admission's priority 10 (background), never waits for production to be idle, and records every response, including every HTTP error body. Reading: production serves the same image and the same request validation as the PP4 candidate (PP4 overlays only remote_dspark, model_runner.py, rejection_sampler_utils.py and its own admission/round-trace modules; no entrypoint or sampling_params file), and both run speculative decoding. A 4xx here means the PROBE is wrong for this image; a probe production accepts but the candidate refuses is a CANDIDATE defect. The tokenize and terminator probes are already tiny and run unmodified with the admission's exact assertions; the chat and cancel probes are capped (request validation does not depend on max_tokens) and their exact checks run only when the capped answer finished. One CONTROL request, the R16 stop probe exactly as PP4-90f cmd/0132 sent it (logit_bias forcing), must come back 400 naming logit_bias and speculative decoding; it confirms the cause on production and never enters the verdict. """ from __future__ import annotations import argparse import hashlib import json import os import subprocess import sys import time import urllib.error import urllib.request from collections.abc import Callable from pathlib import Path from typing import cast from offload.launch.windows import pp4_admission_requests as api from offload.launch.windows import up2_checks as ck from offload.prefill_maxlen.evidence import Json, record PRODUCTION = 'http://HEAD_IP:8000' MAX_TOKENS = 8 TIMEOUT_S = 60.0 CHAT_NAMES = ('grammar', 'penalties', 'json', 'tool', 'reasoning') CONTROL = 'control-r16-logit-bias' ROOT = Path(__file__).resolve().parents[3] Reply = tuple[int | None, dict[str, Json] | None, str | None] def settings(body: dict[str, Json]) -> dict[str, Json]: return {k: v for k, v in body.items() if k not in ('prompt', 'messages')} def capped(body: dict[str, Json]) -> dict[str, Json]: return {**body, 'max_tokens': min(int(str(body['max_tokens'])), MAX_TOKENS)} class Production: """Tiny requests with their own timeout; HTTP errors are logged and recorded, never raised.""" def __init__(self, url: str, timeout: float) -> None: self.url, self.timeout = url, timeout def open(self, endpoint: str, body: dict[str, Json] | None) -> Reply: data = None if body is None else json.dumps(body).encode() request = urllib.request.Request(self.url + endpoint, data, {'Content-Type': 'application/json'}) try: with urllib.request.urlopen(request, timeout=self.timeout) as response: return int(response.status), record(cast(Json, json.load(response))), None except urllib.error.HTTPError as error: refused = api.refused(error, endpoint, body or {}) return refused.status, None, refused.body except (OSError, ValueError) as exc: # transport failure or a non-JSON reply return None, None, f'{type(exc).__name__}: {exc}' def stream_first(self, body: dict[str, Json]) -> tuple[int | None, bool, str | None]: """Open the stream and close at the first content chunk (the cancel probe's disconnect).""" request = urllib.request.Request(self.url + '/v1/chat/completions', json.dumps(body).encode(), {'Content-Type': 'application/json'}) try: with urllib.request.urlopen(request, timeout=self.timeout) as response: status = int(response.status) for line in response: if line.startswith(b'data: {'): choices = ck.rows(record(cast(Json, json.loads(line[6:]))).get('choices', [])) if choices and record(choices[0].get('delta', {})).get('content'): return status, True, None return status, False, None except urllib.error.HTTPError as error: refused = api.refused(error, '/v1/chat/completions', body) return refused.status, False, refused.body except (OSError, ValueError) as exc: return None, False, f'{type(exc).__name__}: {exc}' def row(name: str, endpoint: str, mode: str, body: dict[str, Json], reply: Reply, started: float) -> dict[str, Json]: status, response, error = reply out: dict[str, Json] = {'name': name, 'endpoint': endpoint, 'mode': mode, 'request_sha256': api.sha(cast(Json, body)), 'settings': settings(body), 'status': status, 'error_body': error, 'seconds': round(time.monotonic()-started, 3)} if response is not None: out['response_sha256'] = api.sha(cast(Json, response)) choices = response.get('choices') if isinstance(choices, list) and choices: choice = record(choices[0]) out.update({'finish_reason': choice.get('finish_reason'), 'stop_reason': choice.get('stop_reason')}) if 'usage' in response: out['usage'] = response['usage'] return out def check(out: dict[str, Json], assertion: Callable[..., object], *args: object) -> None: """Run the admission's own assertion; record pass or its STOP message.""" try: assertion(*args) out['check'] = 'pass' except (ValueError, KeyError, TypeError) as exc: out['check'] = 'fail: ' + str(exc) def served_only(served: Json) -> None: ck.require(isinstance(served, list) and [record(m).get('id') for m in served] == [api.MODEL], 'production does not serve exactly '+api.MODEL) def chat_check(name: str, body: dict[str, Json], response: dict[str, Json]) -> None: api.check_chat(name, api.chat_evidence(body, response)[0]) def r16_control(stop: int, eos: int) -> dict[str, Json]: """The R16 (90f0278) stop probe exactly as PP4-90f cmd/0132 sent it: logit_bias forcing.""" body = dict(dict(api.terminator_bodies(stop, eos))['stop']) body.pop('allowed_token_ids', None) return {**body, 'logit_bias': {str(stop): 100}} def spec_decode_refusal(status: Json, error: Json) -> None: ck.require(status == 400 and 'logit_bias' in str(error) and 'speculative decoding' in str(error), 'control not refused for logit_bias under speculative decoding') def plan(eos: int, stop: int | None) -> list[tuple[str, str, str, dict[str, Json]]]: """(name, endpoint, mode, body) in admission order; `stop` None = tokenize not yet answered.""" rows: list[tuple[str, str, str, dict[str, Json]]] = [] for name in CHAT_NAMES[:2]: rows.append((name, '/v1/chat/completions', 'capped', capped(api.chat_body(*api.CHATS[name])))) rows.append(('tokenize', '/tokenize', 'exact', api.TOKENIZE)) for name, body in api.terminator_bodies(-1 if stop is None else stop, eos): ck.require(int(str(body['max_tokens'])) <= MAX_TOKENS, name+': terminator probe is not tiny') rows.append((name, '/v1/completions', 'exact', body)) rows.append((CONTROL, '/v1/completions', 'control', r16_control(-1 if stop is None else stop, eos))) for name in CHAT_NAMES[2:]: rows.append((name, '/v1/chat/completions', 'capped', capped(api.chat_body(*api.CHATS[name])))) rows.append(('cancel', '/v1/chat/completions', 'capped-stream', capped(api.cancel_body()))) return rows def rehearse(server: Production, eos: int, responses: list[dict[str, Json]]) -> list[dict[str, Json]]: rows: list[dict[str, Json]] = [] started = time.monotonic() reply = server.open('/v1/models', None) models = row('models', '/v1/models', 'exact', {}, reply, started) if reply[1] is not None: check(models, served_only, reply[1].get('data')) rows.append(models) stop: int | None = None for name, endpoint, mode, body in plan(eos, None): if name in ('stop', 'eos', CONTROL): body = (r16_control(stop, eos) if name == CONTROL and stop is not None else dict(api.terminator_bodies(-1 if stop is None else stop, eos)).get(name, body)) if name in ('stop', CONTROL) and stop is None: rows.append({'name': name, 'endpoint': endpoint, 'mode': mode, 'status': None, 'check': 'not-run: tokenize gave no single stop token'}) continue started = time.monotonic() if mode == 'capped-stream': status, content, error = server.stream_first(body) out = row(name, endpoint, mode, body, (status, None, error), started) out['content_chunk'] = content if status == 200: check(out, ck.require, content, 'no actual streamed token before client cancel') rows.append(out) continue reply = server.open(endpoint, body) out = row(name, endpoint, mode, body, reply, started) response = reply[1] if mode == 'control': check(out, spec_decode_refusal, reply[0], reply[2]) if response is not None: responses.append({'name': name, 'response': cast(Json, response)}) if name == 'tokenize': check(out, api.stop_token, response) if out['check'] == 'pass': stop = api.stop_token(response) elif mode == 'control': pass elif name in ('stop', 'eos'): check(out, api.check_terminator, name, eos, response) elif out.get('finish_reason') in ('stop', 'tool_calls'): check(out, chat_check, name, body, response) else: out['check'] = f'not-run: capped at {MAX_TOKENS} tokens ({out.get("finish_reason")})' rows.append(out) return rows def verdict(rows: list[dict[str, Json]]) -> dict[str, Json]: controls = [r for r in rows if r.get('mode') == 'control'] rows = [r for r in rows if r.get('mode') != 'control'] refused = [str(r['name']) for r in rows if isinstance(r.get('status'), int) and 400 <= int(str(r['status'])) < 500] broken = [str(r['name']) for r in rows if (r.get('status') is None and 'not-run' not in str(r.get('check', ''))) or (isinstance(r.get('status'), int) and int(str(r['status'])) >= 500)] failed = [str(r['name']) for r in rows if str(r.get('check', '')).startswith('fail')] blocked = [str(r['name']) for r in rows if r.get('status') is None and 'not-run' in str(r.get('check', ''))] reading: list[Json] = [] if refused: reading.append('PROBE WRONG FOR THIS IMAGE: production refused '+', '.join(refused)+' (see error_body)') if broken: reading.append('NO ANSWER (transport or 5xx): '+', '.join(broken)+'; rerun, this is not a probe verdict') if failed: reading.append('PROBE SEMANTICS FAIL on production: '+', '.join(failed)) if blocked: reading.append('NOT RUN (an earlier probe failed): '+', '.join(blocked)) confirmed = [r.get('check') == 'pass' for r in controls if 'not-run' not in str(r.get('check', ''))] cause: bool | None = all(confirmed) if confirmed else None reading.append({True: 'CAUSE CONFIRMED on production: the R16 logit_bias probe is refused under speculative decoding', False: 'CAUSE NOT REPRODUCED: production did not refuse the R16 logit_bias probe as expected (see control row)', None: 'cause control not run'}[cause]) ok = not (refused or broken or failed or blocked) return {'verdict': 'PASS' if ok else 'FAIL', 'cause_confirmed': cause, 'refused': cast(Json, refused), 'no_answer': cast(Json, broken), 'semantic_failures': cast(Json, failed), 'not_run': cast(Json, blocked), 'reading': cast(Json, reading if not ok else ['every admission API probe accepted by production; exact probes exact', *reading])} def source() -> dict[str, Json]: def git(*args: str) -> str: return subprocess.run(['git', '-C', str(ROOT), *args], check=True, capture_output=True, text=True).stdout.strip() files = {p: hashlib.sha256((ROOT/p).read_bytes()).hexdigest() for p in ('offload/launch/windows/pp4_admission_requests.py', 'offload/launch/windows/pp4_api_rehearsal.py')} return {'commit': git('rev-parse', 'HEAD'), 'dirty': bool(git('status', '--porcelain', '--', *files)), 'sha256': cast(Json, files)} def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--execute', action='store_true') parser.add_argument('--url', default=PRODUCTION) parser.add_argument('--eos', type=int, required=True, help="the model's verified EOS id (admission-eos-source.json)") parser.add_argument('--out', type=Path, required=True, help='fresh directory for rehearsal.json + responses.jsonl') parser.add_argument('--timeout', type=float, default=TIMEOUT_S) args = parser.parse_args(argv) ck.require(args.eos >= 0, 'verified model EOS token required') if not args.execute: print(json.dumps({'dry': True, 'url': args.url, 'max_tokens_cap': MAX_TOKENS, 'probes': [ {'name': n, 'endpoint': e, 'mode': m, 'settings': settings(b)} for n, e, m, b in plan(args.eos, None)]}, sort_keys=True)) return 0 ck.require(os.environ.get('WINDOW_COORDINATOR') == 'Deepseek/coordinator', 'coordinator required') ck.require(not args.out.exists(), 'fresh rehearsal directory required') provenance = source() args.out.mkdir(parents=True) responses: list[dict[str, Json]] = [] started = time.time() rows = rehearse(Production(args.url.rstrip('/'), args.timeout), args.eos, responses) result: dict[str, Json] = {**verdict(rows), 'url': args.url, 'eos': args.eos, 'max_tokens_cap': MAX_TOKENS, 'started_unix': started, 'seconds': round(time.time()-started, 3), 'source': provenance, 'probes': cast(Json, rows)} held = args.out/'responses.jsonl' held.touch(mode=0o600) held.write_text(''.join(json.dumps(r, sort_keys=True)+'\n' for r in responses)) (args.out/'rehearsal.json').write_text(json.dumps(result, indent=1, sort_keys=True, allow_nan=False)+'\n') print(json.dumps({k: result[k] for k in ('verdict', 'cause_confirmed', 'reading', 'refused', 'semantic_failures')}, sort_keys=True)) return 0 if result['verdict'] == 'PASS' else 1 if __name__ == '__main__': sys.exit(main())