"""Deliverable 2: what changed under v6.0, when (UTC), and its effect in the data -> out/v6_changes.md (+ csv files).
Inputs: out/version_markers.csv (scripts/version_markers.py: each Worker version's own bundle), out/deployments.csv,
raw/runs_*.json, raw/export/*.jsonl (scripts/export.py), raw/settings_rules.json, seed ids from the repository's bundled data."""
import sys, hashlib; sys.path.insert(0, '${RPB_AUDIT_DIR}/scripts')
from common import *

runs = load_runs()
vers = list(csv.DictReader(open(f'{OUT}/version_markers.csv')))
deps = list(csv.DictReader(open(f'{OUT}/deployments.csv')))
dep_by_ver = {d['version_id']: d['deployment_created_utc'] for d in deps}
sess = defaultdict(list)
for s in load_jsonl('sessions'):
    s['result'] = json.loads(s['result_json']) if s['result_json'] else None; sess[s['run_id']].append(s)
sid2run = {s['id']: s['run_id'] for v in sess.values() for s in v}
judg = defaultdict(list)
for j in load_jsonl('judgments'): judg[sid2run[j['session_id']]].append(j)
logs = defaultdict(list)
for l in load_jsonl('logs'): logs[l['run_id']].append(l)
batches = load_jsonl('batches')
simm = defaultdict(list)
for m in load_jsonl('simmodels'): simm[m['session_id']].append(m)
md = ['# v6.0 changes: timeline and effect in the data', '', 'All times UTC. Versions: Cloudflare Workers API version list of `rp-bench-api` (62 versions, raw/cf_versions_p*.json); '
      'markers: each version\'s own worker.js fetched with `GET .../workers/workers/rp-bench-api/versions/{id}?include=modules` (scripts/version_markers.py -> out/version_markers.csv). '
      'Deployment records (`wrangler deployments list`, raw/deployments_list.txt; API raw/cf_deployments.json) exist only for the 10 most recent deployments (versions 53–62); for earlier versions the version upload time is used (every version shows source `wrangler`, annotation `version_upload`; the deployments themselves are no longer listed).', '']

def first(marker, pred=lambda v: v not in ('0', '', '32e3')):
    v = next((x for x in vers if pred(x[marker])), None)
    prev = vers[int(v['number']) - 2] if v and int(v['number']) > 1 else None
    return v, prev

CHANGES = [
    ('v6.0 protocol (realistic simulator, sampled lengths, fixed personas, target max_tokens 8192, finish reasons)', 'protocol_v6.0', 'protocol comment v6.0', 'No (generation-time; new runs)'),
    ('Single-host leaderboard eligibility rule', 'single_host_rule', 'report: "later additions without score changes"', 'Eligibility only (recomputed with results)'),
    ('Experience-quality sub-score', 'quality_score', 'report: "later additions"', 'No (not in overall)'),
    ('Simulator turn blocked by moderation written by the other simulator model', 'sim_moderation_switch', 'change log 26 Oct', 'No (generation-time; affects sessions generated/resumed after it)'),
    ('Wrong-language session rule (score 0) + hourly re-application to all stored results', 'wrong_language_rule', 'change log 26 Oct; protocol: sessionRuleVersion', 'YES: re-scores stored sessions of minimax/platform/calibration runs and recomputes run results'),
    ('Simulator model fixed by scenario and sample (assignment "slot")', 'slot_assignment', 'change log 26 Oct', 'No (snapshot-level; runs created after it)'),
    ('Judge output limits 32000 -> 16000 (Opus) / 8000 (Sol)', 'opus_judge_maxTokens', 'change log 26 Oct', 'No (snapshot-level; runs created after it)'),
    ('Re-judge a truncated reply with twice the limit', 'judge_limit_doubling', 'change log 26 Oct', 'No (judging-time)'),
    ('Batches submitted only as far as the balance covers', 'batch_balance_hold', 'change log 26 Oct', 'No (scheduling only)'),
    ('Expired batch: segments judged directly (sync API)', 'expired_batch_direct', 'change log 27 Oct', 'No for completed runs; changes judging mode of runs in flight'),
    ('Moderation-refused segment scored by the other official judge (substituteBlocked)', 'substituteBlocked', 'change log 27 Oct', 'No for completed sessions (result computed at session completion); lets previously stuck sessions complete'),
    ('Direct judging: a second unusable output is final', 'direct_final_failure', 'not in change log', 'No (judging-time)'),
]
tl = []
for name, marker, src, can in CHANGES:
    v, prev = first(marker, (lambda x: x == '16e3') if marker == 'opus_judge_maxTokens' else (lambda x: x not in ('0', '')))
    tl.append({'change': name, 'source_of_claim': src, 'first_version': v['number'], 'version_id': v['version_id'], 'version_created_utc': v['created_utc'][:19].replace('T', ' '),
               'deployment_record_utc': dep_by_ver.get(v['version_id'], 'not retained (only last 10 deployments listed)'),
               'previous_version_utc': prev['created_utc'][:19].replace('T', ' ') if prev else '', 'can_change_completed_scores': can})
with open(f'{OUT}/v6_timeline.csv', 'w', newline='') as f:
    w = csv.DictWriter(f, fieldnames=list(tl[0])); w.writeheader(); w.writerows(tl)
md += ['## 1. Timeline: first Worker version containing each change', '', md_table(['change', 'claimed in', 'first version', 'version created (UTC)', 'previous version (UTC)', 'deployment record', 'can it change a completed run\'s scores?'],
       [[t['change'], t['source_of_claim'], t['first_version'], t['version_created_utc'], t['previous_version_utc'], t['deployment_record_utc'], t['can_change_completed_scores']] for t in tl]), '']
md += ['Protocol version strings carry local (UTC+8) dates: "v6.0 · 2026-10-25" first appears in version 4 created 2026-10-24 19:59 UTC (= 25 Oct 03:59 UTC+8); the "26 Oct" changes are in versions created 26 Oct 11:41–13:36 UTC; the "27 Oct" changes in versions created 27 Oct 11:30–11:48 UTC. '
       'No version string changed for any of them: every bundle from version 4 to 62 carries "v6.0 · 2026-10-25" (out/version_markers.csv, column protocol_versions_in_bundle).', '']
# settings: rule pass
st_rows = json.load(open(f'{RAW}/settings_rules.json'))[0]['results']
rule = next(x for x in st_rows if x['key'] == 'session_rule:minimax-rp')
md += [f"Wrong-language rule pass: settings `session_rule:minimax-rp` = {rule['value']} written {utc(rule['updated_at'])} UTC (raw/settings_rules.json), i.e. the hourly cron finished re-scoring all stored results "
       f"{utc(rule['updated_at'])} — after version 26 ({tl[4]['version_created_utc']}).", '']

# 2. runs before/after
plat = [r for r in runs.values() if r['origin'] != 'minimax']
cut = {t['change']: t['version_created_utc'] for t in tl}
md += ['## 2. Runs created before / after each 26–27 Oct change (platform + calibration runs; imported runs are not created by the engine)', '']
rows = []
for r in sorted(plat, key=lambda r: r['created_at']):
    cr = utc(r['created_at']); fin = utc(r['finished_at'])
    if r['created_at'] < 1792886400000 - 86400000 * 0: pass
    rows.append([r['id'][:10], r['display_name'][:30], r['config'].get('mode'), r['pv'], cr, fin or r['status'],
                 'after' if cr >= cut[CHANGES[5][0]] else 'before', 'after' if cr >= cut[CHANGES[6][0]] else 'before',
                 'after' if (fin or '9') >= cut[CHANGES[9][0]] and cr < cut[CHANGES[9][0]] else ('after' if cr >= cut[CHANGES[9][0]] else 'before (finished)'),
                 'spans' if cr < cut[CHANGES[10][0]] <= (fin or '9') else ('after' if cr >= cut[CHANGES[10][0]] else 'before (finished)')])
md += ['Only runs from 25 Oct on are shown in full in out/runs.md; every run created before 2026-10-26 13:19:51 UTC is "before" the slot/rule/limit changes. Runs that matter:', '']
md.append(md_table(['run', 'name', 'mode', 'protocol', 'created', 'finished/status', 'slot assignment (v26)', 'judge limits 16k/8k (v27)', 'expired→direct (v51)', 'substitution (v52)'],
                   [x for x in rows if x[4] >= '2026-10-25']))
md += ['', '1547acce31 (Doubao-Seed-Character) was created 2026-10-26 09:20 UTC, before every 26/27 Oct change, and finished 2026-10-27 13:44 UTC, after all of them: its generation used hash assignment and 32000-token judge limits (snapshot), but its judging was completed under the 27 Oct code (direct judging, substitution). cd542f9e50 and 44e958def4 were created after all changes.', '']

# 3. wrong-language rule effect
md += ['## 3. Wrong-language rule: penalised sessions and score change per run', '']
pen_rows = []
for rid, ss in sess.items():
    scored = [s for s in ss if s['status'] == 'done' and s['result'] and not s['result'].get('excluded')]
    pens = [s for s in scored if s['result'].get('penalty')]
    if not pens: continue
    out = {'run_id': rid, 'name': runs[rid]['display_name'], 'penalised': len(pens), 'scored_sessions': len(scored),
           'sessions': '; '.join(f"{s['seed_id']}#{s['sample_idx'] + 1} ({s['result']['penalty']['detail']}, judges' overall {ov(s['result']['original']):.1f})" for s in pens)}
    for view in ['all', 'zh', 'en']:
        v = [s for s in scored if view == 'all' or s['lang'] == view]
        if not v: continue
        with_r = agg([s['result']['combined'] for s in v])[1]; without = agg([s['result'].get('original') or s['result']['combined'] for s in v])[1]
        out[f'{view}_with_rule'] = round(with_r, 2); out[f'{view}_without_rule'] = round(without, 2); out[f'{view}_change'] = round(with_r - without, 2)
    pen_rows.append(out)
with open(f'{OUT}/v6_penalties.csv', 'w', newline='') as f:
    keys = sorted({k for r in pen_rows for k in r}, key=lambda k: ['run_id', 'name', 'penalised', 'scored_sessions'].index(k) if k in ['run_id', 'name', 'penalised', 'scored_sessions'] else 10)
    w = csv.DictWriter(f, fieldnames=keys); w.writeheader(); w.writerows(pen_rows)
md.append(md_table(['run', 'name', 'penalised / scored', 'overall all: without → with rule (Δ)', 'zh', 'en', 'sessions'],
                   [[r['run_id'][:18], r['name'], f"{r['penalised']}/{r['scored_sessions']}", f"{r['all_without_rule']} → {r['all_with_rule']} ({r['all_change']:+.2f})",
                     f"{r.get('zh_without_rule')} → {r.get('zh_with_rule')} ({r.get('zh_change', 0):+.2f})", f"{r.get('en_without_rule')} → {r.get('en_with_rule')} ({r.get('en_change', 0):+.2f})", r['sessions']] for r in pen_rows]))
tot = sum(r['penalised'] for r in pen_rows)
md += ['', f'{tot} sessions in {len(pen_rows)} runs carry a penalty; all other stored sessions (official, screening, calibration, imported) have none (out/v6_penalties.csv).', '']

# 4. simulator assignment
seeds = {l: sorted(json.loads(x)['id'] for x in open(f'${PROJECT_ROOT}/worker/src/benchmarks/minimax-rp/data/seeds.{l}.jsonl')) for l in ('zh', 'en')}
md += ['## 4. Simulator assignment (hash vs slot)', '']
arows = []
for rid, r in sorted(runs.items(), key=lambda kv: kv[1]['created_at']):
    if r['origin'] == 'minimax' or r['config'].get('mode') == 'calibration': continue
    us = r['usersim']; models = [m['model'] for m in us.get('models', [])]
    if len(models) < 2: continue
    assign = us.get('assignment') or 'hash'; spp = r['config'].get('samplesPerSeed') or 1
    cnt = Counter(); match = tot_ = mixed = 0
    for s in sess[rid]:
        ms = Counter()
        for m in simm.get(s['id'], []):
            if m['source'] == 'user_sim': ms[m['model']] += m['n']
        if not ms: continue
        main = ms.most_common(1)[0][0]; cnt[(s['lang'], main.split('/')[-1])] += 1
        if len(ms) > 1: mixed += 1
        if assign == 'slot':
            pos = seeds[s['lang']].index(s['seed_id']); exp = models[(pos * spp + s['sample_idx']) % len(models)]
        else:
            exp = models[hashlib.sha1(f"usersim:{s['id']}".encode()).digest()[0] % len(models)]
        tot_ += 1; match += exp == main
    arows.append([rid[:10], r['display_name'][:28], r['config'].get('mode'), utc(r['created_at'])[5:16], assign, '; '.join(f'{l} {m}: {n}' for (l, m), n in sorted(cnt.items())), f'{match}/{tot_}', mixed])
md.append(md_table(['run', 'name', 'mode', 'created', 'assignment (snapshot)', 'sessions per language × simulator (majority model of user turns)', 'sessions matching the assignment rule', 'sessions with >1 simulator model'], arows))
md += ['', 'Assignment rule check: hash = sha1("usersim:"+session id)[0] mod 2 (engine/runner.ts simModelFor), slot = (position of seed among its language\'s seeds in id order × samples + sample) mod 2.', '']

# 5. substitution
md += ['## 5. Judge substitution (substituteBlocked)', '']
srows = []
for rid, ss in sess.items():
    for s in ss:
        sub = (s['result'] or {}).get('substituted')
        if not sub: continue
        r = s['result']; a = r['judges'].get('judge_a', {}).get('dims', {}); b = r['judges'].get('judge_b', {}).get('dims', {})
        srows.append([rid[:10], f"{s['seed_id']}#{s['sample_idx'] + 1}", len(sub), ', '.join(f"{x['judge']} seg{x['chunk'] + 1}←{x['by']}" for x in sub),
                      f"{ov(r['combined']):.2f}", f"{ov(b):.2f}", f"{ov(a):.2f}", 'penalised' if r.get('penalty') else ''])
md.append(md_table(['run', 'session', 'segments substituted', 'detail', 'session overall (stored)', 'judge_b alone', 'judge_a incl. substituted segments', 'note'], srows))
import re
blocked = [j for rid in judg for j in judg[rid] if j['status'] == 'failed' and j['error'] and re.search(r'inappropriate content|data_inspection_failed|content[_ ]filter|content management policy|PROHIBITED_CONTENT|blocked the request', j['error'], re.I)]
bl = Counter((sid2run[j['session_id']][:10], j['judge_key']) for j in blocked)
md += ['', 'Judgements recorded as failed with a moderation error: ' + (', '.join(f'{k[0]} {k[1]}: {v}' for k, v in bl.items()) or 'none') + '.', '']
r1547 = sess.get('1547acce317f4fc3', [])
if srows:
    sc = [s for s in r1547 if s['status'] == 'done' and s['result'] and not s['result'].get('excluded')]
    allv = agg([s['result']['combined'] for s in sc])[1]; wo = agg([s['result']['combined'] for s in sc if not s['result'].get('substituted')])[1]
    md += [f'Effect on 1547acce31: overall {allv:.2f} with the {len(srows)} substituted sessions, {wo:.2f} without them ({allv - wo:+.2f}). Without substitution these sessions would have stayed incomplete and the run could not have completed (engine: a session completes only when every segment is settled).', '']

# 6. direct judging / expired batches
md += ['## 6. Expired batches and direct judging', '']
brow = [[b['run_id'][:10], b['id'][:28], b['judge_key'], b['status'], b['items'], (b['provider_status'] or ''), (b['error'] or '')[:40], f"{b['cost']:.2f}", f"{b['reserved']:.2f}" if b['reserved'] else '', utc(b['created_at']), utc(b['updated_at'])] for b in batches]
md.append(md_table(['run', 'batch', 'judge', 'status (DB)', 'items', 'provider status', 'error', 'cost $', 'reserved $', 'created', 'updated'], brow))
r = runs['1547acce317f4fc3']; at = r['config'].get('directJudging', {}).get('at')
J = judg['1547acce317f4fc3']
after = Counter(j['judge_key'] for j in J if j['status'] == 'done' and j['created_at'] >= at)
before = Counter(j['judge_key'] for j in J if j['status'] == 'done' and j['created_at'] < at)
fails = Counter((j['judge_key'], (j['error'] or '')[:60]) for j in J if j['status'] != 'done')
exp_items = sum(b['items'] for b in batches if (b['provider_status'] or '').startswith('expired'))
open_b = [b for b in batches if b['status'] == 'submitted']
dlogs = [l for l in logs['1547acce317f4fc3'] if l['message'].startswith('直接调用裁判评分 ')]
import re as _re
n_direct = sum(int(_re.search(r'评分 (\d+) 个', l['message']).group(1)) for l in dlogs)
n_auto = sum('所在批处理已过期' in l['message'] for l in dlogs)
md += ['', f"1547acce31: {len([b for b in batches if (b['provider_status'] or '').startswith('expired')])} judge_b (GPT-6.1 Sol) batches expired after ~24 h ({exp_items} items, provider status 'expired 197–199/200', no results returned; first expiry logged {utc(min(l['ts'] for l in logs['1547acce317f4fc3'] if 'expired' in l['message']))}). "
       f"Direct judging was then switched on for the whole run by an administrator at {utc(at)} UTC (config.directJudging.at; run log '管理员将剩余未出分的分段改为直接调用裁判评分'), 10 s after version 51 (automatic expired-batch path) was created. "
       f"Logs record {len(dlogs)} direct-judging rounds covering {n_direct} segment requests; {n_auto} rounds are labelled '所在批处理已过期' (the automatic expired-batch path), the rest '本评测已改为直接评分' (the admin switch).",
       f"Scored judgements stored before the switch: {dict(before)}; after: {dict(after)} (judgments.created_at vs directJudging.at). Non-done judgements: {dict(fails)}.",
       f"{len(open_b)} batches are still 'submitted' in judge_batches for runs that are completed or paused ({', '.join(b['run_id'][:10] + ' ' + b['judge_key'] + ' ' + str(b['items']) + ' items' for b in open_b)}); reserved total ${sum(b['reserved'] or 0 for b in open_b):.2f} (where recorded). Their late results would be ignored for segments already judged (pollBatches skips them).",
       '44e958def4 and cd542f9e50 use config.fastJudging (official judges called synchronously, batch=false): no batches at all. So of the two completed official runs, one was judged partly by batch and partly directly (1547acce31), the other entirely synchronously (44e958def4).', '']

# 7. judge limits and truncation
md += ['## 7. Judge output limits and truncation', '']
lrows = []
for rid in ['1547acce317f4fc3', '44e958def4a748ad', 'cd542f9e50224bfe']:
    rr = runs[rid]; J = judg[rid]
    lim = {j['key']: j.get('maxTokens') for j in rr['judges']}
    for k in ['judge_a', 'judge_b']:
        ct = sorted(j['completion_tokens'] for j in J if j['judge_key'] == k and j['status'] == 'done')
        tr = sum(1 for j in J if j['judge_key'] == k and j['error'] and '截断' in j['error'])
        lrows.append([rid[:10], k, lim.get(k), len(ct), ct[len(ct) // 2] if ct else '', ct[-1] if ct else '', sum(c > 8000 for c in ct), sum(c > 16000 for c in ct), tr])
md.append(md_table(['run', 'judge', 'maxTokens (snapshot)', 'done judgements', 'median completion tokens', 'max', '> 8000', '> 16000', 'stored with truncation error'], lrows))
md += ['', 'completion_tokens of a judgement include reasoning tokens as reported by the provider.', '']

# 8. simulator moderation switch
sw = [(sid2run[sid][:10], sid, ms) for sid, ms in simm.items() if len({m['model'] for m in ms if m['source'] == 'user_sim'}) > 1]
swl = [l for rid in logs for l in logs[rid] if '本轮改由另一个用户模拟器' in l['message']]
md += ['## 8. Simulator turn blocked by moderation', '', f'Sessions whose user turns come from more than one simulator model: {len(sw)} ' + '; '.join(f"{r} {sid}: " + ', '.join(f"{m['model'].split('/')[-1]}×{m['n']}" for m in ms) for r, sid, ms in sw) + '.',
       f'Log lines "本轮改由另一个用户模拟器": {len(swl)} (' + '; '.join(f"{l['run_id'][:10]} {utc(l['ts'])}" for l in swl) + ').', '']
open(f'{OUT}/v6_changes.md', 'w').write('\n'.join(md))
print('\n'.join(md))