feat(evals): retain bounded native tracing in the Runner controller (#17493)

Retain bounded native tracing in the Runner controller with attempt-bound relays, private export receipts, offline cost replay, conservative holds and bounded failure diagnostics.

Verified exact head 5cb991421e: full hosted CI, fresh Greptile 5/5, zero unresolved threads, 70 local controller tests including actual Codex/Claude CLI observability validators. Secondary review has no must-fix or should-fix findings and records optional coverage follow-ups. No paid evals or schedule changes.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

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));
   };

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<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 };
+}

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<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);
+});

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<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"); }