diff --git a/packages/paperclip-runner/docs/runner-protocol-live-evals.md b/packages/paperclip-runner/docs/runner-protocol-live-evals.md index 4c7733d1f6e..03a7d7788b3 100644 --- a/packages/paperclip-runner/docs/runner-protocol-live-evals.md +++ b/packages/paperclip-runner/docs/runner-protocol-live-evals.md @@ -311,6 +311,10 @@ The manifest schema is `paperclip.runner-eval.controller/v1`. It contains: `requestSha256`, `provider`, optional `acpxAgent`, exact `model`, `maxUsd`, `maxRequests`, `maxConcurrent`, and `timeoutMs`. Native Codex and ACPX Claude are supported. Each execution has one explicit attempt and no retry. +- Optional `observability`: an explicit `projectId`, four boolean `capture` + switches (`userPrompts`, `assistantResponses`, `toolDetails`, `toolContent`), + `nativeLimits`, and a reviewed full-lifecycle `provisionalBound`. When absent, + the controller starts no exporter or OTLP relay. Use `requestBindingSha256` from `controlled-evals/manifest.mjs` to bind the complete compiled request. It excludes only the temporary working directory @@ -407,8 +411,60 @@ This local controller is not yet wired into the hosted matrix. Its durable attempt-directory guard applies to one shared filesystem, not independent GitHub workers. Hosted integration must bind unique cells to the prepared allocation, reject workflow/job reruns, and preserve the controller evidence -before enabling paid work. Native OTLP routing, report integration, vendor +before enabling paid work. Hosted tracing, destination readback, vendor metering bounds, and actual paid qualification remain separate gates. +## Controller tracing allocations and retained proof + +The opt-in controller uses the existing native launch generator and Product E2E +OTLP relay. `nativeLimits` must explicitly provide `maxRequests`, +`maxRequestBytes`, `maxTotalBytes`, `maxConcurrent`, `maxExpandedBytes`, and +`maxRecords`, each within that relay's supported ceilings. These limits apply +to each selected attempt, including failed attempts. Each attempt additionally +reserves one controller root request, at most 16 KiB of source JSON, 64 KiB +expanded source size, and one record. The root relay has its own credential; +the provider cannot consume its allocation or change its parent. + +Use `observabilityBoundConfiguration(manifest)` from `manifest.mjs` as the +configuration for the observability `provisionalBound`. It binds the complete +attempt selection, compiled-request digests, revisions, project, capture, +native and root capacities, deadline, and operating allowance. Its evidence +SHA-256 must match the observability operating component's evidence. Changing +these inputs invalidates the qualification. The reviewed evidence must cover +the sum of all attempts' native and root capacities and full retention costs; +source bytes and receiver acceptance do not themselves establish billed bytes, +price, or a vendor spending cap. This declaration does not activate an allowance. + +The controller requires `BRAINTRUST_API_KEY` even when model traffic uses a +direct provider key. Both upstream credentials stay on the controller. The +provider receives only its attempt's model and native OTLP tokens. Native +headers and any payload parent must match the exact assigned attempt parent, +including when another parent belongs to the same project. Cancellation closes +both relays, aborts pending forwarding, and retains uncertain outcomes without +retrying them. + +Private archives retain `observability.json`, body-free `otlp-native.jsonl` and +`otlp-controller.jsonl` receipts, and the metadata-only `otlp-root.json`. +Offline replay validates allocation, parent, descriptor, payload digests, +admission/outcome pairs, quotas, closure totals, and the root's identity and +content after the worker is deleted. Missing, partial, quota-exhausted, poisoned, +or unfinished export evidence keeps the attempt incomplete with its full model +allocation held and known model charges preserved. Operating holds remain +reserved once per campaign. OTLP usage is never added to the model ledger. + +`observability.status: "accepted"` requires accepted native log and trace +records plus the controller root. It describes the retained requests only; +`verification: "unverified"` remains until separate Braintrust readback. +Neither field proves complete capture, semantic grading success, or invoice +settlement. The root links revisions, the compiled request, execution, attempt, +provider, model, and cleanup status. Native receipts are checked before the root +is built. Its `controller_status_before_root_export` includes known native export +failures; the final retained report remains authoritative for root-export failures +or later cancellation. Cancellation stays connected through root finalization. +Feedback persistence is explicitly +unavailable in the Runner mock control plane; it does not report zero complaints +or suggestions. Actual grading and provider failures remain in the immutable +Runner artifact. + ## Hosted campaign Use the `Runner Direct Live Protocol Evals` workflow. Dispatch the workflow diff --git a/packages/paperclip-runner/scripts/controlled-evals/archive.mjs b/packages/paperclip-runner/scripts/controlled-evals/archive.mjs index 14194c48d8d..88f77864d44 100644 --- a/packages/paperclip-runner/scripts/controlled-evals/archive.mjs +++ b/packages/paperclip-runner/scripts/controlled-evals/archive.mjs @@ -10,6 +10,7 @@ const MiB = 1_048_576; const MAX_RAW_BYTES = 64 * MiB; const MAX_ARCHIVE_BYTES = 96 * MiB; + function secretValues({ environment = process.env, secrets = [] } = {}) { const configured = [...secrets, ...Object.entries(environment) .filter(([name]) => /(?:KEY|TOKEN|SECRET|PASSWORD|ACCESS_KEY_ID|CONFIG_CONTENT)$/i.test(name)) diff --git a/packages/paperclip-runner/scripts/controlled-evals/controller.test.mjs b/packages/paperclip-runner/scripts/controlled-evals/controller.test.mjs index 68498bd6bb6..6b374dfea81 100644 --- a/packages/paperclip-runner/scripts/controlled-evals/controller.test.mjs +++ b/packages/paperclip-runner/scripts/controlled-evals/controller.test.mjs @@ -1,11 +1,12 @@ import assert from 'node:assert/strict'; import { test } from 'node:test'; import { execFileSync } from 'node:child_process'; -import { mkdtemp, mkdir, readFile, readdir, rm, symlink, truncate, writeFile } from 'node:fs/promises'; +import { mkdtemp, mkdir, open, readFile, readdir, rm, stat, symlink, truncate, writeFile } from 'node:fs/promises'; import { writeFileSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; -import { canonical, costRuntime, requestBindingSha256, sha256, validateManifest } from './manifest.mjs'; +import { fileURLToPath } from 'node:url'; +import { canonical, costRuntime, observabilityBoundConfiguration, requestBindingSha256, sha256, validateManifest } from './manifest.mjs'; import { attemptDirectory, runControlledEval } from './run.mjs'; import { summarizeControlledCampaign } from './report.mjs'; import { readRegularFile, replaceEvidenceFile } from './evidence-files.mjs'; @@ -17,11 +18,12 @@ import assert from 'node:assert/strict'; import {createHash} from 'node:crypto'; import {spawn} from 'node:child_process'; import {once} from 'node:events'; -import {readFile,writeFile,rm,readdir,symlink,truncate} from 'node:fs/promises'; +import {readFile,writeFile,rm,readdir,symlink,truncate,mkdir} from 'node:fs/promises'; import {dirname,join} from 'node:path'; import {execFileSync} from 'node:child_process'; const args=new Map();for(let i=2;i{});"+(orphan?"process.send('ready');":"process.stdout.write('ready');")+"setInterval(()=>{},1000)"],{stdio:orphan?['ignore','inherit','inherit','ipc']:['ignore','pipe','ignore']}); @@ -36,11 +38,35 @@ for(let i=0;i<${options.requests ?? 3};i++){ const response=await fetch(descriptor.baseUrl+(descriptor.provider==='openai'?'/v1/responses':'/v1/messages'),{method:'POST',headers:{authorization:'Bearer '+process.env.PAPERCLIP_RUNNER_EVAL_GATEWAY_TOKEN,'content-type':'application/json'},body:JSON.stringify({model:request.model,input:'synthetic fixture'})}); statuses.push(response.status);await response.text(); } +let nativeObservability; +if(args.has('--observability-config')) { + const raw=await readFile(args.get('--observability-config'),'utf8'),config=JSON.parse(raw); + const parent=decodeURIComponent(config.launch.environment.OTEL_EXPORTER_OTLP_TRACES_HEADERS.slice('x-bt-parent='.length)); + for(const signal of ['logs','traces'])for(let i=0;i<${options.otelRequests ?? 1};i++) { + const body=signal==='logs'?{resourceLogs:[{scopeLogs:[{logRecords:[{body:{stringValue:'synthetic-native-content'}}]}]}]}:{resourceSpans:[{scopeSpans:[{spans:[{traceId:'1'.repeat(32),spanId:'2'.repeat(16),name:'synthetic-native-span'}]}]}]}; + const response=await fetch(config.baseUrl+'/v1/'+signal,{method:'POST',headers:{authorization:'Bearer '+process.env.PAPERCLIP_RUNNER_EVAL_OTLP_TOKEN,'x-bt-parent':parent,'content-type':'application/json'},body:JSON.stringify(body)}); + await response.arrayBuffer(); + } + nativeObservability={mode:'controller_otlp',configured:true,verification:'unverified',projectId:config.projectId,traceId:config.traceId,spanId:config.spanId,descriptorSha256:createHash('sha256').update(raw).digest('hex')}; + if(${Boolean(options.realObservabilityValidator)}) { + const {createRequire}=await import('node:module'); + const require=createRequire(${JSON.stringify(fileURLToPath(new URL('../../../../server/package.json', import.meta.url)))}); + const {tsImport}=await import(require.resolve('tsx/esm/api')); + const {prepareEvalObservabilityRuntime}=await tsImport(${JSON.stringify(fileURLToPath(new URL('../../src/cli/eval-observability-runtime.ts', import.meta.url)))},import.meta.url); + await mkdir(process.env.HOME+'/codex',{recursive:true}); + const prepared=await prepareEvalObservabilityRuntime(args.get('--observability-config'),request, + {evidence:{campaignId:descriptor.campaignId,executionId:descriptor.executionId},transportOptions:{environment:{CODEX_HOME:process.env.HOME+'/codex'}}},process.env); + assert.deepEqual(prepared.evidence,nativeObservability); + for(const signal of ['LOGS','TRACES'])assert(prepared.transportOptions.environment['OTEL_EXPORTER_OTLP_'+signal+'_HEADERS'].includes(encodeURIComponent('Bearer '+process.env.PAPERCLIP_RUNNER_EVAL_OTLP_TOKEN))); + nativeObservability=prepared.evidence; + } + +} if(${Boolean(options.waitForever)})await new Promise(()=>setInterval(()=>{},1000)); if(!${Boolean(options.missingArtifact)})await writeFile(args.get('--output'),JSON.stringify({schema:'paperclip-runner/eval-session-artifact/v1',attemptId:request.attemptId,requestedModel:request.model, requestRouting:${options.missingRoute ? 'null' : "{mode:'controller_gateway',provider:descriptor.provider,declaredUpstreamRoute:descriptor.upstreamRoute,campaignId:descriptor.campaignId,executionId:descriptor.executionId,attemptId:descriptor.attemptId,descriptorSha256:createHash('sha256').update(descriptorRaw).digest('hex')}"}, - snapshot:{status:'closed',process:{runnerExited:true,runnerPid:process.pid,providerPid:process.pid}},statuses, - leaked:${options.leakToken ? 'process.env.PAPERCLIP_RUNNER_EVAL_GATEWAY_TOKEN' : 'null'}})); + nativeObservability, snapshot:{status:'closed',process:{runnerExited:true,runnerPid:process.pid,providerPid:process.pid}},statuses, + leaked:${options.leakToken ? 'process.env.PAPERCLIP_RUNNER_EVAL_GATEWAY_TOKEN' : options.leakOtlpToken ? 'process.env.PAPERCLIP_RUNNER_EVAL_OTLP_TOKEN' : 'null'}})); if(${Boolean(options.invalidEvidence)}) { const root=dirname(args.get('--output')); const name=${JSON.stringify(options.invalidEvidence ?? '')}; @@ -51,6 +77,7 @@ if(${Boolean(options.invalidEvidence)}) { else if(kind==='symlink')await symlink('/dev/zero',file); else {await writeFile(file,'');await truncate(file,16*1024*1024+1);} } + if(${Boolean(options.orphanExit)})process.exit(0); `; @@ -82,11 +109,17 @@ async function fixture(t, { count = 1, allocation = '0.01', providerProfile = 'c await writeFile(requestPath, JSON.stringify(request)); requests.push({ request, requestPath, outputPath }); manifest.attempts.push({ executionId: `execution-${index}`, attemptId: request.attemptId, requestSha256: requestBindingSha256(request), provider: request.provider, ...(request.acpxAgent ? { acpxAgent: request.acpxAgent } : {}), model: 'fixture', maxUsd: allocation, maxRequests: 2, maxConcurrent: 1, timeoutMs: 5000 }); } + if (options.observability) { + manifest.observability = { projectId: '11111111-1111-4111-8111-111111111111', + capture: { userPrompts: false, assistantResponses: false, toolDetails: false, toolContent: false }, + nativeLimits: { maxRequests: 4, maxRequestBytes: 16384, maxTotalBytes: 32768, maxConcurrent: 1, maxExpandedBytes: 65536, maxRecords: 16 } }; + manifest.observability.provisionalBound = bound(observabilityBoundConfiguration(manifest)); + } const manifestPath = join(root, 'manifest.json'); const save = () => writeFile(manifestPath, JSON.stringify(manifest)); await save(); const environment = { PATH: process.env.PATH, OPENAI_API_KEY: 'synthetic-upstream-private', ANTHROPIC_API_KEY: 'other-private', BRAINTRUST_API_KEY: 'other-braintrust', GH_TOKEN: 'other-github', NODE_OPTIONS: '--require must-not-inherit', PYTHONPATH: '/must-not-inherit' }; return { root, manifest, manifestPath, requests, environment, save, - run: (index = 0, transport = () => success(biller), signal) => runControlledEval({ manifestPath, ...requests[index], environment, transport, signal }) }; + run: (index = 0, transport = () => success(biller), signal) => runControlledEval({ manifestPath, ...requests[index], environment, transport, signal, otlpTransport: options.otlpTransport ?? (async () => Response.json({})) }) }; } let sequence = 0; function success(provider = 'openai') { @@ -105,7 +138,7 @@ test('real child requests reserve before dispatch, stop at the cap and retain on assert.equal(ledger.state.snapshot().inflight, 1); return success(); }); - assert.equal(calls, 2); assert.equal(result.code, 0); assert.equal(result.report.cleanupVerified, true); + assert.equal(calls, 2); assert.equal(result.code, 0, JSON.stringify(result.report)); assert.equal(result.report.cleanupVerified, true); const artifact = JSON.parse(await readFile(f.requests[0].outputPath, 'utf8')); assert.deepEqual(artifact.statuses, [200, 200, 429]); assert.equal(artifact.controllerAccounting.accounting.knownUsd, '0.000136000'); @@ -406,10 +439,10 @@ for (const name of ['provider-artifact.json', 'receipt', 'ledger.jsonl']) for (c } test('durable reports bound every evidence file and retain full holds for invalid proof', { timeout: 20_000 }, async t => { - const f = await fixture(t, { requests: 1 }); await f.run(); + const f = await fixture(t, { requests: 1, observability: true }); await f.run(); const root = attemptDirectory(f.manifest, 'attempt-0'); const receipt = `receipts/${(await readdir(join(root, 'receipts')))[0]}`; - for (const name of ['admission.json', 'request.json', 'result.json', 'provider-artifact.json', 'gateway.json', 'ledger.jsonl', receipt]) { + for (const name of ['admission.json', 'request.json', 'result.json', 'provider-artifact.json', 'gateway.json', 'ledger.jsonl', receipt, 'observability.json', 'otlp-native.jsonl', 'otlp-controller.jsonl', 'otlp-root.json']) { const path = join(root, name), original = await readFile(path); for (const kind of ['fifo', 'symlink', 'oversized']) { await rm(path); @@ -534,6 +567,148 @@ test('campaign collection retains and audits private archives, while public proj await assert.rejects(aggregateProtocolEvalCampaign(input), /Missing private controller evidence/); }); +for (const providerProfile of ['codex', 'claude']) test(`controller owns native ${providerProfile} trace routing and offline receipts after workspace deletion`, async t => { + const received = []; + const f = await fixture(t, { providerProfile, observability: true, requests: 1, otlpTransport: async (url, init) => { + received.push({ url: String(url), headers: new Headers(init.headers), body: JSON.parse(Buffer.from(init.body).toString()) }); + return Response.json({}); + } }); + const result = await f.run(); + assert.equal(result.code, 0); assert.equal(received.length, 3); + const evidence = result.report.observability; + assert.equal(evidence.status, 'accepted'); assert.equal(evidence.verification, 'unverified'); + assert.equal(evidence.native.admittedRecords, 2); assert.equal(evidence.controller.admittedRecords, 1); + assert(received.every(item => item.headers.get('authorization') === 'Bearer other-braintrust')); + assert.equal(received[0].headers.get('x-bt-parent'), costRuntime.exportBraintrustAttemptParent(evidence)); + assert.equal(received[2].headers.get('x-bt-parent'), `project_id:${f.manifest.observability.projectId}`); + const root = received[2].body.resourceSpans[0].scopeSpans[0].spans[0]; + assert.equal(root.traceId, evidence.traceId); assert.equal(root.spanId, evidence.spanId); + assert(root.attributes.some(item => item.key === 'paperclip.eval.request_sha256' && item.value.stringValue === f.manifest.attempts[0].requestSha256)); + assert(!JSON.stringify(root).includes('synthetic-native-content')); + const archive = await packControlledAttempt({ manifest: f.manifest, attemptId: 'attempt-0', environment: f.environment }); + assert(!JSON.stringify(archive).includes('other-braintrust')); + await rm(f.manifest.outputDirectory, { recursive: true }); + await rm(f.requests[0].request.session.workingDirectory, { recursive: true }); + const summary = await summarizeControlledEvidence([archive]); + assert.equal(summary.incompleteAttempts, 0); assert.equal(summary.campaigns[0].attempts[0].observability.status, 'accepted'); + assert.equal(summary.campaigns[0].operatingHeldUsd, '0.004000000'); +}); + +test('native capacity, capture, selection and operating bound changes require new qualification', async t => { + const f = await fixture(t, { observability: true }); + for (const mutate of [ + value => { value.observability.nativeLimits.maxTotalBytes++; }, + value => { value.observability.capture.userPrompts = true; }, + value => { value.observability.projectId = '22222222-2222-4222-8222-222222222222'; }, + value => { value.attempts[0].requestSha256 = 'c'.repeat(64); }, + value => { value.overhead.components.find(item => item.component === 'observability').provisionalBound.evidenceSha256 = 'c'.repeat(64); }, + value => { delete value.observability.provisionalBound; }, + ]) { const manifest = structuredClone(f.manifest); mutate(manifest); assert.throws(() => validateManifest(manifest)); } + const noKey = { ...f.environment }; delete noKey.BRAINTRUST_API_KEY; + await assert.rejects(runControlledEval({ manifestPath: f.manifestPath, ...f.requests[0], environment: noKey }), /observability credential/); + await assert.rejects(readFile(join(f.manifest.outputDirectory, 'admission.json')), { code: 'ENOENT' }); +}); + +for (const fault of ['partial', 'quota', 'leak']) test(`native ${fault} preserves model cost and holds the unresolved attempt`, async t => { + const f = await fixture(t, { observability: true, requests: 1, otelRequests: fault === 'quota' ? 3 : 1, + leakOtlpToken: fault === 'leak', otlpTransport: async url => fault === 'partial' + ? Response.json({ partialSuccess: String(url).endsWith('/logs') ? { rejectedLogRecords: 1 } : { rejectedSpans: 1 } }) : Response.json({}) }); + const result = await f.run(); + assert.equal(result.code, 2); assert.equal(result.report.status, 'incomplete'); + const summary = await summarizeControlledCampaign(f.manifest); + assert.equal(summary.knownUsd, '0.000068000'); assert.equal(summary.committedUsd, '0.014000000'); + const artifact = JSON.parse(await readFile(f.requests[0].outputPath, 'utf8')); + assert.equal(artifact.accountingFailure.retryable, false); + if (fault === 'leak') { assert.equal(artifact.leaked, '[redacted]'); assert.equal(result.report.secretExcluded, false); } +}); + +test('offline replay refuses lost receipts and forged acceptance even when file hashes are updated', async t => { + const f = await fixture(t, { observability: true, requests: 1 }); await f.run(); + const root = attemptDirectory(f.manifest, 'attempt-0'), path = join(root, 'otlp-native.jsonl'); + const original = await readFile(path); + await rm(path); + assert.equal((await summarizeControlledCampaign(f.manifest)).incompleteAttempts, 1); + const rows = original.toString().trim().split('\n').map(line => JSON.parse(line)); + rows.find(row => row.type === 'outcome').outcome = 'partial'; + const bytes = Buffer.from(rows.map(row => JSON.stringify(row)).join('\n') + '\n'); + await writeFile(path, bytes); + const reportPath = join(root, 'result.json'), report = JSON.parse(await readFile(reportPath, 'utf8')); + report.observability.receiptHashes.native = sha256(bytes); await writeFile(reportPath, JSON.stringify(report)); + const summary = await summarizeControlledCampaign(f.manifest); + assert.equal(summary.incompleteAttempts, 1); assert.equal(summary.committedUsd, '0.014000000'); +}); + +test('cancellation aborts a pending native export and retains its uncertain receipt and model allocation', async t => { + const cancellation = new AbortController(); let aborted = false; + const f = await fixture(t, { observability: true, requests: 1, otlpTransport: async (_url, init) => { + queueMicrotask(() => cancellation.abort()); + return new Promise((_resolve, reject) => init.signal.addEventListener('abort', () => { aborted = true; reject(new Error('synthetic abort')); }, { once: true })); + } }); + const result = await f.run(0, () => success(), cancellation.signal); + assert.equal(result.code, 2); assert.equal(aborted, true); assert.equal(result.report.child.stopped, 'controller_cancelled'); + const rows = (await readFile(join(attemptDirectory(f.manifest, 'attempt-0'), 'otlp-native.jsonl'), 'utf8')).trim().split('\n').map(line => JSON.parse(line)); + assert.equal(rows.find(row => row.type === 'outcome').outcome, 'uncertain'); assert.equal(rows.at(-1).type, 'closed'); + const summary = await summarizeControlledCampaign(f.manifest); + assert.equal(summary.knownUsd, '0.000068000'); assert.equal(summary.committedUsd, '0.014000000'); +}); + +test('lost ledger finalization still closes both OTLP relays', async t => { + const f = await fixture(t, { observability: true, requests: 1 }); + const root = attemptDirectory(f.manifest, 'attempt-0'); + const result = await f.run(0, async () => { await rm(join(root, 'ledger.jsonl')); return success(); }); + assert.equal(result.code, 2); assert.equal(result.report.failure, 'ledger_evidence_invalid'); + assert.equal((await summarizeControlledCampaign(f.manifest)).committedUsd, '0.014000000'); + for (const file of ['otlp-native.jsonl', 'otlp-controller.jsonl']) { + const rows = (await readFile(join(root, file), 'utf8')).trim().split('\n').map(line => JSON.parse(line)); + assert.equal(rows.at(-1).type, 'closed'); + } + const descriptor = JSON.parse(await readFile(join(root, 'observability.json'), 'utf8')); + await assert.rejects(fetch(descriptor.baseUrl + '/v1/logs')); +}); + +for (const nativeFault of ['partial', 'quota']) test(`root includes ${nativeFault} native export failure before finalization`, async t => { + let rootPayload; + const f = await fixture(t, { observability: true, requests: 1, otelRequests: nativeFault === 'quota' ? 3 : 1, + otlpTransport: async (url, init) => { + if (new Headers(init.headers).get('x-bt-parent').startsWith('project_id:')) { + rootPayload = JSON.parse(Buffer.from(init.body).toString()); return Response.json({}); + } + return nativeFault === 'partial' ? Response.json({ partialSuccess: String(url).endsWith('/logs') ? { rejectedLogRecords: 1 } : { rejectedSpans: 1 } }) : Response.json({}); + } }); + const result = await f.run(); + assert.equal(result.code, 2); assert.equal(result.report.observability.traceFailure, 'native_export_incomplete'); + const attributes = rootPayload.resourceSpans[0].scopeSpans[0].spans[0].attributes; + assert(attributes.some(item => item.key === 'paperclip.eval.controller_status_before_root_export' && item.value.stringValue === 'incomplete')); +}); + +test('cancellation during root export stops forwarding and cannot close the attempt successfully', async t => { + const cancellation = new AbortController(); let rootAborted = false; + const f = await fixture(t, { observability: true, requests: 1, otlpTransport: async (_url, init) => { + if (!new Headers(init.headers).get('x-bt-parent').startsWith('project_id:')) return Response.json({}); + queueMicrotask(() => cancellation.abort()); + return new Promise((_resolve, reject) => init.signal.addEventListener('abort', () => { rootAborted = true; reject(new Error('synthetic root abort')); }, { once: true })); + } }); + const result = await f.run(0, () => success(), cancellation.signal); + assert.equal(rootAborted, true); assert.equal(result.code, 2); assert.equal(result.report.failure, 'controller_cancelled'); + assert.equal(result.report.status, 'incomplete'); assert.equal(result.report.cleanupVerified, true); + const root = attemptDirectory(f.manifest, 'attempt-0'); + const rows = (await readFile(join(root, 'otlp-controller.jsonl'), 'utf8')).trim().split('\n').map(line => JSON.parse(line)); + assert.equal(rows.find(row => row.type === 'outcome').outcome, 'uncertain'); assert.equal(rows.at(-1).type, 'closed'); + const summary = await summarizeControlledCampaign(f.manifest); + assert.equal(summary.committedUsd, '0.014000000'); assert.equal(summary.knownUsd, '0.000068000'); +}); + +for (const name of ['observability.json', 'otlp-native.jsonl', 'otlp-controller.jsonl']) for (const kind of ['fifo', 'symlink', 'oversized']) { + test(`native post-child ${name} rejects ${kind} and retains model cost plus the full hold`, { timeout: 20_000 }, async t => { + const f = await fixture(t, { requests: 1, observability: true, invalidEvidence: name, invalidFileType: kind }); + const result = await f.run(); + assert.equal(result.code, 2); + const summary = await summarizeControlledCampaign(f.manifest); + assert.equal(summary.incompleteAttempts, 1); + assert.equal(summary.knownUsd, '0.000068000'); + assert.equal(summary.committedUsd, '0.014000000'); + }); +} test('credential redaction replaces a swapped symlink without changing its target', async t => { const f = await fixture(t, { leakToken: true, requests: 0 }); @@ -574,3 +749,69 @@ test('missing provider artifacts cannot bypass immutable result binding', async await writeFile(join(directory, 'controller-evidence.json'), JSON.stringify(malformed)); await assert.rejects(readAttemptControlledEvidence(directory), /Malformed immutable controller result proof/); }); + + +test('failed final OTLP receipt write returns its saved incomplete report', async t => { + const f = await fixture(t, { observability: true, requests: 1 }); + const root = attemptDirectory(f.manifest, 'attempt-0'); + const probe = await open(join(f.root, 'prototype-probe'), 'wx'); + const prototype = Object.getPrototypeOf(probe), original = prototype.writeFile; + await probe.close(); + let failed = false; + prototype.writeFile = async function (data, ...options) { + if (String(data).includes('"type":"closed"')) { + const native = await stat(join(root, 'otlp-native.jsonl')).catch(() => null); + if (native && (await this.stat()).ino === native.ino) { + failed = true; + throw Object.assign(new Error('synthetic final receipt write failure'), { code: 'ENOSPC' }); + } + } + return original.call(this, data, ...options); + }; + t.after(() => { prototype.writeFile = original; }); + const result = await f.run(); + assert.equal(failed, true); assert.equal(result.code, 2); + assert.equal(result.report.status, 'incomplete'); + assert.deepEqual(JSON.parse(await readFile(join(root, 'result.json'), 'utf8')), result.report); + const summary = await summarizeControlledCampaign(f.manifest); + assert.equal(summary.knownUsd, '0.000068000'); assert.equal(summary.committedUsd, '0.014000000'); + const controller = (await readFile(join(root, 'otlp-controller.jsonl'), 'utf8')).trim().split('\n').map(line => JSON.parse(line)); + assert.equal(controller.at(-1).type, 'closed'); + const descriptor = JSON.parse(await readFile(join(root, 'observability.json'), 'utf8')); + await assert.rejects(fetch(descriptor.baseUrl + '/v1/logs')); +}); + + +for (const providerProfile of ['codex', 'claude']) test(`controller descriptor round trips through the real ${providerProfile} CLI validator`, async t => { + const f = await fixture(t, { providerProfile, observability: true, realObservabilityValidator: true, requests: 0 }); + const result = await f.run(); + assert.equal(result.code, 0, JSON.stringify(result.report) + '\n' + await readFile(join(attemptDirectory(f.manifest, 'attempt-0'), 'stderr.log'), 'utf8')); + assert.equal(result.report.observability.status, 'accepted'); + const bundle = await packControlledAttempt({ manifest: f.manifest, attemptId: 'attempt-0' }); + assert.equal((await summarizeControlledEvidence([bundle])).incompleteAttempts, 0); +}); + +test('retained failure codes distinguish native startup, child exit and invalid artifacts', async t => { + const startup = await fixture(t, { observability: true, requests: 0 }); + const root = attemptDirectory(startup.manifest, 'attempt-0'); + const probe = await open(join(startup.root, 'startup-probe'), 'wx'); + const prototype = Object.getPrototypeOf(probe), original = prototype.sync; await probe.close(); + let failed = false; + prototype.sync = async function (...args) { + const native = await stat(join(root, 'otlp-native.jsonl')).catch(() => null); + if (native && (await this.stat()).ino === native.ino) { + failed = true; throw Object.assign(new Error('synthetic native startup persistence failure'), { code: 'ENOSPC' }); + } + return original.call(this, ...args); + }; + t.after(() => { prototype.sync = original; }); + const result = await startup.run(); + prototype.sync = original; + assert.equal(failed, true); assert.equal(result.code, 2); + assert.equal(result.report.failure, 'observability_start_failed'); + assert.equal((await summarizeControlledCampaign(startup.manifest)).committedUsd, '0.014000000'); + const child = await fixture(t, { failChild: true, requests: 0 }); + assert.equal((await child.run()).report.failure, 'child_failed'); + const artifact = await fixture(t, { missingRoute: true, requests: 0 }); + assert.equal((await artifact.run()).report.failure, 'artifact_invalid'); +}); diff --git a/packages/paperclip-runner/scripts/controlled-evals/evidence-files.mjs b/packages/paperclip-runner/scripts/controlled-evals/evidence-files.mjs index 75b339258c8..14594399a98 100644 --- a/packages/paperclip-runner/scripts/controlled-evals/evidence-files.mjs +++ b/packages/paperclip-runner/scripts/controlled-evals/evidence-files.mjs @@ -5,6 +5,7 @@ const MiB = 1_048_576; export const FILE_LIMITS = new Map([ ['admission.json', MiB], ['request.json', 8 * MiB], ['gateway.json', 8192], ['ledger.jsonl', 16 * MiB], ['result.json', MiB], ['provider-artifact.json', 16 * MiB], + ['observability.json', 131_072], ['otlp-native.jsonl', 2 * MiB], ['otlp-controller.jsonl', 2 * MiB], ['otlp-root.json', 16_384], ['stdout.log', MiB], ['stderr.log', MiB], ]); export const RECEIPT = /^receipts\/[a-f0-9]{8}(?:-[a-f0-9]{4}){3}-[a-f0-9]{12}\.json$/; diff --git a/packages/paperclip-runner/scripts/controlled-evals/manifest.mjs b/packages/paperclip-runner/scripts/controlled-evals/manifest.mjs index ba1070a4b06..1c437cd0f1e 100644 --- a/packages/paperclip-runner/scripts/controlled-evals/manifest.mjs +++ b/packages/paperclip-runner/scripts/controlled-evals/manifest.mjs @@ -34,7 +34,7 @@ export async function readJson(file, maximum = 1_048_576) { } export function validateManifest(value, now = Date.now(), { admission = true } = {}) { fields(value, ['schema', 'campaignId', 'maxUsd', 'deadlineAt', 'pricingVerifiedAt', 'pricingExpiresAt', 'overhead', - 'sourceRevision', 'definitionsRevision', 'runnerCli', 'runnerPackage', 'runnerd', 'outputDirectory', 'route', 'projectId', 'models', 'attempts']); + 'sourceRevision', 'definitionsRevision', 'runnerCli', 'runnerPackage', 'runnerd', 'outputDirectory', 'route', 'projectId', 'models', 'attempts', 'observability']); const deadline = Date.parse(value.deadlineAt); const verified = Date.parse(value.pricingVerifiedAt), expires = Date.parse(value.pricingExpiresAt); if (value.schema !== 'paperclip.runner-eval.controller/v1' || !id(value.campaignId) @@ -81,6 +81,7 @@ export function validateManifest(value, now = Date.now(), { admission = true } = attempts.add(attempt.attemptId); executions.add(attempt.executionId); allocated += units(attempt.maxUsd); } if (allocated > units(value.maxUsd)) throw new Error('Controlled allocations exceed the campaign budget'); + validateObservability(value); return { manifest: structuredClone(value), manifestSha256: sha256(canonical(value)), allocatedUsd: usd(allocated), overheadUsd: usd(overhead) }; } export function bindRequest(manifest, request) { @@ -91,3 +92,28 @@ export function bindRequest(manifest, request) { || request.runnerd.path !== manifest.runnerd.path) throw new Error('Controlled request does not match its allocated identity'); return attempt; } + +export const ROOT_OTLP_LIMITS = Object.freeze({ maxRequests: 1, maxRequestBytes: 16_384, maxTotalBytes: 16_384, + maxConcurrent: 1, maxExpandedBytes: 65_536, maxRecords: 1 }); +export function observabilityBoundConfiguration(manifest) { + const policy = manifest.observability; + return { schema: 'paperclip.runner-eval.observability-bound/v1', campaignId: manifest.campaignId, deadlineAt: manifest.deadlineAt, + sourceRevision: manifest.sourceRevision, definitionsRevision: manifest.definitionsRevision, + projectId: policy.projectId, capture: policy.capture, nativeLimits: policy.nativeLimits, rootLimits: ROOT_OTLP_LIMITS, + attempts: manifest.attempts.map(({ executionId, attemptId, requestSha256 }) => ({ executionId, attemptId, requestSha256 })), + overhead: manifest.overhead }; +} +function validateObservability(manifest) { + const policy = manifest.observability; + if (policy === undefined) return; + fields(policy, ['projectId', 'capture', 'nativeLimits', 'provisionalBound']); + fields(policy.capture, ['userPrompts', 'assistantResponses', 'toolDetails', 'toolContent']); + if (!/^[a-f0-9]{8}(?:-[a-f0-9]{4}){3}-[a-f0-9]{12}$/i.test(policy.projectId ?? '') + || manifest.route === 'braintrust' && policy.projectId !== manifest.projectId + || Object.keys(policy.capture).length !== 4 || Object.values(policy.capture).some(value => typeof value !== 'boolean') + || canonical(Object.keys(policy.nativeLimits ?? {}).sort()) !== canonical(Object.keys(costRuntime.EVAL_OTLP_LIMITS).sort()) + || Object.entries(costRuntime.EVAL_OTLP_LIMITS).some(([key, max]) => !Number.isSafeInteger(policy.nativeLimits[key]) || policy.nativeLimits[key] < 1 || policy.nativeLimits[key] > max)) throw new Error('Invalid controller observability allocation'); + const operating = manifest.overhead.components.find(item => item.component === 'observability'); + if (!operating || policy.provisionalBound?.evidenceSha256 !== operating.provisionalBound?.evidenceSha256) throw new Error('Observability capacity requires its operating bound evidence'); + costRuntime.reservationBound('operating_allowance', policy.provisionalBound, observabilityBoundConfiguration(manifest)); +} diff --git a/packages/paperclip-runner/scripts/controlled-evals/observability.mjs b/packages/paperclip-runner/scripts/controlled-evals/observability.mjs new file mode 100644 index 00000000000..0bb3d142fdd --- /dev/null +++ b/packages/paperclip-runner/scripts/controlled-evals/observability.mjs @@ -0,0 +1,110 @@ +import { randomBytes } from 'node:crypto'; +import { writeFile } from 'node:fs/promises'; +import { join } from 'node:path'; +import { limitFor, readRegularFile } from './evidence-files.mjs'; +import { canonical, costRuntime, ROOT_OTLP_LIMITS, sha256 } from './manifest.mjs'; + +function launchFor(manifest, request, identity, baseUrl) { + const state = new costRuntime.CapabilityMockControlPlaneAdapter(request.session.seed).snapshot(); + return costRuntime.buildProviderObservabilityLaunch({ kind: request.provider === 'codex' ? 'codex' : 'claude', + companyId: request.session.companyId ?? state.company.id, agentId: request.session.actorId ?? state.actors[0].id, + compatibilityKey: identity.traceId, revision: identity.traceId, enabled: true, capture: manifest.observability.capture, + logsEndpoint: `${baseUrl}/v1/logs`, tracesEndpoint: `${baseUrl}/v1/traces`, + headers: { 'x-bt-parent': costRuntime.exportBraintrustAttemptParent(identity) } }); +} +const policyFor = (manifest, attempt, identity, deadlineAt, native) => ({ projectId: manifest.observability.projectId, + campaignId: manifest.campaignId, attemptId: attempt.attemptId, deadlineAt: new Date(deadlineAt).toISOString(), + limits: native ? manifest.observability.nativeLimits : ROOT_OTLP_LIMITS, + expectedParent: native ? costRuntime.exportBraintrustAttemptParent(identity) : `project_id:${identity.projectId}` }); +function rootTrace(manifest, attempt, identity, startedAt, finishedAt, cleanupVerified, failure) { + const attributes = { + 'paperclip.eval.family': 'runner_protocol', 'paperclip.eval.campaign_id': manifest.campaignId, + 'paperclip.eval.execution_id': attempt.executionId, 'paperclip.eval.attempt_id': attempt.attemptId, + 'paperclip.eval.source_revision': manifest.sourceRevision, 'paperclip.eval.definitions_revision': manifest.definitionsRevision, + 'paperclip.eval.request_sha256': attempt.requestSha256, 'paperclip.eval.provider': attempt.provider, + 'paperclip.eval.profile': attempt.acpxAgent ?? attempt.provider, 'paperclip.eval.model': attempt.model, + 'paperclip.eval.controller_status_before_root_export': failure ? 'incomplete' : 'closed', + 'paperclip.eval.cleanup_verified': String(cleanupVerified), 'paperclip.eval.feedback_persistence': 'unavailable_mock_control_plane', + 'paperclip.eval.grading': 'retained_artifact_separate', + }; + return { resourceSpans: [{ resource: { attributes: [{ key: 'service.name', value: { stringValue: 'paperclip-runner-evals' } }] }, + scopeSpans: [{ scope: { name: 'paperclip.runner-eval-controller', version: '1' }, spans: [{ traceId: identity.traceId, spanId: identity.spanId, + name: 'paperclip.runner.eval.attempt', kind: 1, startTimeUnixNano: String(BigInt(Date.parse(startedAt)) * 1_000_000n), + endTimeUnixNano: String(BigInt(Date.parse(finishedAt)) * 1_000_000n), + attributes: Object.entries(attributes).map(([key, value]) => ({ key, value: { stringValue: value } })) }] }] }] }; +} +const nativeEvidence = (identity, descriptorSha256) => ({ mode: 'controller_otlp', configured: true, verification: 'unverified', ...identity, descriptorSha256 }); + +export async function startControlledObservability({ manifest, attempt, request, root, apiKey, deadlineAt, startedAt, transport, signal }) { + const identity = { projectId: manifest.observability.projectId, traceId: randomBytes(16).toString('hex'), spanId: randomBytes(8).toString('hex') }; + let native, controller; + const onAbort = () => { void close().catch(() => {}); }; + const close = async () => { + signal?.removeEventListener('abort', onAbort); + const results = await Promise.allSettled([native?.close(), controller?.close()]); + if (results.some(result => result.status === 'rejected')) throw new Error('OTLP closure unverified'); + }; + try { + signal?.throwIfAborted(); + native = await costRuntime.startEvalOtlpRelay({ ...policyFor(manifest, attempt, identity, deadlineAt, true), deadlineAt, apiKey, + receiptPath: join(root, 'otlp-native.jsonl'), transport }); + controller = await costRuntime.startEvalOtlpRelay({ ...policyFor(manifest, attempt, identity, deadlineAt, false), deadlineAt, apiKey, + receiptPath: join(root, 'otlp-controller.jsonl'), transport }); + signal?.addEventListener('abort', onAbort, { once: true }); + signal?.throwIfAborted(); + const descriptor = { schema: 'paperclip.runner-eval.observability/v1', campaignId: manifest.campaignId, executionId: attempt.executionId, + attemptId: attempt.attemptId, ...identity, baseUrl: native.baseUrl, launch: launchFor(manifest, request, identity, native.baseUrl) }; + const raw = JSON.stringify(descriptor) + '\n'; + const descriptorPath = join(root, 'observability.json'); + await writeFile(descriptorPath, raw, { flag: 'wx', mode: 0o600 }); + let finished; + return { descriptorPath, token: native.token, secrets: [native.token, controller.token], + expectedEvidence: nativeEvidence(identity, sha256(raw)), close, closeNative: native.close, + finish: input => finished ??= (async () => { + const finishedAt = new Date().toISOString(); + let traceFailure = input.failure; + try { + await native.close(); + signal?.throwIfAborted(); + try { + const receipts = costRuntime.inspectEvalOtlpReceipts(await readRegularFile(join(root, 'otlp-native.jsonl'), limitFor('otlp-native.jsonl')), + policyFor(manifest, attempt, identity, deadlineAt, true)); + if (receipts.status !== 'accepted' || !receipts.signals.logs.records || !receipts.signals.traces.records) traceFailure ??= 'native_export_incomplete'; + } catch { traceFailure ??= 'native_evidence_incomplete'; } + const body = JSON.stringify(rootTrace(manifest, attempt, identity, startedAt, finishedAt, input.cleanupVerified, traceFailure)); + await writeFile(join(root, 'otlp-root.json'), body, { flag: 'wx', mode: 0o600 }); + const response = await fetch(`${controller.baseUrl}/v1/traces`, { method: 'POST', redirect: 'error', + signal: AbortSignal.any([...(signal ? [signal] : []), AbortSignal.timeout(Math.max(1, Math.min(10_000, deadlineAt - Date.now())))]), + headers: { authorization: `Bearer ${controller.token}`, 'content-type': 'application/json', 'x-bt-parent': `project_id:${identity.projectId}` }, body }); + await response.arrayBuffer(); + } finally { await close(); } + return inspectControlledObservability({ manifest, attempt, request, artifact: input.artifact, + report: { startedAt, traceFinishedAt: finishedAt, cleanupVerified: input.cleanupVerified, traceFailure }, + deadlineAt, readEvidence: name => readRegularFile(join(root, name), limitFor(name)) }); + })() }; + } catch (error) { await close(); throw error; } +} + +export async function inspectControlledObservability({ manifest, attempt, request, artifact, report, deadlineAt, readEvidence }) { + const descriptorBytes = await readEvidence('observability.json'); + if (descriptorBytes.length > 131_072) throw new Error('Oversized observability descriptor'); + const descriptor = JSON.parse(descriptorBytes.toString('utf8')); + const identity = { projectId: descriptor.projectId, traceId: descriptor.traceId, spanId: descriptor.spanId }; + if (!/^http:\/\/127\.0\.0\.1:[1-9][0-9]{0,4}\/otel$/.test(descriptor.baseUrl) || Number(new URL(descriptor.baseUrl).port) > 65535 + || descriptor.projectId !== manifest.observability.projectId || canonical(descriptor) !== canonical({ + schema: 'paperclip.runner-eval.observability/v1', campaignId: manifest.campaignId, executionId: attempt.executionId, attemptId: attempt.attemptId, + ...identity, baseUrl: descriptor.baseUrl, launch: launchFor(manifest, request, identity, descriptor.baseUrl) }) + || canonical(artifact?.nativeObservability) !== canonical(nativeEvidence(identity, sha256(descriptorBytes)))) throw new Error('Native observability identity or configuration mismatch'); + const bytes = { native: await readEvidence('otlp-native.jsonl'), controller: await readEvidence('otlp-controller.jsonl'), root: await readEvidence('otlp-root.json') }; + const native = costRuntime.inspectEvalOtlpReceipts(bytes.native, policyFor(manifest, attempt, identity, deadlineAt, true)); + const controller = costRuntime.inspectEvalOtlpReceipts(bytes.controller, policyFor(manifest, attempt, identity, deadlineAt, false)); + if (!Number.isFinite(Date.parse(report.traceFinishedAt)) || Date.parse(report.traceFinishedAt) < Date.parse(report.startedAt) + || bytes.root.length > ROOT_OTLP_LIMITS.maxRequestBytes || bytes.root.toString('utf8') !== JSON.stringify(rootTrace(manifest, attempt, identity, + report.startedAt, report.traceFinishedAt, report.cleanupVerified, report.traceFailure))) throw new Error('Controller root trace mismatch'); + const rootRows = bytes.controller.toString('utf8').trim().split('\n').map(line => JSON.parse(line)); + if (rootRows.filter(row => row.type === 'admitted').length !== 1 || !rootRows.some(row => row.type === 'admitted' && row.payloadSha256 === sha256(bytes.root))) throw new Error('Controller root trace receipt missing'); + return { ...identity, verification: 'unverified', status: native.status === 'accepted' && controller.status === 'accepted' + && native.signals.logs.records > 0 && native.signals.traces.records > 0 && controller.signals.traces.records === 1 ? 'accepted' : 'incomplete', + descriptorSha256: sha256(descriptorBytes), native, controller, receiptHashes: { native: sha256(bytes.native), controller: sha256(bytes.controller), root: sha256(bytes.root) }, + traceFinishedAt: report.traceFinishedAt, traceFailure: report.traceFailure }; +} diff --git a/packages/paperclip-runner/scripts/controlled-evals/report.mjs b/packages/paperclip-runner/scripts/controlled-evals/report.mjs index c7cb1664347..b3614f1b72a 100644 --- a/packages/paperclip-runner/scripts/controlled-evals/report.mjs +++ b/packages/paperclip-runner/scripts/controlled-evals/report.mjs @@ -1,6 +1,7 @@ import { join, resolve } from 'node:path'; import { pathToFileURL } from 'node:url'; import { canonical, costRuntime, readJson, requestBindingSha256, sha256, units, usd, validateManifest } from './manifest.mjs'; +import { inspectControlledObservability } from './observability.mjs'; import { attemptDirectory } from './run.mjs'; import { limitFor, readRegularFile } from './evidence-files.mjs'; @@ -23,7 +24,7 @@ export async function summarizeControlledCampaign(value, { filesByAttempt } = {} if (bytes.length > maximum) throw new Error('Controller document exceeds its byte limit'); return JSON.parse(bytes.toString('utf8')); }; - let snapshot, report, verified = false; + let snapshot, report, observability, verified = false; try { const ledgerBytes = await readEvidence('ledger.jsonl'); const ledger = costRuntime.parseEvalCostLedger(ledgerBytes.toString('utf8')); @@ -59,6 +60,11 @@ export async function summarizeControlledCampaign(value, { filesByAttempt } = {} const artifact = JSON.parse(artifactBytes.toString('utf8')); const descriptorBytes = await readEvidence('gateway.json'); const descriptor = JSON.parse(descriptorBytes.toString('utf8')); + if (manifest.observability) { + observability = await inspectControlledObservability({ manifest, attempt: allocation, request, artifact, + report: { ...report, traceFinishedAt: report.observability?.traceFinishedAt, traceFailure: report.observability?.traceFailure }, deadlineAt: ledgerDeadline, readEvidence }); + if (canonical(observability) !== canonical(report.observability) || observability.status !== 'accepted') throw new Error('Incomplete controller observability evidence'); + } else if (artifact.nativeObservability || report.observability) throw new Error('Unallocated observability evidence'); if (report.schema !== 'paperclip.runner-eval.control-result/v1' || report.manifestSha256 !== manifestSha256 || report.campaignId !== manifest.campaignId || report.executionId !== allocation.executionId || report.attemptId !== allocation.attemptId || report.requestSha256 !== allocation.requestSha256 || report.ledgerSha256 !== sha256(ledgerBytes) @@ -81,6 +87,7 @@ export async function summarizeControlledCampaign(value, { filesByAttempt } = {} const commitment = verified ? observedCommitment : observedCommitment > units(allocation.maxUsd) ? observedCommitment : units(allocation.maxUsd); known += cost; committed += commitment; attempts.push({ executionId: allocation.executionId, attemptId: allocation.attemptId, status: verified ? 'retained' : 'incomplete', + ...(manifest.observability ? { observability: observability ?? { status: 'incomplete', verification: 'unverified' } } : {}), allocationUsd: allocation.maxUsd, knownUsd: usd(cost), heldUsd: usd(commitment - cost), committedUsd: usd(commitment), unknownRequests: snapshot?.unknownRequests ?? null, boundedUnresolvedRequests: snapshot?.boundedUnresolvedRequests ?? null }); } diff --git a/packages/paperclip-runner/scripts/controlled-evals/run.mjs b/packages/paperclip-runner/scripts/controlled-evals/run.mjs index 48187a8991e..5389b2dba23 100644 --- a/packages/paperclip-runner/scripts/controlled-evals/run.mjs +++ b/packages/paperclip-runner/scripts/controlled-evals/run.mjs @@ -4,6 +4,7 @@ import { lstat, mkdir, open, readFile, realpath, rename, rm, writeFile } from 'n import { dirname, isAbsolute, join, relative, resolve, sep } from 'node:path'; import { pathToFileURL } from 'node:url'; import { parseArgs } from 'node:util'; +import { startControlledObservability } from './observability.mjs'; import { limitFor, readRegularFile, replaceEvidenceFile } from './evidence-files.mjs'; import { bindRequest, canonical, costRuntime, readJson, sha256, validateManifest } from './manifest.mjs'; @@ -18,7 +19,8 @@ const absent = pid => { try { process.kill(pid, 0); return false; } catch (error export const attemptDirectory = (manifest, attemptId) => join(manifest.outputDirectory, sha256(attemptId)); async function runChild(input) { - const child = spawn(process.execPath, ['--import', input.loader, input.cli, '--request', input.requestPath, '--output', input.artifactPath, '--gateway-config', input.descriptorPath], { + const child = spawn(process.execPath, ['--import', input.loader, input.cli, '--request', input.requestPath, '--output', input.artifactPath, '--gateway-config', input.descriptorPath, + ...(input.observabilityPath ? ['--observability-config', input.observabilityPath] : [])], { cwd: input.cwd, env: input.environment, detached: true, stdio: ['ignore', 'pipe', 'pipe'], }); let stopped = null, force, escalation; @@ -61,7 +63,7 @@ export function controlledResultArtifact(artifact, report, attempt) { controllerAccounting: report }; } -export async function runControlledEval({ manifestPath, requestPath, outputPath, environment = process.env, signal, transport }) { +export async function runControlledEval({ manifestPath, requestPath, outputPath, environment = process.env, signal, transport, otlpTransport }) { if (!['linux', 'darwin'].includes(process.platform)) throw new Error('Controlled eval process ownership requires Linux or macOS'); signal?.throwIfAborted(); const { manifest, manifestSha256 } = validateManifest(await readJson(manifestPath)); @@ -83,6 +85,8 @@ export async function runControlledEval({ manifestPath, requestPath, outputPath, const keyName = manifest.route === 'braintrust' ? 'BRAINTRUST_API_KEY' : provider === 'openai' ? 'OPENAI_API_KEY' : 'ANTHROPIC_API_KEY'; const key = environment[keyName]; if (!key || /[\r\n]/.test(key)) throw new Error('Missing controlled eval upstream credential'); + const observabilityKey = manifest.observability ? environment.BRAINTRUST_API_KEY : undefined; + if (manifest.observability && (!observabilityKey || /[\r\n]/.test(observabilityKey))) throw new Error('Missing controlled observability credential'); const model = manifest.models.find(item => item.provider === provider && item.price.model === attempt.model); const root = attemptDirectory(manifest, attempt.attemptId); await mkdir(manifest.outputDirectory, { recursive: true, mode: 0o700 }); @@ -115,12 +119,14 @@ registerHooks({ resolve(specifier, context, nextResolve) { await mkdir(join(root, 'tmp'), { mode: 0o700 }); const ledgerPath = join(root, 'ledger.jsonl'); const ledger = await costRuntime.openEvalCostLedger(ledgerPath, limits); - let gateway, child, artifact, failure = null, cleanupVerified = false, secretExcluded = true; - let providerArtifactSha256; - const secrets = [key]; + let gateway, observability, observabilityEvidence, child, artifact, failure = null, cleanupVerified = false, secretExcluded = true; + let providerArtifactSha256, failureStage = 'controlled_attempt_incomplete'; + const secrets = [...new Set([key, observabilityKey].filter(Boolean))]; const redact = raw => secrets.reduce((text, secret) => text.split(secret).join('[redacted]'), raw); try { + failureStage = 'request_admission_failed'; await ledger.admitAttempt(attempt.executionId, attempt.attemptId); + failureStage = 'gateway_start_failed'; gateway = await costRuntime.startEvalGateway({ ledger, identity: { executionId: attempt.executionId, attemptId: attempt.attemptId, phase: 'agent' }, route: { ...model, mode: manifest.route, projectId: manifest.projectId, endpointName: provider === 'openai' ? 'OPENAI_API_KEY' : 'ANTHROPIC_API_KEY', apiKey: key }, deadlineAt, transport, persist: receipt => atomicJson(join(root, 'receipts', `${receipt.requestId}.json`), receipt), @@ -129,15 +135,25 @@ registerHooks({ resolve(specifier, context, nextResolve) { const descriptor = { schema: 'paperclip.runner-eval.gateway/v1', provider, model: attempt.model, baseUrl: gateway.baseUrl, campaignId: manifest.campaignId, executionId: attempt.executionId, attemptId: attempt.attemptId, upstreamRoute: manifest.route }; const descriptorPath = join(root, 'gateway.json'), descriptorRaw = JSON.stringify(descriptor) + '\n'; + failureStage = 'gateway_descriptor_failed'; await writeFile(descriptorPath, descriptorRaw, { flag: 'wx', mode: 0o600 }); const childEnvironment = Object.fromEntries(['PATH', 'LANG', 'LANGUAGE', 'LC_ALL', 'TZ'].flatMap(name => environment[name] === undefined ? [] : [[name, environment[name]]])); Object.assign(childEnvironment, { HOME: join(root, 'home'), TMPDIR: join(root, 'tmp'), TMP: join(root, 'tmp'), TEMP: join(root, 'tmp'), PAPERCLIP_RUNNER_EVAL_GATEWAY_TOKEN: gateway.token }); + if (manifest.observability) { + failureStage = 'observability_start_failed'; + observability = await startControlledObservability({ manifest, attempt, request, root, apiKey: observabilityKey, deadlineAt, startedAt, transport: otlpTransport, signal }); + secrets.push(...observability.secrets); + childEnvironment.PAPERCLIP_RUNNER_EVAL_OTLP_TOKEN = observability.token; + } const artifactPath = join(root, 'provider-artifact.json'); + failureStage = 'child_failed'; signal?.throwIfAborted(); if (Date.now() >= deadlineAt) throw new Error('Controlled attempt deadline reached before spawn'); - child = await runChild({ cli: stagedCli, loader, requestPath: retainedRequestPath, artifactPath, descriptorPath, - cwd: request.session.workingDirectory ?? root, environment: childEnvironment, deadlineAt, signal, closeGateway: () => gateway.close() }); + child = await runChild({ cli: stagedCli, loader, requestPath: retainedRequestPath, artifactPath, descriptorPath, observabilityPath: observability?.descriptorPath, + cwd: request.session.workingDirectory ?? root, environment: childEnvironment, deadlineAt, signal, closeGateway: () => Promise.all([gateway.close(), observability?.close()]) }); + failureStage = 'output_retention_failed'; for (const stream of ['stdout', 'stderr']) await writeFile(join(root, `${stream}.log`), redact(child[stream]), { flag: 'wx', mode: 0o600 }); + failureStage = child.code === 0 ? 'artifact_invalid' : 'child_failed'; const raw = (await readRegularFile(artifactPath, limitFor('provider-artifact.json'))).toString('utf8'); secretExcluded = secrets.every(secret => !raw.includes(secret)); if (!secretExcluded) await replaceEvidenceFile(artifactPath, redact(raw)); @@ -146,6 +162,8 @@ registerHooks({ resolve(specifier, context, nextResolve) { const expected = { mode: 'controller_gateway', provider, declaredUpstreamRoute: manifest.route, campaignId: manifest.campaignId, executionId: attempt.executionId, attemptId: attempt.attemptId, descriptorSha256: sha256(descriptorRaw) }; if (canonical(artifact.requestRouting) !== canonical(expected) || artifact.attemptId !== attempt.attemptId || artifact.requestedModel !== attempt.model) throw new Error('Provider did not retain the assigned route'); + if (observability && canonical(artifact.nativeObservability) !== canonical(observability.expectedEvidence)) throw new Error('Provider did not retain the assigned observability route'); + failureStage = 'provider_cleanup_or_credential_unverified'; const processInfo = artifact.snapshot?.process; const pids = [...new Set([child.pid, processInfo?.runnerPid, processInfo?.providerPid, processInfo?.sidecarPid, processInfo?.agentPid].filter(pid => Number.isSafeInteger(pid) && pid > 0))]; cleanupVerified = !child.stopped && artifact.snapshot?.status === 'closed' && processInfo?.runnerExited === true @@ -153,34 +171,46 @@ registerHooks({ resolve(specifier, context, nextResolve) { && Number.isSafeInteger(processInfo.providerPid) && processInfo.providerPid > 0 && pids.every(absent); if (!cleanupVerified || !secretExcluded) throw new Error('Controlled provider cleanup or credential exclusion is unverified'); } catch { - failure = 'controlled_attempt_incomplete'; + failure = failureStage; } finally { try { await gateway?.close(); } catch { failure = 'gateway_cleanup_unverified'; } try { await ledger.close(); } catch { failure = 'ledger_close_failed'; } + try { await observability?.closeNative(); } catch { failure ??= 'observability_cleanup_unverified'; } } - if (cleanupVerified) { - try { await rm(executables, { recursive: true }); } - catch { cleanupVerified = false; failure ??= 'executable_cleanup_unverified'; } - } - const snapshot = ledger.snapshot(); - if (snapshot.overBudget || snapshot.conflictingRequests || snapshot.inflight || snapshot.unknownRequests !== snapshot.boundedUnresolvedRequests) failure ??= 'request_accounting_unbounded'; - const receiptHashes = {}; try { - for (const request of snapshot.requests) receiptHashes[request.requestId] = sha256(await readRegularFile(join(root, 'receipts', `${request.requestId}.json`), limitFor(`receipts/${request.requestId}.json`))); - } catch { failure ??= 'request_receipt_missing'; } - let ledgerSha256; - try { ledgerSha256 = sha256(await readRegularFile(ledgerPath, limitFor('ledger.jsonl'))); } - catch { failure ??= 'ledger_evidence_invalid'; } - const report = { schema: 'paperclip.runner-eval.control-result/v1', campaignId: manifest.campaignId, manifestSha256, - executionId: attempt.executionId, attemptId: attempt.attemptId, requestSha256: attempt.requestSha256, - allocationUsd: attempt.maxUsd, limits, startedAt, finishedAt: new Date().toISOString(), status: failure ? 'incomplete' : 'closed', failure, - child: child ? { code: child.code, signal: child.signal, stopped: child.stopped } : null, - cleanupVerified, secretExcluded, providerArtifactSha256, ledgerSha256, receiptHashes, accounting: snapshot }; - await atomicJson(join(root, 'result.json'), report); - const resultArtifact = controlledResultArtifact(artifact, report, attempt); - await mkdir(dirname(resolve(outputPath)), { recursive: true, mode: 0o700 }); - await writeFile(outputPath, JSON.stringify(resultArtifact, null, 2) + '\n', { flag: 'wx', mode: 0o600 }); - return { code: failure ? 2 : child?.code ?? 2, report }; + if (cleanupVerified) { + try { await rm(executables, { recursive: true }); } + catch { cleanupVerified = false; failure ??= 'executable_cleanup_unverified'; } + } + const snapshot = ledger.snapshot(); + if (snapshot.overBudget || snapshot.conflictingRequests || snapshot.inflight || snapshot.unknownRequests !== snapshot.boundedUnresolvedRequests) failure ??= 'request_accounting_unbounded'; + const receiptHashes = {}; + try { + for (const request of snapshot.requests) receiptHashes[request.requestId] = sha256(await readRegularFile(join(root, 'receipts', `${request.requestId}.json`), limitFor(`receipts/${request.requestId}.json`))); + } catch { failure ??= 'request_receipt_missing'; } + let ledgerSha256; + try { ledgerSha256 = sha256(await readRegularFile(ledgerPath, limitFor('ledger.jsonl'))); } + catch { failure ??= 'ledger_evidence_invalid'; } + if (observability) { + try { + observabilityEvidence = await observability.finish({ artifact, cleanupVerified, failure }); + if (observabilityEvidence.status !== 'accepted') failure ??= 'observability_export_incomplete'; + } catch { failure ??= 'observability_evidence_incomplete'; } + } + if (signal?.aborted) failure = 'controller_cancelled'; + const report = { schema: 'paperclip.runner-eval.control-result/v1', campaignId: manifest.campaignId, manifestSha256, + executionId: attempt.executionId, attemptId: attempt.attemptId, requestSha256: attempt.requestSha256, + allocationUsd: attempt.maxUsd, limits, startedAt, finishedAt: new Date().toISOString(), status: failure ? 'incomplete' : 'closed', failure, + child: child ? { code: child.code, signal: child.signal, stopped: child.stopped } : null, + ...(manifest.observability ? { observability: observabilityEvidence ?? { status: 'incomplete', verification: 'unverified' } } : {}), + cleanupVerified, secretExcluded, providerArtifactSha256, ledgerSha256, receiptHashes, accounting: snapshot }; + await atomicJson(join(root, 'result.json'), report); + const resultArtifact = controlledResultArtifact(artifact, report, attempt); + await mkdir(dirname(resolve(outputPath)), { recursive: true, mode: 0o700 }); + await writeFile(outputPath, JSON.stringify(resultArtifact, null, 2) + '\n', { flag: 'wx', mode: 0o600 }); + return { code: failure ? 2 : child?.code ?? 2, report }; + } finally { try { await observability?.close(); } catch {} } + } if (process.argv[1] && import.meta.url === pathToFileURL(resolve(process.argv[1])).href) { diff --git a/packages/paperclip-runner/scripts/controlled-evals/shared-cost-runtime.ts b/packages/paperclip-runner/scripts/controlled-evals/shared-cost-runtime.ts index b8591390e09..f77c817efd2 100644 --- a/packages/paperclip-runner/scripts/controlled-evals/shared-cost-runtime.ts +++ b/packages/paperclip-runner/scripts/controlled-evals/shared-cost-runtime.ts @@ -1,3 +1,9 @@ export { EvalSpendState, openEvalCostLedger, readEvalCostLedger, parseEvalCostLedger, validEvalLedgerId } from '../../../../tests/runner-e2e/cost-ledger.js'; export { evalRequestReservation, startEvalGateway } from '../../../../tests/runner-e2e/eval-gateway.js'; export { EVAL_OVERHEAD_COMPONENTS, validateOperatingOverhead } from '../../../../tests/runner-e2e/operating-overhead.js'; +export { startEvalOtlpRelay, EVAL_OTLP_LIMITS } from '../../../../tests/runner-e2e/eval-otlp-relay.js'; +export { inspectEvalOtlpReceipts } from '../../../../tests/runner-e2e/eval-otlp-receipts.js'; +export { exportBraintrustAttemptParent } from '../../../../tests/runner-e2e/braintrust.js'; +export { reservationBound } from '../../../../tests/runner-e2e/cost-bound-evidence.js'; +export { buildProviderObservabilityLaunch } from '../../../adapter-utils/src/agent-observability.js'; +export { CapabilityMockControlPlaneAdapter } from '../../src/mock-core/capability-mock-control-plane-adapter.js'; diff --git a/tests/runner-e2e/eval-otlp-payload.ts b/tests/runner-e2e/eval-otlp-payload.ts index 345734f73ec..6ad9cf7c1c0 100644 --- a/tests/runner-e2e/eval-otlp-payload.ts +++ b/tests/runner-e2e/eval-otlp-payload.ts @@ -3,7 +3,8 @@ import { evalOtlpSchema as root, protobuf } from "./eval-otlp-schema.js"; export type EvalOtlpSignal = "traces" | "logs"; export const isEvalProjectId = (value: string) => /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(value); -export function validateEvalOtlpParent(value: string, projectId: string): string { +export function validateEvalOtlpParent(value: string, projectId: string, expectedParent?: string): string { + if (expectedParent !== undefined && value !== expectedParent) throw new Error("unexpected_otlp_parent"); if (!isEvalProjectId(projectId)) throw new Error("invalid_otlp_project"); if (value === `project_id:${projectId}`) return value; if (value.length > 2048 || !/^[A-Za-z0-9+/]+={0,2}$/.test(value)) throw new Error("invalid_otlp_parent"); @@ -47,7 +48,7 @@ function validateWire(body: Buffer, type: import("../../server/node_modules/prot inspect(type, body.length, 0); } -export function validateEvalOtlpPayload(body: Buffer, contentType: string, signal: EvalOtlpSignal, projectId: string) { +export function validateEvalOtlpPayload(body: Buffer, contentType: string, signal: EvalOtlpSignal, projectId: string, expectedParent?: string) { const type = root.lookupType(signal === "traces" ? "TracesData" : "LogsData"); if (contentType !== "application/json") validateWire(body, type); const decoded: unknown = contentType === "application/json" ? JSON.parse(body.toString("utf8")) : type.decode(body).toJSON(); @@ -63,11 +64,11 @@ export function validateEvalOtlpPayload(body: Buffer, contentType: string, signa if (!parent || Object.keys(parent).length !== 1) throw new Error("invalid_otlp_parent"); const text = parent.stringValue ?? parent.string_value; if (typeof text !== "string") throw new Error("invalid_otlp_parent"); - validateEvalOtlpParent(text, projectId); + validateEvalOtlpParent(text, projectId, expectedParent); } if (Object.hasOwn(row, "braintrust.parent")) { if (typeof row["braintrust.parent"] !== "string") throw new Error("invalid_otlp_parent"); - validateEvalOtlpParent(row["braintrust.parent"], projectId); + validateEvalOtlpParent(row["braintrust.parent"], projectId, expectedParent); } Object.values(row).forEach(item => inspect(item, depth + 1)); }; diff --git a/tests/runner-e2e/eval-otlp-receipts.ts b/tests/runner-e2e/eval-otlp-receipts.ts new file mode 100644 index 00000000000..ed3fd83b16f --- /dev/null +++ b/tests/runner-e2e/eval-otlp-receipts.ts @@ -0,0 +1,66 @@ +import { EVAL_OTLP_LIMITS } from "./eval-otlp-relay.js"; + +export interface EvalOtlpReceiptPolicy { + projectId: string; campaignId: string; attemptId: string; deadlineAt: string; expectedParent: string; + limits: { [K in keyof typeof EVAL_OTLP_LIMITS]: number }; +} +const equal = (a: unknown, b: unknown) => JSON.stringify(a, (_key, value) => value && typeof value === "object" && !Array.isArray(value) + ? Object.fromEntries(Object.entries(value).sort(([a], [b]) => a.localeCompare(b))) : value) + === JSON.stringify(b, (_key, value) => value && typeof value === "object" && !Array.isArray(value) + ? Object.fromEntries(Object.entries(value).sort(([a], [b]) => a.localeCompare(b))) : value); +const integer = (value: unknown): value is number => Number.isSafeInteger(value) && Number(value) >= 0; + +export function inspectEvalOtlpReceipts(bytes: Buffer, expected: EvalOtlpReceiptPolicy) { + if (bytes.length > 2_097_152 || !bytes.length || bytes[bytes.length - 1] !== 10) throw new Error("Incomplete OTLP receipts"); + const rows = bytes.toString("utf8").trimEnd().split("\n").map(line => { + if (Buffer.byteLength(line) > 8192) throw new Error("Oversized OTLP receipt"); + const row = JSON.parse(line); + if (!row || typeof row !== "object" || Array.isArray(row)) throw new Error("Invalid OTLP receipt"); + return row; + }); + const policy = rows.shift(), closed = rows.pop(); + const opened = Date.parse(policy?.openedAt), deadline = Date.parse(expected.deadlineAt); + if (!equal(policy, { type: "policy", version: 2, sizing: "otlp-v1.9-expanded-source-json", ...expected, openedAt: policy?.openedAt }) + || !Number.isFinite(opened) || !Number.isFinite(deadline) || opened >= deadline + || closed?.type !== "closed" || !Number.isFinite(Date.parse(closed.at)) || Date.parse(closed.at) < opened) throw new Error("OTLP receipt policy or closure mismatch"); + const usage = { admittedRequests: 0, admittedWireBytes: 0, admittedBytes: 0, admittedExpandedBytes: 0, + admittedRecords: 0, attemptedForwardBytes: 0, acceptedRequests: 0, incompleteRequests: 0 }; + const signals = { logs: { requests: 0, records: 0 }, traces: { requests: 0, records: 0 } }; + const requests = new Map(); + let quotaExhausted = false, pending = 0; + for (const row of rows) { + const at = Date.parse(row.at); + if (!Number.isFinite(at) || at < opened || at > Date.parse(closed.at)) throw new Error("Invalid OTLP receipt time"); + if (row.type === "quota_exhausted") { + if (quotaExhausted || Object.keys(usage).some(key => !integer(row[key]) || row[key] > closed[key])) throw new Error("Invalid OTLP quota receipt"); + quotaExhausted = true; + } else if (row.type === "admitted") { + if (at >= deadline || !/^[a-f0-9]{8}(?:-[a-f0-9]{4}){3}-[a-f0-9]{12}$/.test(row.id) || requests.has(row.id) + || !["logs", "traces"].includes(row.signal) || !/^[a-f0-9]{64}$/.test(row.payloadSha256) + || ![row.wireBytes, row.bodyBytes, row.expandedBytes, row.recordCount].every(integer) + || row.wireBytes > expected.limits.maxRequestBytes || row.bodyBytes > expected.limits.maxRequestBytes + || ++pending > expected.limits.maxConcurrent) throw new Error("Invalid OTLP admission receipt"); + requests.set(row.id, { bodyBytes: row.bodyBytes, recordCount: row.recordCount, signal: row.signal, finished: false }); + usage.admittedRequests++; usage.admittedWireBytes += row.wireBytes; usage.admittedBytes += row.bodyBytes; + usage.admittedExpandedBytes += row.expandedBytes; usage.admittedRecords += row.recordCount; + if (usage.admittedRequests > expected.limits.maxRequests || usage.admittedBytes > expected.limits.maxTotalBytes + || usage.admittedExpandedBytes > expected.limits.maxExpandedBytes || usage.admittedRecords > expected.limits.maxRecords) throw new Error("OTLP receipts exceed allocation"); + } else if (row.type === "outcome") { + const request = requests.get(row.id); + if (!request || request.finished || !["accepted", "partial", "rejected", "uncertain", "not_dispatched"].includes(row.outcome) + || typeof row.dispatched !== "boolean" || (row.outcome === "not_dispatched") !== !row.dispatched + || row.httpStatus !== undefined && (!integer(row.httpStatus) || row.httpStatus < 100 || row.httpStatus > 599) + || row.rejectedRecords !== undefined && !integer(row.rejectedRecords) + || ["accepted", "partial"].includes(row.outcome) && (!integer(row.rejectedRecords) || row.httpStatus < 200 || row.httpStatus >= 300 || row.httpStatus === undefined + || (row.outcome === "accepted") !== (row.rejectedRecords === 0))) throw new Error("Invalid OTLP outcome receipt"); + request.finished = true; pending--; + if (row.dispatched) usage.attemptedForwardBytes += request.bodyBytes; + if (row.outcome === "accepted") { + usage.acceptedRequests++; signals[request.signal].requests++; signals[request.signal].records += request.recordCount; + } else usage.incompleteRequests++; + } else throw new Error("Unknown OTLP receipt"); + } + if (pending || closed.poisoned !== false || Object.entries(usage).some(([key, value]) => closed[key] !== value)) throw new Error("Unfinished or inconsistent OTLP receipts"); + return { ...usage, signals, quotaExhausted, status: quotaExhausted || usage.incompleteRequests ? "incomplete" : usage.admittedRecords ? "accepted" : "empty", + verification: "unverified" as const }; +} diff --git a/tests/runner-e2e/eval-otlp-relay.test.ts b/tests/runner-e2e/eval-otlp-relay.test.ts index 7b220dc0f45..55f15eebe23 100644 --- a/tests/runner-e2e/eval-otlp-relay.test.ts +++ b/tests/runner-e2e/eval-otlp-relay.test.ts @@ -273,3 +273,20 @@ it("rejects unknown protobuf fields instead of excluding forwarded data from siz expect(response.status).toBe(400); expect(upstream).not.toHaveBeenCalled(); }); + +it("binds native relay headers and payload parents to the assigned attempt within the same project", async () => { + const expectedParent = exportBraintrustAttemptParent({ projectId, spanId: "1234567890abcdef", traceId: "abcdef0123456789abcdef0123456789" }); + const other = exportBraintrustAttemptParent({ projectId, spanId: "abcdef0123456789", traceId: "abcdef0123456789abcdef0123456789" }); + const upstream = vi.fn(async () => Response.json({})); + const f = await fixture({ expectedParent, transport: upstream }); + expect((await f.send()).status).toBe(400); + expect((await f.send("{}", { "x-bt-parent": other })).status).toBe(400); + for (const parent of [other, `project_id:${projectId}`]) { + const body = { resourceLogs: [{ resource: { attributes: [{ key: "braintrust.parent", value: { stringValue: parent } }] } }] }; + expect((await f.send(JSON.stringify(body), { "x-bt-parent": expectedParent }, "/v1/logs")).status).toBe(400); + } + expect(upstream).not.toHaveBeenCalled(); + expect((await f.send("{}", { "x-bt-parent": expectedParent })).status).toBe(200); + await f.relay.close(); + expect((await f.records())[0].expectedParent).toBe(expectedParent); +}); diff --git a/tests/runner-e2e/eval-otlp-relay.ts b/tests/runner-e2e/eval-otlp-relay.ts index d34efd814e5..bb71078e283 100644 --- a/tests/runner-e2e/eval-otlp-relay.ts +++ b/tests/runner-e2e/eval-otlp-relay.ts @@ -14,7 +14,7 @@ export const EVAL_OTLP_DEFAULT_LIMITS = { ...EVAL_OTLP_LIMITS, export async function startEvalOtlpRelay(input: { projectId: string; apiKey: string; campaignId: string; attemptId: string; receiptPath: string; deadlineAt: number; - limits?: { [K in keyof typeof EVAL_OTLP_LIMITS]: number }; transport?: typeof fetch; + limits?: { [K in keyof typeof EVAL_OTLP_LIMITS]: number }; transport?: typeof fetch; expectedParent?: string; }) { const limits = input.limits ?? EVAL_OTLP_DEFAULT_LIMITS; if (!isEvalProjectId(input.projectId) || !input.apiKey.trim() || /[\r\n]/.test(input.apiKey) @@ -24,6 +24,7 @@ export async function startEvalOtlpRelay(input: { || Object.keys(limits).sort().join(",") !== Object.keys(EVAL_OTLP_LIMITS).sort().join(",") || Object.entries(EVAL_OTLP_LIMITS).some(([key, maximum]) => !Number.isSafeInteger(limits[key as keyof typeof limits]) || limits[key as keyof typeof limits] < 1 || limits[key as keyof typeof limits] > maximum)) throw new Error("Invalid eval OTLP relay policy"); + if (input.expectedParent !== undefined) validateEvalOtlpParent(input.expectedParent, input.projectId); await mkdir(path.dirname(input.receiptPath), { recursive: true, mode: 0o700 }); const file = await open(input.receiptPath, "ax", 0o600); let writes = Promise.resolve(), poisoned = false, closed = false, quotaRecorded = false; @@ -33,7 +34,7 @@ export async function startEvalOtlpRelay(input: { return writes; }; try { await record({ type: "policy", version: 2, sizing: "otlp-v1.9-expanded-source-json", projectId: input.projectId, campaignId: input.campaignId, - attemptId: input.attemptId, limits, deadlineAt: new Date(input.deadlineAt).toISOString(), openedAt: new Date().toISOString() }); } + attemptId: input.attemptId, ...(input.expectedParent === undefined ? {} : { expectedParent: input.expectedParent }), limits, deadlineAt: new Date(input.deadlineAt).toISOString(), openedAt: new Date().toISOString() }); } catch (error) { await file.close(); throw error; } const token = randomBytes(32).toString("hex"); const pending = new Set>(), controllers = new Set(); @@ -57,7 +58,7 @@ export async function startEvalOtlpRelay(input: { if (!["application/json", "application/x-protobuf"].includes(contentType ?? "") || encoding && encoding !== "gzip" && encoding !== "identity") return deny(response, 415, "unqualified_encoding"); let body: Buffer, parent: string, wireBytes = 0, expandedBytes = 0, recordCount = 0; try { - parent = validateEvalOtlpParent(String(request.headers["x-bt-parent"] ?? `project_id:${input.projectId}`), input.projectId); + parent = validateEvalOtlpParent(String(request.headers["x-bt-parent"] ?? `project_id:${input.projectId}`), input.projectId, input.expectedParent); const chunks: Buffer[] = []; for await (const chunk of request) { wireBytes += chunk.length; @@ -66,7 +67,7 @@ export async function startEvalOtlpRelay(input: { } body = Buffer.concat(chunks); if (encoding === "gzip") body = gunzipSync(body, { maxOutputLength: limits.maxRequestBytes }); - const validated = validateEvalOtlpPayload(body, contentType!, signal, input.projectId); + const validated = validateEvalOtlpPayload(body, contentType!, signal, input.projectId, input.expectedParent); ({ body, expandedBytes, recordCount } = validated); if (body.length > limits.maxRequestBytes) throw new Error("request_limit"); } catch { return deny(response, 400, "invalid_or_oversized_payload"); }