diff --git a/packages/paperclip-runner/docs/runner-protocol-live-evals.md b/packages/paperclip-runner/docs/runner-protocol-live-evals.md index c8762c97b4e..5c96e02108c 100644 --- a/packages/paperclip-runner/docs/runner-protocol-live-evals.md +++ b/packages/paperclip-runner/docs/runner-protocol-live-evals.md @@ -233,6 +233,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 @@ -323,8 +327,56 @@ 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. 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 fc328ee72a5..f17ef9b882c 100644 --- a/packages/paperclip-runner/scripts/controlled-evals/archive.mjs +++ b/packages/paperclip-runner/scripts/controlled-evals/archive.mjs @@ -12,6 +12,7 @@ const MAX_ARCHIVE_BYTES = 96 * MiB; 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], ]); 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/controller.test.mjs b/packages/paperclip-runner/scripts/controlled-evals/controller.test.mjs index 820b24280d0..de56b6fbaa7 100644 --- a/packages/paperclip-runner/scripts/controlled-evals/controller.test.mjs +++ b/packages/paperclip-runner/scripts/controlled-evals/controller.test.mjs @@ -4,7 +4,7 @@ import { mkdtemp, mkdir, readFile, readdir, rm, symlink, writeFile } from 'node: 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 { canonical, costRuntime, observabilityBoundConfiguration, requestBindingSha256, sha256, validateManifest } from './manifest.mjs'; import { attemptDirectory, runControlledEval } from './run.mjs'; import { summarizeControlledCampaign } from './report.mjs'; import { decodeControlledEvidence, packControlledAttempt, readControlledEvidence, readAttemptControlledEvidence, summarizeControlledEvidence } from './archive.mjs'; @@ -32,11 +32,22 @@ 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.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.orphanExit)})process.exit(0); `; @@ -68,11 +79,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') { @@ -91,7 +108,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'); @@ -473,3 +490,100 @@ test('campaign collection retains and audits private archives, while public proj await rm(join(attempt, 'controller-evidence.json')); 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'); + await assert.rejects(f.run(0, async () => { await rm(join(root, 'ledger.jsonl')); return success(); }), { code: 'ENOENT' }); + 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')); +}); diff --git a/packages/paperclip-runner/scripts/controlled-evals/manifest.mjs b/packages/paperclip-runner/scripts/controlled-evals/manifest.mjs index 0ca1bb655d0..333e13cb81d 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..6ceeeb7dc5a --- /dev/null +++ b/packages/paperclip-runner/scripts/controlled-evals/observability.mjs @@ -0,0 +1,97 @@ +import { randomBytes } from 'node:crypto'; +import { readFile, writeFile } from 'node:fs/promises'; +import { join } from 'node:path'; +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': 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 }) { + const identity = { projectId: manifest.observability.projectId, traceId: randomBytes(16).toString('hex'), spanId: randomBytes(8).toString('hex') }; + let native, controller; + const close = async () => { + const results = await Promise.allSettled([native?.close(), controller?.close()]); + if (results.some(result => result.status === 'rejected')) throw new Error('OTLP closure unverified'); + }; + try { + 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 }); + 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(); + try { + await native.close(); + const body = JSON.stringify(rootTrace(manifest, attempt, identity, startedAt, finishedAt, input.cleanupVerified, input.failure)); + 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.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: input.failure }, + deadlineAt, readEvidence: name => readFile(join(root, 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 4dc48445cf4..5a30f01f298 100644 --- a/packages/paperclip-runner/scripts/controlled-evals/report.mjs +++ b/packages/paperclip-runner/scripts/controlled-evals/report.mjs @@ -2,6 +2,7 @@ import { readFile } from 'node:fs/promises'; 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'; export async function summarizeControlledCampaign(value, { filesByAttempt } = {}) { @@ -21,7 +22,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')); @@ -57,6 +58,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) @@ -79,6 +85,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 cef1f272b0d..0c112cbc51e 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 { bindRequest, canonical, costRuntime, readJson, sha256, validateManifest } from './manifest.mjs'; async function atomicJson(file, value) { @@ -17,7 +18,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; @@ -60,7 +62,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)); @@ -82,6 +84,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 }); @@ -114,9 +118,9 @@ 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 gateway, observability, observabilityEvidence, child, artifact, failure = null, cleanupVerified = false, secretExcluded = true; let providerArtifactSha256; - const secrets = [key]; + const secrets = [...new Set([key, observabilityKey].filter(Boolean))]; const redact = raw => secrets.reduce((text, secret) => text.split(secret).join('[redacted]'), raw); try { await ledger.admitAttempt(attempt.executionId, attempt.attemptId); @@ -131,11 +135,16 @@ registerHooks({ resolve(specifier, context, nextResolve) { 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) { + observability = await startControlledObservability({ manifest, attempt, request, root, apiKey: observabilityKey, deadlineAt, startedAt, transport: otlpTransport }); + secrets.push(...observability.secrets); + childEnvironment.PAPERCLIP_RUNNER_EVAL_OTLP_TOKEN = observability.token; + } const artifactPath = join(root, 'provider-artifact.json'); 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()]) }); for (const stream of ['stdout', 'stderr']) await writeFile(join(root, `${stream}.log`), redact(child[stream]), { flag: 'wx', mode: 0o600 }); const raw = await readFile(artifactPath, 'utf8'); if (Buffer.byteLength(raw) > 16_777_216) throw new Error('Provider artifact exceeds its byte limit'); @@ -146,6 +155,7 @@ 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'); 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 @@ -157,28 +167,38 @@ registerHooks({ resolve(specifier, context, nextResolve) { } 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 readFile(join(root, 'receipts', `${request.requestId}.json`))); - } catch { failure ??= 'request_receipt_missing'; } - const ledgerSha256 = sha256(await readFile(ledgerPath)); - 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 readFile(join(root, 'receipts', `${request.requestId}.json`))); + } catch { failure ??= 'request_receipt_missing'; } + const ledgerSha256 = sha256(await readFile(ledgerPath)); + if (observability) { + try { + observabilityEvidence = await observability.finish({ artifact, cleanupVerified, failure }); + if (observabilityEvidence.status !== 'accepted') failure ??= 'observability_export_incomplete'; + } catch { failure ??= 'observability_evidence_incomplete'; } + } + 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 { await observability?.close(); } } 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"); }