| """WIN2-R3 seeded A/A pilot recipe, an end-of-restore add-on (F-METHOD §3.1 pilot and §3.4 A/A band; no extra HOLD). |
| |
| Two starts of the same production configuration, each running the identical seeded cells and the distributional probe: |
| - start A: the window's LAST control start (production configuration on the experiment port), after its own cells; |
| - start B: restored production, inside the window's HOLD, between W8 `check` (PASS) and W8 `close`. W8 itself and its |
| greedy-identity gate are unchanged; the pilot only reads. |
| `run` measures one start; `compare(A, B)` gives the verdict and the numbers the other rules register: |
| - PASS (single profile): every B1 seeded cell has identical output digests AND identical per-request round counts in |
| both starts (determinism of seeded speculative decoding at B1). B >= 2 cells are reported, not required (production |
| kernels are not batch-invariant). |
| - per cell and metric: the paired-difference CV (for `resolution.py --diff-cv`) and the start difference (one A/A, so a |
| 1-df between-start SD estimate for `decide_paired_window(start_sd=...)`; accumulate over pilots before relying on it); |
| - the distributional A/A comparison (feed `dist_gate.register_band`). |
| DP-2 bridging cell (98aa385; start A only, so it needs no extra HOLD): bench2 legal-short and bench2 summary-32K at 1,500 |
| tokens, the corpus short and 32k-0 prompts at 512 and 1,500 tokens, three seeded pairs each at production settings (T 1, |
| thinking high, priority 10), report-only (`bridge_report`). It separates answer length from date on the corpus prompts and |
| bridges the bench2 prompts across 10-24..26; a summary-32K reading near 1.06 instead of DP-2's 1.589 would point to a |
| production change (H4). Every prompt is pinned by its sha256 (window_cells.PROMPT_SHA256); the bench2 prompts are rebuilt |
| with bench2's own algorithm (`bench2_texts`) and refused unless they hash to DP-2's registered values. |
| COST (ESTIMATED): seeded cells ~5 min per start (short-b1 38 s, 32k-b1 56 s, 32k-b2 88 s, 32k-b3 120 s), distributional |
| probe ~6.5 min per start for the default `short,32k` subset (one cold prefill per prompt; F-METHOD's "+2 min" covers |
| only the short prompts), bridge ~8.6 min of cells plus up to ~1 min of /tokenize calls rebuilding summary-32K (start A). |
| Coordinator-only; `run` without --execute prints the plan. |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| import ast |
| import hashlib |
| import json |
| import math |
| import os |
| import random |
| import statistics |
| import sys |
| from collections.abc import Mapping, Sequence |
| from pathlib import Path |
| from typing import Protocol, cast |
|
|
| from offload.gb10_budget import window_cells as wc |
| from offload.launch.windows import decision_v2 as d |
| from offload.launch.windows import dist_gate as dg |
| from offload.prefill_maxlen.evidence import Json, number, record |
|
|
| SCHEMA = 'dsv41-aa-pilot-v1' |
| PILOT_CELLS = ('seeded-short-b1', 'seeded-32k-b1', 'seeded-32k-b2', 'seeded-32k-b3') |
| METRICS = {'decode_tps': True, 'ms_per_round': False, 'accepted_per_round': True} |
| DIST_SUBSET = ('short', '32k') |
| COORDINATOR = 'Deepseek/coordinator' |
|
|
| |
| BENCH2 = Path(__file__).resolve().parents[2] / 'bench/bench2.py' |
| BENCH2_SEED = 42 |
| BENCH2_SUMMARY_CTX = 32000 |
| BENCH2_TITLE = 'MATERIALS (public court documents)' |
| |
| |
| BRIDGE_CELLS = tuple(wc.SeededCell(name, (key,), seed_base=base, pairs=3, max_tokens=tokens, prefill_s=prefill, |
| pair_s=32.0 if tokens == 1500 else 11.0, report_only=True) |
| for name, key, tokens, base, prefill in ( |
| ('bridge-bench2-legal-1500', 'bench2-legal-short', 1500, 201_000, 1.0), |
| ('bridge-bench2-summary32k-1500', 'bench2-summary-32k', 1500, 202_000, 20.0), |
| ('bridge-short-512', 'short', 512, 203_000, 2.0), |
| ('bridge-short-1500', 'short', 1500, 204_000, 2.0), |
| ('bridge-32k0-512', '32k-0', 512, 205_000, 20.0), |
| ('bridge-32k0-1500', '32k-0', 1500, 206_000, 20.0))) |
| |
| DP2_REFERENCE = { |
| 'bridge-bench2-legal-1500': (1.002, 0.986, 1.018, '10-22..24, n 77'), |
| 'bridge-bench2-summary32k-1500': (1.589, 1.563, 1.615, '10-22..24, n 70'), |
| 'bridge-short-512': (1.294, 1.232, 1.355, '10-27, n 40'), |
| 'bridge-short-1500': (1.146, 1.089, 1.202, '10-26, n 8'), |
| 'bridge-32k0-512': (1.048, 1.000, 1.096, '10-27, n 40'), |
| 'bridge-32k0-1500': (0.949, 0.861, 1.036, '10-26, n 8')} |
| |
| H4_PRODUCTION_CHANGE = 1.06 |
|
|
|
|
| class Counter(Protocol): |
| def tokens(self, text: str) -> int: ... |
| def raw_tokens(self, text: str) -> int: ... |
|
|
|
|
| class BridgeHttp(wc.SeededHttp): |
| def raw_tokens(self, text: str) -> int: |
| return int(number(self.post('/tokenize', {'model': self.model()['id'], 'prompt': text})['count'])) |
|
|
|
|
| def bench2_workloads(path: Path = BENCH2) -> dict[str, str]: |
| """bench2's WORKLOADS literal, read without importing bench2 (its telemetry import exists only on the hosts).""" |
| for node in ast.parse(path.read_text()).body: |
| if isinstance(node, ast.Assign) and any(isinstance(t, ast.Name) and t.id == 'WORKLOADS' for t in node.targets): |
| value = ast.literal_eval(node.value) |
| if isinstance(value, dict) and all(isinstance(k, str) and isinstance(v, str) for k, v in value.items()): |
| return dict(value) |
| raise ValueError(f'{path}: no WORKLOADS literal') |
|
|
|
|
| def bench2_docs(data_dir: Path) -> list[dict[str, str]]: |
| docs: list[dict[str, str]] = [] |
| for name in ('opinions_cl.jsonl', 'filings_recap.jsonl'): |
| for line in (data_dir / name).read_text().splitlines(): |
| row = record(cast(Json, json.loads(line))) |
| docs.append({k: str(row.get(k, '')) for k in ('title', 'court', 'date', 'text')}) |
| return docs |
|
|
|
|
| def bench2_record(api: Counter, docs: Sequence[Mapping[str, str]], target_tokens: int, seed: int) -> str: |
| """bench2 build_real, step for step (its trim loop included): the text must come out byte-identical.""" |
| order = list(docs) |
| random.Random(seed).shuffle(order) |
| parts: list[str] = [] |
| chars = 0 |
| ratio = 3.3 |
| for d in order: |
| parts.append(f"=== {d.get('title', '')} ({d.get('court', '')}, {d.get('date', '')}) ===\n{d['text']}") |
| chars += len(parts[-1]) |
| if chars / ratio >= target_tokens * 1.02: |
| n = api.raw_tokens('\n\n'.join(parts)) |
| ratio = chars / n |
| if n >= target_tokens: |
| break |
| text = '\n\n'.join(parts) |
| while api.raw_tokens(text) > target_tokens: |
| text = text[: int(len(text) - (api.raw_tokens(text) - target_tokens) * ratio) - 50] |
| return text |
|
|
|
|
| def bench2_prompt(api: Counter, docs: Sequence[Mapping[str, str]], ctx: int, instruction: str, seed: int) -> tuple[str, int]: |
| """bench2 make_prompt (filler 'real'): the longest record prefix whose chat prompt fits ctx tokens.""" |
| rec = bench2_record(api, docs, ctx + 4096, seed) |
|
|
| def wrap(length: int) -> str: |
| return f'{BENCH2_TITLE}:\n{rec[:length]}\n\nEND OF MATERIALS.\n\n{instruction}' |
| lo, hi = 0, len(rec) |
| best, count = wrap(0), api.tokens(wrap(0)) |
| if count > ctx: |
| raise ValueError(f'requested context {ctx} is smaller than instruction/template ({count})') |
| while lo <= hi: |
| middle = (lo + hi) // 2 |
| candidate = wrap(middle) |
| n = api.tokens(candidate) |
| if n <= ctx: |
| best, count, lo = candidate, n, middle + 1 |
| else: |
| hi = middle - 1 |
| if ctx - count > 16: |
| raise ValueError(f'corpus cannot fill requested context {ctx}; measured {count}') |
| return best, count |
|
|
|
|
| def bench2_texts(api: Counter, data_dir: Path) -> dict[str, tuple[str, int]]: |
| """The two bench2 prompts DP-2 measured, rebuilt and refused unless they hash to the registered values.""" |
| workloads = bench2_workloads() |
| legal = workloads['legal'] |
| out = {'bench2-legal-short': (legal, api.tokens(legal)), |
| 'bench2-summary-32k': bench2_prompt(api, bench2_docs(data_dir), BENCH2_SUMMARY_CTX, workloads['summary'], |
| BENCH2_SEED)} |
| for key, (text, _) in out.items(): |
| sha = hashlib.sha256(text.encode()).hexdigest() |
| if sha != wc.PROMPT_SHA256[key]: |
| raise ValueError(f'{key}: rebuilt prompt hashes to {sha[:12]}, not DP-2\'s registered {wc.PROMPT_SHA256[key][:12]} ' |
| '(bench data, tokenizer or template changed); the bridge never runs another prompt') |
| return out |
|
|
|
|
| def bridge_report(results: Mapping[str, Json]) -> dict[str, Json]: |
| """Per bridge cell: MEASURED accepted/round (mean of the three seeded pairs, per pair) beside DP-2's reference, the |
| 512-vs-1,500 difference per corpus prompt on one day, and the H4 flag. Report only: it never decides anything.""" |
| cells: dict[str, Json] = {} |
| means: dict[str, float] = {} |
| for cell in BRIDGE_CELLS: |
| if cell.name not in results: |
| continue |
| values = wc.seeded_values(record(results[cell.name]), 'accepted_per_round') |
| means[cell.name] = statistics.fmean(values) |
| ref, low, high, where = DP2_REFERENCE[cell.name] |
| cells[cell.name] = {'prompt': cell.prompts[0], 'max_tokens': cell.max_tokens, |
| 'accepted_per_round': means[cell.name], 'per_pair': cast(Json, values), |
| 'dp2_reference': {'mean': ref, 'ci95': [low, high], 'when': where}, |
| 'within_dp2_ci': low <= means[cell.name] <= high} |
| length: dict[str, Json] = {} |
| for key, short, long in (('short', 'bridge-short-512', 'bridge-short-1500'), ('32k-0', 'bridge-32k0-512', 'bridge-32k0-1500')): |
| if short in means and long in means: |
| length[key] = {'same_day_512_minus_1500': means[short] - means[long], |
| 'dp2_cross_day_512_minus_1500': DP2_REFERENCE[short][0] - DP2_REFERENCE[long][0]} |
| summary = means.get('bridge-bench2-summary32k-1500') |
| flag: Json = None |
| if summary is not None: |
| midpoint = (H4_PRODUCTION_CHANGE + DP2_REFERENCE['bridge-bench2-summary32k-1500'][0]) / 2 |
| flag = ('H4 FLAG: summary-32K reads {:.3f}, nearer 1.06 than DP-2\'s 1.589: a production change between 10-24 and ' |
| '10-26, not prompt identity'.format(summary) if summary < midpoint else |
| 'summary-32K reads {:.3f}: consistent with prompt identity (DP-2 H1), no production change'.format(summary)) |
| return {'schema': SCHEMA, 'type': 'bridge', 'cells': cells, 'length_vs_date': length, 'h4': flag, |
| 'status': 'MEASURED seeded pairs (3 per cell), report only'} |
|
|
|
|
| def run(seeded_api: wc.SeededApi, dist_api: dg.DistApi, texts: dict[str, tuple[str, int]], plan: wc.Plan, |
| materials: str, dist_rows: Mapping[str, Json], *, cells: Sequence[str] = PILOT_CELLS, |
| start: str = 'A', bridge_texts: Mapping[str, tuple[str, int]] | None = None) -> dict[str, Json]: |
| """One start: every pilot seeded cell, (start A) the DP-2 bridging cell, then the distributional probe (same salt |
| prefix as the plan). Start A must carry `bridge_texts` (the two bench2 prompts, `bench2_texts`); start B never runs |
| the bridge.""" |
| if start not in ('A', 'B'): |
| raise ValueError(f'pilot start must be A (last control start) or B (restored production), not {start!r}') |
| if (start == 'A') != (bridge_texts is not None): |
| raise ValueError('the DP-2 bridging cell runs in start A only, with its bench2 prompts (bench2_texts)') |
| results: dict[str, Json] = {} |
| for name in cells: |
| results[name] = wc.run_seeded_cell(seeded_api, wc.SEEDED_CELLS[name], texts, plan) |
| out: dict[str, Json] = {'schema': SCHEMA, 'type': 'start', 'start': start, 'cells': results} |
| if bridge_texts is not None: |
| merged = {**texts, **bridge_texts} |
| bridge: dict[str, Json] = {c.name: wc.run_seeded_cell(seeded_api, c, merged, plan) for c in BRIDGE_CELLS} |
| out['bridge'] = bridge |
| out['bridge_report'] = bridge_report(bridge) |
| out['dist'] = dg.probe(dist_api, materials, dist_rows, plan.salt_prefix) |
| return out |
|
|
|
|
| def _diff_stats(a: Mapping[str, Json], b: Mapping[str, Json], metric: str, higher: bool) -> dict[str, Json]: |
| x, y = wc.seeded_values(a, metric), wc.seeded_values(b, metric) |
| diffs = [((v - u) if higher else (u - v)) / u for u, v in zip(x, y)] |
| sd = statistics.stdev(diffs) |
| mean = statistics.fmean(diffs) |
| return {'pairs': len(diffs), 'mean_diff': mean, 'diff_cv': sd, 'sign_flip_p': d.sign_flip_p(diffs), |
| 'start_sd_estimate_1df': abs(mean) / math.sqrt(2)} |
|
|
|
|
| def compare(a: Mapping[str, Json], b: Mapping[str, Json]) -> dict[str, Json]: |
| """Start A against start B on the pilot cells (the bridge ran in A only and is reported, never compared).""" |
| ca, cb = record(a['cells']), record(b['cells']) |
| if set(ca) != set(cb): |
| raise ValueError('the two starts ran different pilot cells') |
| cells: dict[str, Json] = {} |
| single_pass = True |
| for name in sorted(ca): |
| x, y = record(ca[name]), record(cb[name]) |
| identical = wc.streams_identical(x, y) |
| rounds = [record(p)['rounds'] == record(q)['rounds'] |
| for p, q in zip(cast(list[Json], x['pairs']), cast(list[Json], y['pairs']))] |
| digests = [record(p)['output_sha256'] == record(q)['output_sha256'] |
| for p, q in zip(cast(list[Json], x['pairs']), cast(list[Json], y['pairs']))] |
| profile = str(x['profile']) |
| if profile == 'single' and not identical: |
| single_pass = False |
| cells[name] = {'profile': profile, 'identical': identical, 'pairs_same_rounds': sum(rounds), |
| 'pairs_same_output': sum(digests), 'pairs': len(rounds), |
| 'metrics': {m: _diff_stats(x, y, m, hib) for m, hib in METRICS.items()}} |
| singles = [c for c in cells.values() if record(c)['profile'] == 'single'] |
| return {'schema': SCHEMA, 'type': 'compare', |
| 'verdict': 'PASS' if single_pass and singles else 'FAIL', |
| 'reason': ('B1 seeded streams and round counts identical across the two starts' if single_pass and singles |
| else 'a B1 seeded cell differs between the two starts (or no B1 cell ran): seeded pairing at B1 ' |
| 'cannot assume identical streams; report measured diff CVs only'), |
| 'cells': cells, 'dist': dg.compare(record(a['dist']), record(b['dist']))} |
|
|
|
|
| def plan_text(cells: Sequence[str]) -> str: |
| bridge = sum(c.estimate_s for c in BRIDGE_CELLS) |
| return ('DRY RUN seeded A/A pilot (end-of-restore add-on, no extra HOLD). Start A = the last control start, start B = ' |
| 'restored production between W8 check and W8 close. Each start: seeded cells ' + ', '.join(cells) + |
| f' (6 seeds x {wc.SEEDED_MAX_TOKENS} tokens, registered seeds), then the distributional probe (' |
| + ','.join(DIST_SUBSET) + ' subset, 8 fixed positions per prompt). PASS = identical B1 streams and round counts. ' |
| 'Start A also runs the DP-2 bridging cell (report only): ' + ', '.join(c.name for c in BRIDGE_CELLS) |
| + f' (3 seeded pairs each, ESTIMATED {bridge / 60:.1f} min of cells plus the summary-32K rebuild).') |
|
|
|
|
| def main(argv: list[str] | None = None) -> int: |
| ap = argparse.ArgumentParser(description=__doc__) |
| sub = ap.add_subparsers(dest='cmd', required=True) |
| r = sub.add_parser('run', help='one start') |
| r.add_argument('--execute', action='store_true') |
| r.add_argument('--url', required=True, help='the start under test (tunnel 127.0.0.1:18001 or production :8000)') |
| r.add_argument('--cuts', type=Path, required=True, help='window_cells build output (numeric cuts)') |
| r.add_argument('--dist-cuts', type=Path, required=True, help='dist_gate cuts output') |
| r.add_argument('--salt', required=True) |
| r.add_argument('--deadline-utc', type=float, required=True) |
| r.add_argument('--out', type=Path, required=True) |
| r.add_argument('--data-dir', type=Path, default=Path('/opt/dsv41/data/bench-data')) |
| r.add_argument('--start', choices=('A', 'B'), required=True, |
| help='A = the last control start (runs the DP-2 bridge too), B = restored production') |
| c = sub.add_parser('compare') |
| c.add_argument('a', type=Path) |
| c.add_argument('b', type=Path) |
| c.add_argument('--out', type=Path, required=True) |
| args = ap.parse_args(argv) |
| if args.cmd == 'compare': |
| result = compare(record(cast(Json, json.loads(args.a.read_text()))), record(cast(Json, json.loads(args.b.read_text())))) |
| with args.out.open('x') as handle: |
| handle.write(json.dumps(result, allow_nan=False) + '\n') |
| print(json.dumps({'verdict': result['verdict'], 'reason': result['reason']})) |
| return 0 if result['verdict'] == 'PASS' else 1 |
| if not args.execute: |
| print(plan_text(PILOT_CELLS)) |
| return 0 |
| if os.environ.get('WINDOW_COORDINATOR') != COORDINATOR: |
| print(f'STOP: the A/A pilot is coordinator-only (WINDOW_COORDINATOR={COORDINATOR})', file=sys.stderr) |
| return 2 |
| if args.out.exists(): |
| print('STOP: fresh --out required', file=sys.stderr) |
| return 2 |
| from offload.prefill_trace.window_requests import corpus |
| materials, hashes = corpus(args.data_dir) |
| texts = wc.texts_from(record(cast(Json, json.loads(args.cuts.read_text()))), materials, hashes) |
| dist = record(cast(Json, json.loads(args.dist_cuts.read_text()))) |
| if dist.get('corpus_sha256') != hashes: |
| raise ValueError('public corpus changed since the distributional cuts were built') |
| rows = record(dist['prompts']) |
| keys = {p.key for p in dg.selected(list(DIST_SUBSET))} |
| plan = wc.Plan(args.deadline_utc, 4 * 1024 ** 2, [], args.salt) |
| api = BridgeHttp(args.url) |
| bridge = bench2_texts(api, args.data_dir) if args.start == 'A' else None |
| result = run(api, dg.Http(args.url), texts, plan, materials, {k: rows[k] for k in sorted(keys)}, |
| start=args.start, bridge_texts=bridge) |
| with args.out.open('x') as handle: |
| handle.write(json.dumps(result, allow_nan=False) + '\n') |
| print(f'aa_pilot run: wrote {args.out} (numbers, digests and logprob vectors only)') |
| return 0 |
|
|
|
|
| if __name__ == '__main__': |
| raise SystemExit(main()) |
|
|
