feat(evals): retain bounded native trace evidence in the Runner controller

Bind native exports and a separate root trace to each attempt and its reviewed operating allowance. Replay private receipts after worker deletion and retain holds on incomplete exports.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-24 14:30:03 -07:00
1 parent 39d53931f9
commit 9b7e77cf6c
12 files changed
+450 -42

No files matched your search

@@ -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
@@ -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$/;
@@ -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'));
});
@@ -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));
}
@@ -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 };
}
@@ -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 });
}
@@ -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) {
@@ -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';
+5 -4
View File
@@ -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));
};
+66
View File
@@ -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<string, { bodyBytes: number; recordCount: number; signal: keyof typeof signals; finished: boolean }>();
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 };
}
+17
View File
@@ -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<typeof fetch>(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);
});
+5 -4
View File
@@ -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<Promise<void>>(), controllers = new Set<AbortController>();
@@ -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"); }