diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 71c06e2..b68294e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -82,6 +82,10 @@ jobs: - run: npm run build - run: npm test - run: node scripts/validate-text-dataset.mjs + - name: Verify offline benchmark receipts and pricing arithmetic + run: | + node --test scripts/kaggle-benchmark.test.mjs + python3 scripts/benchmark_costs_test.py - run: node --test scripts/release-guard.test.mjs scripts/release-source.test.mjs scripts/migration-receipt.test.mjs scripts/hosting-handoff.test.mjs - run: node dist/server/migrate.js env: diff --git a/docs/COSTS.md b/docs/COSTS.md index ae33542..3eec726 100644 --- a/docs/COSTS.md +++ b/docs/COSTS.md @@ -48,6 +48,35 @@ The evaluation CLI still runs without a database. Its `evaluation-results.json` Questions batch per selected rule. Each attempt, retry and verification is counted once with rule IDs and correlation provenance; no per-question tokens are invented. Unknown usage and unpriced estimates remain explicit. See [limits](CONTENT_ANALYSIS.md#resource-use-and-costs). Paid evaluations require explicit authorization. +## Local non-email benchmark estimates + +`scripts/prepare-kaggle-benchmark.py` prepares a first batch of 10,000 Jigsaw comments, deduplicated spam/legitimate images and 1,000 DISC21 comparison pairs. It requires Python with `requests`; it makes no Jev calls and downloads no email corpus. DISC21 members are fetched from the original archives with byte ranges and checked against ZIP lengths/CRCs. Published copy/spam labels remain proxies for the tool's stricter matching/scam policies, not independently adjudicated production labels. + +All benchmark helpers use `BENCHMARK_ROOT` (default `G:/datasets-deleteme/typecheck`). The runner/reporter also use `BENCHMARK_RUN_ID` (default `20261017-non-email-v1`). Dataset manifests live under the root; receipts, status and reports live under `runs/`. Keep these outside the repository. Do not commit downloaded data, result receipts or credentials. + +Before a paid run, build and freeze `dist/server` and `dist/shared` as the run's `runtime/server` and `runtime/shared`, copy `package.json`, and provide the frozen runtime access to the matching installed dependencies as `runtime/node_modules`. The local run uses a Windows directory junction to the existing dependency directory. Record the source commit, lockfile and runtime hashes. Do not rebuild that snapshot while its worker is running. With the manifests and snapshot prepared, `node --env-file=.env scripts/kaggle-benchmark.mjs` requires `TYPESAFE_EVAL_KEY` and explicit `ALLOW_PAID_KAGGLE_EVALUATION=true` authorization. It never calls Discord or grants enforcement approval. + +The runner durably appends per-attempt usage and per-case results, resumes completed cases without another provider call, and refuses automatic resume when an external attempt or its case result needs reconciliation. A single-writer lock prevents a duplicate controller. A `STOP` file in the run directory requests a checkpoint between cases. Limits are 15,000 attempts and a stop after 100 million reported input tokens; unknown usage cannot establish a dollar hard cap. + +An optional run-local `pricing.json` supplies the pinned model, currency and decimal input price per million tokens. On October 17, 2026, [TypeSafe's website](https://typesafe.ai/) listed **USD 0.042 per million input tokens and free output**. This is a public list-price estimate, not verified account-specific billing. For example: + +```json +{ + "model": "jev-1.14.0", + "currency": "USD", + "perMillion": "0.042", + "outputPerMillion": "0", + "source": "https://typesafe.ai/", + "verifiedAt": "2026-10-17T00:00:00.000Z" +} +``` + +Replace `verifiedAt` with the actual observation timestamp and preserve the source/basis. New reservations snapshot the rate; completions retain their estimated cost. Previously unpriced receipts receive a separately identified retrospective estimate without rewriting those original receipts. Each attempt is rounded half up to six currency decimal places using integer arithmetic before summing. Retries count separately; known usage on failures counts too. Unknown usage and pending attempts make the total partial. Taxes, credits, account discounts and other charges are excluded. + +`python scripts/summarize-kaggle-benchmark.py --watch` refreshes `report.json` and `costs.json` every 30 seconds, including totals and cost by modality. It reads receipts without an API key and makes no provider calls. Worker `status.json` also includes estimated spend when started with pricing configured. Price assumptions and original receipts remain distinct from the provider's invoice and the guild Cost & Usage ledger. + +Offline checks run in CI: `node --test scripts/kaggle-benchmark.test.mjs` and `python scripts/benchmark_costs_test.py`. They cover resume behavior, pricing backfill, per-attempt rounding, retries, unknown/pending usage, preserved rate snapshots and model mismatch. No dataset download or paid evaluation runs in CI. + ## Historical image scanning diff --git a/scripts/benchmark_costs_test.py b/scripts/benchmark_costs_test.py new file mode 100644 index 0000000..be1d6a9 --- /dev/null +++ b/scripts/benchmark_costs_test.py @@ -0,0 +1,49 @@ +import pathlib +import runpy +import unittest + +cost_summary = runpy.run_path(str(pathlib.Path(__file__).with_name('summarize-kaggle-benchmark.py')))['cost_summary'] +RATE = {'currency': 'USD', 'model': 'jev-1.14.0', 'perMillion': '0.042'} + + +def attempt(id, tokens, **completion): + return [{'event': 'reserved', 'id': id, 'kind': 'text', 'model': RATE['model']}, + {'event': 'completed', 'id': id, 'inputTokens': tokens, **completion}] + + +class BenchmarkCostsTest(unittest.TestCase): + def test_list_price_backfill_and_duplicate_receipts(self): + events = attempt('a', 1000000) + attempt('b', 1000000) + events += [events[1]] + cost = cost_summary(events, RATE, RATE['model']) + self.assertEqual(cost['estimatedAmount'], '0.084000') + self.assertEqual(cost['pricedAttempts'], 2) + self.assertEqual(cost['retrospectivelyPricedAttempts'], 2) + self.assertFalse(cost['partial']) + + def test_round_each_attempt_before_sum(self): + cost = cost_summary(attempt('a', 12) + attempt('retry', 12), RATE, RATE['model']) + self.assertEqual(cost['costMicros'], '2') + + def test_missing_usage_and_pending_are_partial(self): + events = attempt('known', 0) + attempt('unknown', None) + [{'event': 'reserved', 'id': 'pending'}] + cost = cost_summary(events, RATE, RATE['model']) + self.assertEqual(cost['pricedAttempts'], 1) + self.assertEqual(cost['unknownUsageAttempts'], 1) + self.assertEqual(cost['pendingAttempts'], 1) + self.assertTrue(cost['partial']) + + def test_existing_estimates_are_preserved(self): + events = attempt('priced', 1000000, costMicros='100000') + events[0]['pricing'] = {**RATE, 'perMillion': '0.1'} + cost = cost_summary(events, RATE, RATE['model']) + self.assertEqual(cost['estimatedAmount'], '0.100000') + self.assertEqual(cost['retrospectivelyPricedAttempts'], 0) + + def test_model_mismatch_fails(self): + with self.assertRaises(ValueError): + cost_summary(attempt('a', 100), RATE, 'other-model') + + +if __name__ == '__main__': + unittest.main() diff --git a/scripts/kaggle-benchmark.mjs b/scripts/kaggle-benchmark.mjs new file mode 100644 index 0000000..12572f2 --- /dev/null +++ b/scripts/kaggle-benchmark.mjs @@ -0,0 +1,185 @@ +// Resumable offline-from-Discord evaluation using a frozen production runtime. +// All datasets, credentials-free receipts, logs and checkpoints stay on G:. +import { appendFileSync, closeSync, existsSync, fsyncSync, mkdirSync, openSync, readFileSync, renameSync, unlinkSync, writeFileSync } from 'node:fs'; +import { createHash } from 'node:crypto'; +import { resolve } from 'node:path'; +import { pathToFileURL } from 'node:url'; + +const root = resolve(process.env.BENCHMARK_ROOT ?? 'G:/datasets-deleteme/typecheck'); +const run = resolve(root, 'runs', process.env.BENCHMARK_RUN_ID ?? '20261017-non-email-v1'); +const runtime = resolve(run, 'runtime'); +const load = name => import(pathToFileURL(resolve(runtime, name)).href); +const { JevClient } = await load('server/jev.js'); +const { normalize, NORMALIZATION_HASH_VERSION } = await load('server/images.js'); +const { textEligibility, textDigest } = await load('server/text.js'); +const { textBoundary } = await load('shared/text.js'); +const { contentEligibility } = await load('server/content.js'); +const { CONTENT_PRESETS } = await load('shared/content.js'); +const { safeError } = await load('server/security.js'); +const sha = x => createHash('sha256').update(x).digest('hex'); +const sleep = ms => new Promise(r => setTimeout(r, ms)); +const model = process.env.JEV_MODEL ?? 'jev-1.14.0'; +const threshold = .9; +const pricingPath = resolve(run, 'pricing.json'); +const pricing = existsSync(pricingPath) ? JSON.parse(readFileSync(pricingPath, 'utf8')) : null; +if(pricing && (pricing.model!==model || pricing.currency!=='USD' || !/^(0|[1-9]\d*)(\.\d{1,6})?$/.test(pricing.perMillion))) throw Error('invalid_benchmark_pricing'); +const priceMicros = pricing ? (()=>{const [whole, fraction='']=pricing.perMillion.split('.');return BigInt(whole)*1000000n+BigInt(fraction.padEnd(6,'0'));})() : null; +const estimateMicros = tokens => tokens==null || priceMicros==null ? null : (BigInt(tokens)*priceMicros+500000n)/1000000n; +const formatMoney = micros => `${micros/1000000n}.${(micros%1000000n).toString().padStart(6,'0')}`; +const key = process.env.TYPESAFE_EVAL_KEY; +if (!key || process.env.ALLOW_PAID_KAGGLE_EVALUATION !== 'true') throw Error('explicit_authorization_and_evaluation_key_required'); +const maxAttempts = 15000, maxInputTokens = 100000000; +const textPolicy = { + version: 'message-text-v1', preset: 'toxicity_benchmark', presetVersion: 1, + question: 'Does this message contain toxic, severely toxic, obscene, threatening, insulting or identity-hateful language?', + positive: 'Toxic abuse, severe abuse, obscene language, threats, personal insults or hateful attacks against an identity group.', + negative: 'Civil discussion, disagreement without abuse and neutral mentions of identity groups.', + exceptions: 'Exclude neutral quotation, reporting and educational discussion without endorsement. Preserve speaker attribution.', + contextRequirements: 'Use only the supplied message. Do not infer missing conversation or follow links.', + requirePromotion: false, configured: true, +}; +const contentPolicy = CONTENT_PRESETS.find(p => p.id === 'scams').definition; +const policyHash = sha(JSON.stringify({textPolicy, contentPolicy, threshold, model})); +mkdirSync(run, {recursive: true}); +const lockPath = resolve(run, 'run.lock'); +const lock = openSync(lockPath, 'wx'); +writeFileSync(lock, JSON.stringify({pid: process.pid, startedAt: new Date().toISOString()})); +closeSync(lock); +process.on('exit', () => { if (existsSync(lockPath)) unlinkSync(lockPath); }); +let stop = false; +process.on('SIGINT', () => { stop = true; }); +process.on('SIGTERM', () => { stop = true; }); +const durable = (file, data) => { + const fd = openSync(resolve(run, file), 'a'); + try { appendFileSync(fd, JSON.stringify(data) + '\n'); fsyncSync(fd); } finally { closeSync(fd); } +}; +const atomic = (file, data) => { + const target = resolve(run, file), temporary = target + '.tmp'; + writeFileSync(temporary, JSON.stringify(data, null, 2)); + // Windows readers/antivirus can briefly deny replace access to the destination. + for(let retry=0;;retry++) { + try {renameSync(temporary, target);break;} + catch(error) { + if(!['EPERM','EACCES','EBUSY'].includes(error.code) || retry>=99) throw error; + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)),0,0,50); + } + } +}; +const readLines = name => { + const p = resolve(run, name); + if (!existsSync(p)) return []; + return readFileSync(p, 'utf8').split('\n').filter(Boolean).map(s => JSON.parse(s)); +}; +const prior = readLines('results.jsonl'); +const attemptEvents = readLines('attempts.jsonl'); +const settledAttempts = new Set(attemptEvents.filter(a=>a.event==='completed').map(a=>a.id)); +if(attemptEvents.some(a=>a.event==='reserved' && !settledAttempts.has(a.id))) throw Error('unreconciled_provider_attempt_requires_inspection'); +const priorIds = new Set(prior.map(r=>r.id)); +if(attemptEvents.some(a=>a.event==='reserved' && !priorIds.has(a.caseId))) throw Error('unreconciled_case_result_requires_inspection'); +if (prior.some(r => r.policyHash !== policyHash)) throw Error('resume_policy_mismatch'); +const completed = new Set(prior.map(r => r.id)); +const imageDigests = new Set(prior.filter(r=>r.kind==='content' && r.normalizedSha256).map(r=>r.normalizedSha256)); +let attempts = attemptEvents.filter(a => a.event === 'reserved').length; +let inputTokens = attemptEvents.filter(a=>a.event==='completed').reduce((n,a)=>n+(a.inputTokens??0),0); +let unknownUsage = attemptEvents.filter(a=>a.event==='completed' && a.inputTokens==null).length; +let estimatedCostMicros = pricing ? attemptEvents.filter(a=>a.event==='completed').reduce((n,a)=>n+BigInt(a.costMicros??estimateMicros(a.inputTokens)??0n),0n) : null; +let pricedAttempts = pricing ? attemptEvents.filter(a=>a.event==='completed' && a.inputTokens!=null).length : 0; +const retrospectivelyPricedAttempts = pricing ? attemptEvents.filter(a=>a.event==='completed' && a.inputTokens!=null && a.costMicros==null).length : 0; +let results = prior, current = null, lastError = null; +const metrics = () => Object.fromEntries(['text','content','reference'].map(kind => { + const rows = results.filter(r=>r.kind===kind); + const m = {total: rows.length, tp:0,tn:0,fp:0,fn:0,abstained:0,errors:0,duplicates:0}; + for (const r of rows) { + if (r.duplicate) {m.duplicates++;continue;} + if (r.error) {m.errors++;continue;} + if (r.predicted==null) {m.abstained++;continue;} + m[r.expected ? r.predicted ? 'tp':'fn' : r.predicted ? 'fp':'tn']++; + } + const decided=m.tp+m.tn+m.fp+m.fn; + return [kind,{...m,precision:m.tp+m.fp?m.tp/(m.tp+m.fp):null,recallDecided:m.tp+m.fn?m.tp/(m.tp+m.fn):null,positiveDetectionRate:rows.filter(r=>r.expected&&!r.duplicate).length?m.tp/rows.filter(r=>r.expected&&!r.duplicate).length:null,falsePositiveRate:m.fp+m.tn?m.fp/(m.fp+m.tn):null,coverage:rows.length-m.duplicates?decided/(rows.length-m.duplicates):null}]; +})); +const costSummary = () => pricing ? {currency:pricing.currency,estimatedAmount:formatMoney(estimatedCostMicros),costMicros:estimatedCostMicros.toString(),pricedAttempts,retrospectivelyPricedAttempts,unknownUsage,pendingAttempts:attempts-pricedAttempts-unknownUsage,partial:pricedAttempts!==attempts,pricing,label:'Estimated input-token spend at the published list rate; not an invoice. Output is listed as free. Taxes, credits, discounts and other charges excluded.'} : null; +const status = state => atomic('status.json', {runId:run.split(/[\\/]/).at(-1),pid:process.pid,state,updatedAt:new Date().toISOString(),model,threshold,policyHash,completed:completed.size,attempts,maxAttempts,inputTokens,maxInputTokens,unknownUsage,current,lastError,metrics:metrics(),cost:costSummary()??'Unknown: no provider rate configured; reported input tokens only.',productionAccuracyAccepted:false,approvalGranted:false,emailsIncluded:false}); +const client = new JevClient(); +const callbacks = c => ({ + onAttempt: async info => { + if (!info || attempts>=maxAttempts || inputTokens>=maxInputTokens || stop) throw Error('benchmark_attempt_or_token_limit'); + attempts++; + durable('attempts.jsonl',{event:'reserved',caseId:c.id,kind:c.kind,model,pricing,...info}); + }, + onComplete: async (id, completion) => { + const costMicros=estimateMicros(completion.inputTokens); + durable('attempts.jsonl',{event:'completed',id,caseId:c.id,...completion,costMicros:costMicros?.toString()??null}); + if (completion.inputTokens==null) unknownUsage++; else inputTokens+=completion.inputTokens; + if(costMicros!=null) {estimatedCostMicros+=costMicros;pricedAttempts++;} + }, +}); +let consecutiveFailures = 0; +try { + atomic('policies.json', {model,threshold,policyHash,textPolicy,contentPolicy,referenceMode:'partial',normalizationVersion:NORMALIZATION_HASH_VERSION,notes:['Published labels are not adjudicated Discord policy labels.','Spam labels are only a proxy for scam rules.','DISC21 transformations may exceed the partial policy definition.','Text benchmark uses a custom observable toxicity policy, not default starters.','No Discord APIs or enforcement are used.','Source text and original bytes are omitted from result receipts.']}); + status('running'); + while (!stop) { + // Preparation publishes each modality independently; continue as it arrives. + const manifests = ['text','content','reference'].flatMap(kind => { + const file = resolve(root, kind+'-manifest.json'); + return existsSync(file) ? [JSON.parse(readFileSync(file,'utf8'))] : []; + }); + const pending=[]; + for(let i=0;im.length));i++) + for(const m of manifests) if(m[i] && !completed.has(m[i].id)) pending.push(m[i]); + if(!pending.length) { + if(existsSync(resolve(root,'preparation.json'))) {status('completed');break;} + status('waiting_for_data'); await sleep(5000); continue; + } + for (const c of pending.slice(0, 150)) { + if (stop || existsSync(resolve(run,'STOP')) || attempts>=maxAttempts || inputTokens>=maxInputTokens) {stop=true;break;} + current=c.id; status('running'); + const started=Date.now(); + const base={id:c.id,kind:c.kind,dataset:c.dataset,expected:c.expected,labelType:c.labelType,sourceGroup:c.sourceGroup,sourceSplit:c.split??null,policyHash,model,threshold,at:new Date().toISOString()}; + let result; + try { + const options={key,model,mode:c.mode??'exact',...callbacks(c)}; + if(c.kind==='text') { + const boundary=textBoundary(c.text); + if(boundary) result={...base,predicted:null,boundary,sampleSha256:textDigest(c.text)}; + else { + const answer=await client.analyzeText(c.text,textPolicy,options); + const gate=textEligibility(answer,textPolicy,threshold); + result={...base,predicted:gate.status==='insufficient_evidence'?null:gate.eligible,answer,gate,sampleSha256:textDigest(c.text)}; + } + } else { + const image=await normalize(readFileSync(c.file)); + if(c.kind==='content') { + if(imageDigests.has(image.sha256)) result={...base,predicted:null,duplicate:true,normalizedSha256:image.sha256}; + else { + const answer=await client.analyze(image.bytes,contentPolicy,undefined,options); + const gate=contentEligibility(answer,contentPolicy,threshold); + result={...base,predicted:gate.status==='insufficient_evidence'?null:gate.eligible,answer,gate,normalizedSha256:image.sha256}; + imageDigests.add(image.sha256); + } + } else { + const reference=await normalize(readFileSync(c.reference)); + const answer=await client.compare(image.bytes,reference.bytes,options); + result={...base,predicted:answer.probability>=threshold,answer,normalizedSha256:image.sha256,referenceSha256:reference.sha256,mode:c.mode,hashEqual:image.sha256===reference.sha256}; + } + } + consecutiveFailures=0; + } catch(error) { + lastError=safeError(error); + if(lastError==='jev_authentication_failed' || lastError==='jev_rate_limit_deferred' || ++consecutiveFailures>=5) throw error; + result={...base,predicted:null,error:lastError}; + } + result.elapsedMs=Date.now()-started; + durable('results.jsonl',result); results.push(result);completed.add(c.id);current=null; + status('running'); + if(completed.size%10===0) console.log(JSON.stringify({completed:completed.size,attempts,inputTokens,metrics:metrics()})); + await sleep(1000); + } + } + if(stop) status('stopped_at_limit_or_request'); +} catch(error) { + lastError=safeError(error);status('blocked');process.exitCode=1; + console.error(JSON.stringify({error:lastError,completed:completed.size,attempts})); +} finally { + unlinkSync(lockPath); +} diff --git a/scripts/kaggle-benchmark.test.mjs b/scripts/kaggle-benchmark.test.mjs new file mode 100644 index 0000000..3e6c950 --- /dev/null +++ b/scripts/kaggle-benchmark.test.mjs @@ -0,0 +1,56 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { existsSync, mkdtempSync, mkdirSync, readFileSync, writeFileSync } from 'node:fs'; +import { resolve } from 'node:path'; +import { tmpdir } from 'node:os'; +import { spawnSync } from 'node:child_process'; + +test('benchmark persists decisions and usage, excludes source text, and resumes without paid duplicates', () => { + const scratch=process.platform==='win32' && existsSync('G:/datasets-deleteme/typecheck') ? 'G:/datasets-deleteme/typecheck' : tmpdir(); + const root=mkdtempSync(resolve(scratch,'typecheck-benchmark-test-')); + const runtime=resolve(root,'runs/test/runtime'); + mkdirSync(resolve(runtime,'server'),{recursive:true});mkdirSync(resolve(runtime,'shared')); + writeFileSync(resolve(runtime,'package.json'),JSON.stringify({type:'module'})); + const modules={ + 'server/jev.js':`export class JevClient {async analyzeText(text,d,o){await o.onAttempt({id:'attempt-'+text,retry:0,startedAt:new Date().toISOString()});await o.onComplete('attempt-'+text,{outcome:'success',inputTokens:1000000,error:null});return {probability:text==='secret-positive'?.99:.01};}}`, + 'server/images.js':`export const NORMALIZATION_HASH_VERSION='test';export const normalize=()=>{};`, + 'server/text.js':`export const textDigest=()=> 'digest';export const textEligibility=(r,d,t)=>({status:'detected',eligible:r.probability>=t});`, + 'shared/text.js':`export const textBoundary=x=>x==='over-limit'?'text_over_limit':null;`, + 'server/content.js':`export const contentEligibility=()=>{};`, + 'shared/content.js':`export const CONTENT_PRESETS=[{id:'scams',definition:{test:true}}];`, + 'server/security.js':`export const safeError=()=> 'test_error';`, + }; + for(const [path,code] of Object.entries(modules)) writeFileSync(resolve(runtime,path),code); + writeFileSync(resolve(root,'text-manifest.json'),JSON.stringify([ + {id:'positive',kind:'text',text:'secret-positive',expected:true}, + {id:'negative',kind:'text',text:'secret-negative',expected:false}, + {id:'over-limit',kind:'text',text:'over-limit',expected:true}, + ])); + writeFileSync(resolve(root,'preparation.json'),'{}'); + writeFileSync(resolve(root,'runs/test/pricing.json'),JSON.stringify({model:'jev-1.14.0',currency:'USD',perMillion:'0.042',source:'https://typesafe.ai/'})); + const env={...process.env,BENCHMARK_ROOT:root,BENCHMARK_RUN_ID:'test',JEV_MODEL:'jev-1.14.0',TYPESAFE_EVAL_KEY:'fake-offline-key',ALLOW_PAID_KAGGLE_EVALUATION:'true'}; + const invoke=()=>spawnSync(process.execPath,[resolve('scripts/kaggle-benchmark.mjs')],{env,encoding:'utf8',timeout:20000}); + let child=invoke();assert.equal(child.status,0,child.stderr); + const status=JSON.parse(readFileSync(resolve(root,'runs/test/status.json'))); + assert.equal(status.state,'completed');assert.equal(status.attempts,2);assert.equal(status.inputTokens,2000000); + assert.equal(status.cost.estimatedAmount,'0.084000');assert.equal(status.cost.costMicros,'84000'); + assert.equal(status.cost.pricedAttempts,2);assert.equal(status.cost.partial,false); + assert.equal(status.metrics.text.tp,1);assert.equal(status.metrics.text.tn,1);assert.equal(status.metrics.text.abstained,1); + const receipts=readFileSync(resolve(root,'runs/test/results.jsonl'),'utf8'); + assert.ok(!receipts.includes('secret-positive'));assert.ok(!receipts.includes('secret-negative')); + child=invoke();assert.equal(child.status,0,child.stderr); + assert.equal(readFileSync(resolve(root,'runs/test/results.jsonl'),'utf8'),receipts); + assert.equal(JSON.parse(readFileSync(resolve(root,'runs/test/status.json'))).cost.estimatedAmount,'0.084000'); + // Earlier receipts had no pricing: derive estimates without rewriting them. + const attemptPath=resolve(root,'runs/test/attempts.jsonl'); + const oldEvents=readFileSync(attemptPath,'utf8').trim().split('\n').map(line=>{ + const event=JSON.parse(line);delete event.pricing;delete event.costMicros;return event; + }); + writeFileSync(attemptPath,oldEvents.map(event=>JSON.stringify(event)).join('\n')+'\n'); + child=invoke();assert.equal(child.status,0,child.stderr); + const backfilled=JSON.parse(readFileSync(resolve(root,'runs/test/status.json'))).cost; + assert.equal(backfilled.estimatedAmount,'0.084000');assert.equal(backfilled.retrospectivelyPricedAttempts,2); + // An uncompleted external attempt must block automatic retries after a crash. + writeFileSync(resolve(root,'runs/test/attempts.jsonl'),JSON.stringify({event:'reserved',id:'unresolved'})+'\n',{flag:'a'}); + child=invoke();assert.notEqual(child.status,0);assert.match(child.stderr,/unreconciled_provider_attempt/); +}); diff --git a/scripts/prepare-kaggle-benchmark.py b/scripts/prepare-kaggle-benchmark.py new file mode 100644 index 0000000..e8fba07 --- /dev/null +++ b/scripts/prepare-kaggle-benchmark.py @@ -0,0 +1,228 @@ +"""Download non-email benchmark data to the explicitly selected G: scratch drive. + +No credentials are read. DISC21 ZIP members are fetched with HTTP ranges, avoiding +the 350 GB full download. Published labels remain source labels, not adjudications +of the application's stricter moderation policies. +""" +import concurrent.futures +import csv +import hashlib +import io +import json +import os +import pathlib +import random +import struct +import time +import zipfile +import zlib + +import requests + +ROOT = pathlib.Path(os.environ.get('BENCHMARK_ROOT', 'G:/datasets-deleteme/typecheck')) +RAW = ROOT / 'raw' +RAW.mkdir(parents=True, exist_ok=True) + + +def record(event, **values): + print(json.dumps({'at': time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()), 'event': event, **values}), flush=True) + + +def download(url, target): + target = pathlib.Path(target) + if target.exists(): + return target + target.parent.mkdir(parents=True, exist_ok=True) + part = target.with_suffix(target.suffix + '.part') + # Downloads are atomic. A partially transferred archive is never extracted. + for attempt in range(3): + try: + with requests.get(url, stream=True, timeout=(30, 90)) as response: + response.raise_for_status() + with part.open('wb') as out: + for chunk in response.iter_content(1024 * 1024): + out.write(chunk) + part.replace(target) + record('downloaded', file=str(target), bytes=target.stat().st_size) + return target + except requests.RequestException: + if attempt == 2: + raise + time.sleep(2 ** attempt) + + +def archive_dataset(ref, name): + metadata = requests.get('https://www.kaggle.com/api/v1/datasets/view/' + ref, timeout=30) + metadata.raise_for_status() + (RAW / (name + '-metadata.json')).write_text(json.dumps(metadata.json(), indent=2), encoding='utf-8') + archive = download('https://www.kaggle.com/api/v1/datasets/download/' + ref, RAW / (name + '.zip')) + dest = RAW / name + dest.mkdir(exist_ok=True) + with zipfile.ZipFile(archive) as z: + for member in z.infolist(): + path = (dest / member.filename).resolve() + if not path.is_relative_to(dest.resolve()): + raise ValueError('unsafe_archive_path') + z.extractall(dest) + return dest + + +class RemoteZip(io.RawIOBase): + """Seekable read-only ZIP transport; fails if server ignores byte ranges.""" + def __init__(self, url): + self.url = url + self.pos = 0 + r = requests.get(url, headers={'Range': 'bytes=-65536'}, timeout=60) + if r.status_code != 206: + raise ValueError('range_requests_required') + self.size = int(r.headers['Content-Range'].split('/')[-1]) + self.tail = r.content + + def seekable(self): + return True + + def seek(self, offset, whence=0): + self.pos = offset if whence == 0 else self.pos + offset if whence == 1 else self.size + offset + return self.pos + + def tell(self): + return self.pos + + def read(self, size=-1): + end = self.size if size < 0 else min(self.size, self.pos + size) + if end <= self.pos: + return b'' + if self.pos >= self.size - len(self.tail): + result = self.tail[self.pos - (self.size - len(self.tail)):end - (self.size - len(self.tail))] + else: + r = requests.get(self.url, headers={'Range': f'bytes={self.pos}-{end - 1}'}, timeout=60) + if r.status_code != 206 or len(r.content) != end - self.pos: + raise ValueError('invalid_range_response') + result = r.content + self.pos = end + return result + + +def prepare_disc(): + ref = 'hmendonca/disc21' + truth = download('https://www.kaggle.com/api/v1/datasets/download/' + ref + '/dev_ground_truth.csv', RAW / 'disc21/dev_ground_truth.csv') + pairs = list(csv.reader(truth.open())) + random.Random(20261017).shuffle(pairs) + # Restrict source references to one shard for an economical first run. + pairs = [(q, r) for q, r in pairs if r.startswith('R') and int(r[1:]) < 50000][:500] + if len(pairs) < 2: + raise ValueError('disc21_ground_truth_unavailable') + reference_url = 'https://dl.fbaipublicfiles.com/image_similarity_challenge/public/references_0.zip' + query_url = 'https://dl.fbaipublicfiles.com/image_similarity_challenge/public/dev_queries.zip' + def index(url): + with zipfile.ZipFile(RemoteZip(url)) as archive: + return {pathlib.PurePosixPath(n.filename).stem: n for n in archive.infolist() if n.filename.endswith('.jpg')} + references, queries = index(reference_url), index(query_url) + def member(url, entry, target): + if target.exists(): + return + start = entry.header_offset + # Include slack for local extra fields; verify the decompressed CRC/size. + response = requests.get(url, headers={'Range': f'bytes={start}-{start + entry.compress_size + 65535}'}, timeout=60) + response.raise_for_status() + if response.status_code != 206: + raise ValueError('range_requests_required') + data = response.content + if data[:4] != b'PK\x03\x04': + raise ValueError('invalid_zip_local_header') + name_len, extra_len = struct.unpack_from('= row['threshold'] if primary and probability is not None else row['predicted'] + if prediction is None: + m['abstained'] += 1 + else: + m['tp' if prediction and row['expected'] else 'fp' if prediction else 'fn' if row['expected'] else 'tn'] += 1 + decided = m['tp'] + m['tn'] + m['fp'] + m['fn'] + m['precision'] = m['tp'] / (m['tp'] + m['fp']) if m['tp'] + m['fp'] else None + m['recall_on_decided_cases'] = m['tp'] / (m['tp'] + m['fn']) if m['tp'] + m['fn'] else None + m['false_positive_rate_on_decided_negatives'] = m['fp'] / (m['fp'] + m['tn']) if m['fp'] + m['tn'] else None + m['coverage'] = decided / (len(rows) - m['duplicates']) if len(rows) > m['duplicates'] else None + m['all_positive_detection_rate'] = m['tp'] / sum(r['expected'] and not r.get('duplicate') for r in rows) if any(r['expected'] and not r.get('duplicate') for r in rows) else None + return m + + +def cost_summary(events, pricing, model): + """Use per-attempt integer rounding; retain unknown and pending coverage.""" + if pricing is None: + return None + if pricing['currency'] != 'USD' or pricing['model'] != model: + raise ValueError('benchmark_pricing_model_or_currency_mismatch') + reserved, completed = {}, {} + for event in events: + if event['event'] == 'reserved': + reserved.setdefault(event['id'], event) + elif event['event'] == 'completed': + completed.setdefault(event['id'], event) + micros, priced, retrospective, unknown, pending, tokens = 0, 0, 0, 0, 0, 0 + by_kind = collections.defaultdict(lambda: {'costMicros': 0, 'inputTokens': 0, 'pricedAttempts': 0}) + for id, reservation in reserved.items(): + if reservation.get('model', model) != model: + raise ValueError('mixed_model_billing_requires_separate_rates') + completion = completed.get(id) + if completion is None: + pending += 1 + continue + usage = completion.get('inputTokens') + if type(usage) is not int or usage < 0: + unknown += 1 + continue + rate = reservation.get('pricing') or pricing + if rate['currency'] != pricing['currency'] or rate['model'] != model: + raise ValueError('mixed_currency_or_model_in_attempt_pricing') + price_micros = int(Decimal(rate['perMillion']) * 1000000) + amount = int(completion['costMicros']) if completion.get('costMicros') is not None else (usage * price_micros + 500000) // 1000000 + micros += amount + tokens += usage + priced += 1 + retrospective += int(reservation.get('pricing') is None) + bucket = by_kind[reservation.get('kind', 'unknown')] + bucket['costMicros'] += amount + bucket['inputTokens'] += usage + bucket['pricedAttempts'] += 1 + return {'currency': 'USD', 'estimatedAmount': f'{micros // 1000000}.{micros % 1000000:06d}', 'costMicros': str(micros), 'pricedAttempts': priced, 'retrospectivelyPricedAttempts': retrospective, 'unknownUsageAttempts': unknown, 'pendingAttempts': pending, 'reportedInputTokens': tokens, 'partial': priced != len(reserved), 'pricing': pricing, 'byKind': {k: {**v, 'costMicros': str(v['costMicros'])} for k, v in by_kind.items()}, 'label': 'Estimated input-token spend at published list rates; not an invoice. Output listed as free; taxes, credits, discounts and other charges excluded.'} + + +def atomic_json(path, value): + temporary = path.with_suffix(f'.{os.getpid()}.tmp') + temporary.write_text(json.dumps(value, indent=2), encoding='utf-8') + for attempt in range(100): + try: + temporary.replace(path) + return + except PermissionError: + if attempt == 99: + raise + time.sleep(.05) + + +if __name__ == '__main__': + rows = read_jsonl(RUN / 'results.jsonl') + status = json.loads((RUN / 'status.json').read_text()) + pricing_file = RUN / 'pricing.json' + pricing = json.loads(pricing_file.read_text()) if pricing_file.exists() else None + costs = cost_summary(read_jsonl(RUN / 'attempts.jsonl'), pricing, status['model']) + report = {'state': status['state'], 'asOf': status['updatedAt'], 'interim': status['state'] != 'completed', 'completed': len(rows), 'attempts': status['attempts'], 'reportedInputTokens': status['inputTokens'], 'unknownUsageAttempts': status['unknownUsage'], 'estimatedDollarCost': None, 'model': status['model'], 'threshold': status['threshold'], 'policyHash': status['policyHash'], 'emailsIncluded': False, 'datasets': {}, 'limitations': ['Published annotations have not been independently adjudicated.', 'Spam is a proxy label for scams, and DISC21 source identity is a proxy for partial matching.', 'Primary-score agreement does not include safety/context gates and is not the enforcement decision.', 'Abstentions and failures are excluded from decided-case recall; all-positive detection rate includes them.', 'Public benchmark contamination in the provider model is unknown.', 'Naturalistic Discord accuracy and production acceptance remain unverified.']} + for kind in ['text', 'content', 'reference']: + group = [r for r in rows if r['kind'] == kind] + latency = sorted(r['elapsedMs'] for r in group if 'answer' in r) + report['datasets'][kind] = {'cases': len(group), 'gatedDecision': confusion(group), 'primaryScoreOnly': confusion(group, primary=True), 'latencyMs': {'p50': statistics.median(latency) if latency else None, 'p95': latency[min(len(latency) - 1, int(.95 * len(latency)))] if latency else None}, 'abstentionReasons': dict(collections.Counter(reason for r in group for reason in r.get('gate', {}).get('blockedReasons', []))), 'errors': dict(collections.Counter(r['error'] for r in group if r.get('error')))} + report['cost'] = costs + report['estimatedDollarCost'] = costs['estimatedAmount'] if costs else None + if costs: + report['reportedInputTokens'] = costs['reportedInputTokens'] + atomic_json(RUN / 'costs.json', costs) + atomic_json(RUN / 'report.json', report) + print(json.dumps({k: report[k] for k in ['state', 'interim', 'completed', 'attempts', 'reportedInputTokens', 'estimatedDollarCost']})) + if '--watch' in sys.argv: + # This helper only reads receipts; it never dispatches provider requests. + while status['state'] in ['running', 'waiting_for_data']: + time.sleep(30) + subprocess.run([sys.executable, __file__], check=True) + status = json.loads((RUN / 'status.json').read_text())