File size: 3,500 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 | """CPU checks closing replay check's missing-file and partial-equivalence gaps."""
from __future__ import annotations
import argparse
import json
from pathlib import Path
import re
from collections.abc import Mapping
def response_policy(response: Mapping[str, object]) -> None:
choices = response.get('choices')
if not isinstance(choices, list) or len(choices) != 1 or not isinstance(choices[0], dict):
raise ValueError('refused or malformed replay response: expected one choice')
choice = choices[0]
if response.get('error') or response.get('refusal') or choice.get('refusal'):
raise ValueError('refused replay response')
# None preserves old fake/legacy response fixtures; explicit error/cancel/filter finishes refuse.
if choice.get('finish_reason') not in (None, 'stop', 'length'):
raise ValueError('refused replay finish reason')
usage = response.get('usage') or {}
if not isinstance(usage, dict):
raise ValueError('malformed replay usage')
details = usage.get('prompt_tokens_details') or {}
if not isinstance(details, dict):
raise ValueError('malformed replay cache accounting')
if details.get('cached_tokens'):
raise RuntimeError('cached prompt tokens: replay rows were not recomputed')
def response_entry(meta: Mapping[str, object], entry: Mapping[str, object]) -> None:
for key in ('seq_id', 'request_id', 'L', 'gen_start', 'from', 'tokens_sha256'):
if meta[key] != entry[key]:
raise ValueError('response metadata differs from replay plan: '+key)
usage = meta.get('usage', {})
if not isinstance(usage, dict):
raise ValueError('missing response usage')
if (usage.get('prompt_tokens_details') or {}).get('cached_tokens', 0) != 0:
raise ValueError('cached response refused')
def shard_complete(text: str, records: int, k: int) -> None:
start = re.search(r'^(\d+) blocks in shard '+str(k)+r'/3, (\d+) to go$', text, re.M)
end = re.fullmatch(r'done (\d+) blocks in [0-9]+ s', text.splitlines()[-1] if text else '')
if not start or not end or start[1] != start[2] or end[1] != start[2]:
raise ValueError('equivalence shard did not finish a fresh complete population')
if records != int(start[1]):
raise ValueError('partial equivalence shard records')
def responses(root: Path) -> None:
plan = [json.loads(line) for line in (root/'plan.jsonl').read_text().splitlines()]
if not plan:
raise ValueError('empty plan')
for entry in plan:
meta = json.loads((root/'logprobs'/(entry['seq_id']+'.json')).read_text())
response_entry(meta, entry)
print(json.dumps({'responses_checked': len(plan), 'cached': 0}))
def equiv_complete(work: Path) -> None:
total = 0
for k in range(3):
text = (work/f'drafts-{k}.log').read_text()
records = (work/f'drafts-{k}of3.jsonl').read_text().splitlines()
shard_complete(text, len(records), k)
total += len(records)
if total != 542:
raise ValueError(f'expected all 542 A-capture blocks, found {total}')
print(json.dumps({'complete_shards': 3, 'blocks': total, 'every': 1}))
def main() -> None:
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument('check', choices=('responses', 'equiv-complete'))
ap.add_argument('root', type=Path)
a = ap.parse_args()
(responses if a.check == 'responses' else equiv_complete)(a.root)
if __name__ == '__main__':
main()
|