mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-25 02:26:12 +02:00
fix(evals): verify stored gateway request coverage (#17432)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Product E2E campaigns retain a request ledger and send native traces and gateway logs to Braintrust. > - Native trace verification did not prove that each charged gateway request had a stored log row. > - A hosted probe showed that `x-bt-span-id` identifies the row `id`, while the stored `span_id` differs. Propagated campaign metadata was absent. > - This pull request joins retained request receipts to the correct Braintrust rows during scheduled finalization. > - Operators can see missing request coverage without changing costs, releasing holds, or rerunning agents. ## Linked Issues or Issue Description Refs #17354 and #17371. **Describe the bug** A campaign could pass native trace and experiment indexing checks while gateway request logs remained unverified. The runbook also claimed that propagated metadata made a request searchable after cancellation before response headers. The hosted gateway did not preserve that metadata in retained evidence. **Steps to reproduce** Inspect a retained gateway response header and query project logs by its `x-bt-span-id`. The matching record uses that value as `id`, not `span_id`. Compare every model reservation with its saved response receipt, including setup and canceled work. **Expected behavior** Verify each request against the stored row and report absent or conflicting evidence. Keep trace coverage separate from final usage and invoice reconciliation. ## What Changed - Added a bounded, read-only gateway index verifier. It matches row IDs, provider and model, and retains the actual span and trace IDs. - Included setup, failed and canceled requests in coverage. Missing receipts or row IDs, duplicate attribution, conflicting records and evidence changes prevent verification. - Added a separate scheduled gateway coverage gate and private receipt. Native trace and experiment checks remain independent. - Shared the existing bounded Braintrust query reader. Queries select only identity fields, with at most 100 IDs per batch, three rounds and a 45-second deadline. - Corrected the runbook's metadata guarantee and documented the historical `gatewaySpanId` receipt field. ## Verification - `pnpm test:e2e:runner:unit`: 4,120 passed, two existing skips; all 137 Node contracts passed. - Final repair: 54 focused tests passed, including scheduled-finalization coverage and all five optional identity fields. Each conflicting-field regression failed before the fix; absent or matching partial metadata remains supported. - `pnpm test:e2e:runner:typecheck`: passed after rebuilding stale local plugin SDK declarations. The initial stale-declaration failure remains recorded in the validation logs. - `git diff --check`: passed. - Read-only live verification of retained evidence: all ten local Codex requests and the one canceled drain-probe request matched stored rows. The hosted Claude campaign matched eight of ten; its two pre-header cancellations correctly remained incomplete. - The first hosted lookup exceeded the 1 MiB read limit when selecting full metadata. The final identity-only projection passed the same lookup. - Original journals stayed byte-identical. These checks made zero model calls and did not change provider data or release holds. - Full repository CI passed on `957ad0b684b80267d230fc4fb56d4f002ef089ad`: [run 40105676630](https://github.com/paperclipai/paperclip/actions/runs/40105676630), including typecheck, tests and build. Both current-head automated reviews scored 5/5 with no open findings. - The superseded initial CI run was canceled after the repair push to release the queue; its completed results remain available. No failed job was retried on the final commit. ## Risks Missing gateway IDs now fail scheduled trace coverage. A failed or canceled call may have no stored row, so this can require investigation even if product assertions pass. Indexing can lag; bounded query retries do not retry model work. Stored identity does not prove final billing or gateway span nesting under native attempt traces. Private identifiers remain in private evidence only. Recurrence and billing settings are unchanged. ## Model Used OpenAI Codex, GPT-6 family. The runtime does not expose a more specific model ID or context-window size in this session. Used reasoning, repository tools, local code execution and read-only API verification. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
f66d590899
commit
661f33f994
7 files changed
+423
-29
No files matched your search
@@ -156,16 +156,17 @@ keys. The server environment strips the gateway token and upstream credentials.
|
||||
|
||||
Every upstream attempt reserves its maximum estimated charge in an fsynced
|
||||
journal before dispatch. Requests, retries, and cancelled calls retain distinct
|
||||
IDs. Braintrust gateway requests propagate the reserved request ID, campaign,
|
||||
IDs. Braintrust gateway requests include the reserved request ID, campaign,
|
||||
execution, attempt and phase as allowlisted `paperclip_eval_*` metadata in the
|
||||
SDK-compatible parent context. The exact context is retained in the private
|
||||
receipt before dispatch, so a request canceled before response headers can
|
||||
still be looked up by `metadata.paperclip_eval_request_id` in project logs.
|
||||
This context contains no provider credential, prompt or response, and does not
|
||||
create an extra model call or a synthetic parent span. Direct provider routes
|
||||
receive no Braintrust context. Older project-only receipts cannot acquire this
|
||||
identity retroactively; time or session matches remain candidate evidence.
|
||||
A matching logged span does not itself prove complete usage or final billing.
|
||||
SDK-compatible parent context. The private receipt saves that context before
|
||||
dispatch. A live hosted probe did not retain the propagated metadata, so this
|
||||
header alone does not prove lookup by request ID after a pre-header cancellation.
|
||||
The returned `x-bt-span-id`, when present, identifies the Braintrust log row's
|
||||
`id`; the stored `span_id` and `root_span_id` can differ. The private receipt's
|
||||
historical field name `gatewaySpanId` continues to hold that returned row ID.
|
||||
Direct provider routes receive no Braintrust context. Older receipts cannot
|
||||
acquire a missing gateway row ID from a time or session match.
|
||||
A matching logged row does not itself prove complete usage or final billing.
|
||||
Unknown usage keeps the reservation held. One shared writer serializes
|
||||
admission across cells, so concurrent requests cannot each spend the same remaining
|
||||
allowance. Campaign deadline, request count, and in-flight count are also enforced.
|
||||
@@ -735,6 +736,24 @@ Braintrust can double-count native/log observations; use the deduplicated gatewa
|
||||
ledger as the estimate authority. Private indexing receipts contain only counts,
|
||||
identities and verification status, without trace contents or credentials.
|
||||
|
||||
Scheduled finalization separately verifies every non-infrastructure request in
|
||||
the campaign journal against its private gateway receipt. It queries Braintrust
|
||||
by row `id`, checks the expected provider and model, and retains the actual span
|
||||
and trace IDs in `gateway-indexing-receipt.json`. Setup, failed and canceled
|
||||
requests remain in the denominator. Missing receipts, missing row IDs, duplicate
|
||||
row attribution, conflicting records and query errors fail the gateway coverage
|
||||
gate. A receipt or journal change during inspection also prevents verification.
|
||||
The file reports whether campaign metadata was retained, independently of the
|
||||
receipt-to-row match. It contains no prompts, responses or credentials.
|
||||
|
||||
Each lookup batches at most 100 rows, makes at most three rounds, and has a
|
||||
45-second deadline. It performs no inference, writes no vendor records, changes
|
||||
no journal amounts and releases no holds. The findings include
|
||||
`gatewayIndexVerification`; a missing match is `gateway_indexing_incomplete`.
|
||||
This verifies stored identity and coverage, not provider billing or automatic
|
||||
nesting of gateway spans under native attempt traces. Those remain separate
|
||||
qualification requirements.
|
||||
|
||||
## Inspection and limits
|
||||
|
||||
Retained deterministic grades can be imported into a private Braintrust experiment
|
||||
|
||||
@@ -0,0 +1,174 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import path from "node:path";
|
||||
import { afterEach, expect, it, vi } from "vitest";
|
||||
import { verifyBraintrustGatewayIndexing } from "./braintrust-gateway-indexing.js";
|
||||
import { openEvalCostLedger, readEvalCostLedger } from "./cost-ledger.js";
|
||||
import { forwardEvalRequest, type EvalGatewayReceipt, type EvalGatewayRoute } from "./eval-gateway.js";
|
||||
|
||||
const directories: string[] = [];
|
||||
afterEach(async () => { for (const directory of directories.splice(0)) await rm(directory, { recursive: true, force: true }); });
|
||||
const config = { projectId: "11111111-1111-4111-8111-111111111111", apiKey: "private-key-canary" };
|
||||
async function fixture(count = 1, headers = true) {
|
||||
const directory = await mkdtemp(path.join(tmpdir(), "gateway-index-test-")); directories.push(directory);
|
||||
const ledgerPath = path.join(directory, "ledger.jsonl"), deadlineAt = Date.now() + 60_000;
|
||||
const limits = { campaignId: "campaign", maxUsd: "1", deadlineAt: new Date(deadlineAt).toISOString(), maxRequests: 20, maxConcurrent: 1 };
|
||||
const ledger = await openEvalCostLedger(ledgerPath, limits);
|
||||
await mkdir(`${ledgerPath}.requests`);
|
||||
const route: EvalGatewayRoute = { ...config, provider: "openai", mode: "braintrust", endpointName: "OPENAI_API_KEY",
|
||||
maximumInputTokens: 1000, maximumOutputTokens: 100, timeoutMs: 2000,
|
||||
price: { provider: "openai", model: "fixture", version: "fixture-1", sourceUrl: "https://provider.invalid/prices", inputUsdPerMillion: "2",
|
||||
cachedReadUsdPerMillion: "1", outputUsdPerMillion: "10", cacheWrite30mUsdPerMillion: "2", cacheWrite12hUsdPerMillion: "2" } };
|
||||
const rows: Array<Record<string, unknown>> = [], receipts: EvalGatewayReceipt[] = [];
|
||||
for (let i = 0; i < count; i++) {
|
||||
const rowId = randomUUID();
|
||||
receipts.push(await forwardEvalRequest({ ledger, identity: { executionId: "hello", attemptId: "attempt", phase: i === 0 ? "setup" : "agent" }, route,
|
||||
payload: { model: "fixture", input: "private-prompt-canary" }, deadlineAt,
|
||||
onHeaders: () => {}, onChunk: async () => {},
|
||||
persist: async receipt => { await writeFile(path.join(`${ledgerPath}.requests`, `${receipt.requestId}.json`), JSON.stringify(receipt)); },
|
||||
transport: async () => {
|
||||
if (!headers) throw new Error("private-upstream-error-canary");
|
||||
return Response.json({ object: "response", status: "completed", id: `provider-response-${i}`, model: "fixture",
|
||||
usage: { input_tokens: 10, input_tokens_details: { cache_write_tokens: 0, cached_tokens: 2 }, output_tokens: 5 } },
|
||||
{ headers: { "x-bt-used-endpoint": "OPENAI_API_KEY", "x-bt-span-id": rowId } });
|
||||
},
|
||||
}));
|
||||
rows.push({ id: rowId, span_id: randomUUID(), root_span_id: randomUUID(), provider: "openai", model: "fixture" });
|
||||
}
|
||||
await ledger.close();
|
||||
return { input: { ledgerPath, campaignId: "campaign", expected: [{ executionId: "hello", provider: "openai" as const, model: "fixture" }] },
|
||||
ledgerPath, rows, receipts, limits, receiptFile: (index = 0) => path.join(`${ledgerPath}.requests`, `${receipts[index]!.requestId}.json`) };
|
||||
}
|
||||
|
||||
it("joins setup and agent requests by returned row ID without adding indexed costs to the journal", async () => {
|
||||
const f = await fixture(2), before = await readFile(f.ledgerPath, "utf8");
|
||||
const transport = vi.fn<typeof fetch>(async () => Response.json({ data: f.rows }));
|
||||
const result = await verifyBraintrustGatewayIndexing(config, f.input, transport);
|
||||
expect(result).toMatchObject({ status: "verified", expectedRequests: 2, verifiedRequests: 2 });
|
||||
expect(result.requests.map(row => row.phase)).toEqual(["setup", "agent"]);
|
||||
expect(result.requests[0]).toMatchObject({ gatewayRowId: f.rows[0]!.id, spanId: f.rows[0]!.span_id,
|
||||
traceId: f.rows[0]!.root_span_id, campaignMetadataPresent: false });
|
||||
const sql = JSON.parse(String(transport.mock.calls[0]![1]!.body)).query;
|
||||
expect(sql).toContain(`WHERE id IN ('${f.rows[0]!.id}'`);
|
||||
expect(sql).not.toContain("WHERE span_id");
|
||||
expect(sql).not.toMatch(/[, ]metadata[, ]|input|output|prompt|completion/);
|
||||
expect(transport).toHaveBeenCalledTimes(1);
|
||||
expect(await readFile(f.ledgerPath, "utf8")).toBe(before);
|
||||
expect(JSON.stringify(result)).not.toMatch(/costUsd|private-|apiKey/);
|
||||
});
|
||||
|
||||
it("retains a pre-header failure as missing coverage and preserves its hold", async () => {
|
||||
const f = await fixture(1, false), transport = vi.fn<typeof fetch>();
|
||||
const before = (await readEvalCostLedger(f.ledgerPath)).state.snapshot();
|
||||
const result = await verifyBraintrustGatewayIndexing(config, f.input, transport);
|
||||
expect(result).toMatchObject({ status: "incomplete", expectedRequests: 1, verifiedRequests: 0 });
|
||||
expect(result.requests[0]?.reason).toBe("missing_gateway_row_id");
|
||||
expect(transport).not.toHaveBeenCalled();
|
||||
expect((await readEvalCostLedger(f.ledgerPath)).state.snapshot()).toEqual(before);
|
||||
expect(before.heldUsd).not.toBe("0.000000000");
|
||||
});
|
||||
|
||||
it.each(["missing", "malformed", "oversized", "wrong-request", "wrong-route", "wrong-model"])("reports a %s private receipt", async fault => {
|
||||
const f = await fixture(), transport = vi.fn<typeof fetch>();
|
||||
if (fault === "missing") await rm(f.receiptFile());
|
||||
else if (fault === "malformed") await writeFile(f.receiptFile(), "private-error-canary");
|
||||
else if (fault === "oversized") await writeFile(f.receiptFile(), "x".repeat(262_145));
|
||||
else await writeFile(f.receiptFile(), JSON.stringify({ ...f.receipts[0],
|
||||
...(fault === "wrong-request" ? { requestId: randomUUID() } : fault === "wrong-route" ? { route: "direct" }
|
||||
: { usage: { ...f.receipts[0]!.usage, model: "other" } }) }));
|
||||
const result = await verifyBraintrustGatewayIndexing(config, f.input, transport);
|
||||
expect(result.status).not.toBe("verified");
|
||||
expect(transport).not.toHaveBeenCalled();
|
||||
expect(JSON.stringify(result)).not.toContain("private-");
|
||||
});
|
||||
|
||||
it.each(["wrong-provider", "wrong-model", "invalid-trace", "duplicate", "foreign-row"])("rejects %s indexed evidence", async fault => {
|
||||
const f = await fixture();
|
||||
if (fault === "wrong-provider") f.rows[0]!.provider = "anthropic";
|
||||
if (fault === "wrong-model") f.rows[0]!.model = "other";
|
||||
if (fault === "invalid-trace") f.rows[0]!.root_span_id = "private-token-canary";
|
||||
if (fault === "duplicate") f.rows.push(f.rows[0]!);
|
||||
if (fault === "foreign-row") f.rows[0]!.id = randomUUID();
|
||||
const result = await verifyBraintrustGatewayIndexing(config, f.input, async () => Response.json({ data: f.rows }), { rounds: 1 });
|
||||
expect(result.status).not.toBe("verified");
|
||||
expect(result.verifiedRequests).toBe(0);
|
||||
expect(JSON.stringify(result)).not.toContain("private-");
|
||||
});
|
||||
|
||||
it.each(["request_id", "campaign_id", "execution_id", "attempt_id", "phase"])("rejects a conflicting %s even when other campaign metadata is absent", async field => {
|
||||
const f = await fixture(), before = await readFile(f.ledgerPath, "utf8");
|
||||
f.rows[0]![field] = "different-work";
|
||||
const transport = vi.fn<typeof fetch>(async () => Response.json({ data: f.rows }));
|
||||
const result = await verifyBraintrustGatewayIndexing(config, f.input, transport);
|
||||
expect(result).toMatchObject({ status: "incomplete", expectedRequests: 1, verifiedRequests: 0 });
|
||||
expect(result.requests[0]?.reason).toBe("indexed_identity_conflict");
|
||||
expect(transport).toHaveBeenCalledTimes(1);
|
||||
expect(await readFile(f.ledgerPath, "utf8")).toBe(before);
|
||||
});
|
||||
|
||||
it.each(["absent", "partial", "complete"])("allows %s campaign metadata when no identity contradicts the receipt", async coverage => {
|
||||
const f = await fixture();
|
||||
Object.assign(f.rows[0]!, { request_id: null, campaign_id: null, execution_id: null, attempt_id: null, phase: null });
|
||||
if (coverage !== "absent") f.rows[0]!.request_id = f.receipts[0]!.requestId;
|
||||
if (coverage === "complete") Object.assign(f.rows[0]!, { campaign_id: "campaign", execution_id: "hello", attempt_id: "attempt", phase: "setup" });
|
||||
const result = await verifyBraintrustGatewayIndexing(config, f.input, async () => Response.json({ data: f.rows }));
|
||||
expect(result).toMatchObject({ status: "verified", expectedRequests: 1, verifiedRequests: 1 });
|
||||
expect(result.requests[0]?.campaignMetadataPresent).toBe(coverage === "complete");
|
||||
});
|
||||
|
||||
it("refuses to attribute one gateway row to two charged requests", async () => {
|
||||
const f = await fixture(2), transport = vi.fn<typeof fetch>();
|
||||
await writeFile(f.receiptFile(1), JSON.stringify({ ...f.receipts[1], gatewaySpanId: f.receipts[0]!.gatewaySpanId }));
|
||||
const result = await verifyBraintrustGatewayIndexing(config, f.input, transport);
|
||||
expect(result.requests.map(row => row.reason)).toEqual(["duplicate_gateway_row_id", "duplicate_gateway_row_id"]);
|
||||
expect(result.status).toBe("incomplete");
|
||||
expect(transport).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("queries delayed rows again without querying rows already found", async () => {
|
||||
const f = await fixture(2), pause = vi.fn(async () => {});
|
||||
const transport = vi.fn<typeof fetch>().mockResolvedValueOnce(Response.json({ data: [f.rows[0]] }))
|
||||
.mockResolvedValueOnce(Response.json({ data: [f.rows[1]] }));
|
||||
const result = await verifyBraintrustGatewayIndexing(config, f.input, transport, { pause });
|
||||
expect(result.status).toBe("verified");
|
||||
expect(transport).toHaveBeenCalledTimes(2);
|
||||
const second = JSON.parse(String(transport.mock.calls[1]![1]!.body)).query;
|
||||
expect(second).not.toContain(f.rows[0]!.id);
|
||||
expect(second).toContain(f.rows[1]!.id);
|
||||
});
|
||||
|
||||
it.each(["rejected", "malformed", "oversized", "cancelled"])("keeps %s lookup failures visible without response bodies", async fault => {
|
||||
const f = await fixture();
|
||||
const transport = vi.fn<typeof fetch>(async () => fault === "rejected" ? new Response("private-canary", { status: 401 })
|
||||
: fault === "oversized" ? new Response("private-canary".repeat(100_000)) : new Response("private-canary"));
|
||||
const result = await verifyBraintrustGatewayIndexing(config, f.input, transport, { rounds: 1,
|
||||
...(fault === "cancelled" ? { signal: AbortSignal.abort() } : {}) });
|
||||
expect(result.status).toBe("unavailable");
|
||||
expect(JSON.stringify(result)).not.toContain("private-canary");
|
||||
if (fault === "cancelled") expect(transport).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it.each(["receipt", "journal"])("rejects evidence when the %s changes during lookup", async kind => {
|
||||
const f = await fixture();
|
||||
const transport = vi.fn<typeof fetch>(async () => {
|
||||
if (kind === "receipt") await writeFile(f.receiptFile(), JSON.stringify({ ...f.receipts[0], gatewaySpanId: randomUUID() }));
|
||||
else {
|
||||
const ledger = await openEvalCostLedger(f.ledgerPath, f.limits);
|
||||
await ledger.reserve(randomUUID(), { executionId: "hello", attemptId: "another", phase: "setup" }, "0.01");
|
||||
await ledger.close();
|
||||
}
|
||||
return Response.json({ data: f.rows });
|
||||
});
|
||||
const result = await verifyBraintrustGatewayIndexing(config, f.input, transport);
|
||||
expect(result.status).toBe("unavailable");
|
||||
expect(result.verifiedRequests).toBe(0);
|
||||
expect(result.requests[0]?.reason).toBe(kind === "receipt" ? "receipt_changed" : "ledger_changed");
|
||||
});
|
||||
|
||||
it("rejects a mismatched campaign or selection before lookup", async () => {
|
||||
const f = await fixture(), transport = vi.fn<typeof fetch>();
|
||||
await expect(verifyBraintrustGatewayIndexing(config, { ...f.input, campaignId: "other" }, transport)).rejects.toThrow("campaign mismatch");
|
||||
await expect(verifyBraintrustGatewayIndexing(config, { ...f.input, expected: [{ ...f.input.expected[0]!, executionId: "other" }] }, transport)).rejects.toThrow("selection mismatch");
|
||||
expect(transport).not.toHaveBeenCalled();
|
||||
});
|
||||
@@ -0,0 +1,124 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import { readFile, stat } from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import { setTimeout as delay } from "node:timers/promises";
|
||||
import type { EvalBraintrustConfig } from "./braintrust.js";
|
||||
import { queryBraintrustRows } from "./braintrust-query.js";
|
||||
import { readEvalCostLedger, validEvalLedgerId } from "./cost-ledger.js";
|
||||
|
||||
type Status = "verified" | "incomplete" | "unavailable";
|
||||
interface ExpectedExecution { executionId: string; provider: "openai" | "anthropic"; model: string }
|
||||
interface RequestIndex {
|
||||
requestId: string; executionId: string; attemptId: string; phase: string;
|
||||
status: Status; reason?: string; receiptSha256?: string;
|
||||
gatewayRowId?: string; spanId?: string; traceId?: string;
|
||||
campaignMetadataPresent?: boolean;
|
||||
}
|
||||
export interface BraintrustGatewayIndexReceipt {
|
||||
version: 1; projectId: string; campaignId: string; checkedAt: string;
|
||||
ledgerCheckpoint: string; status: Status; expectedRequests: number; verifiedRequests: number;
|
||||
requests: RequestIndex[];
|
||||
}
|
||||
const uuid = (value: unknown): value is string => typeof 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);
|
||||
const spanIdentity = (value: unknown): value is string => uuid(value) || typeof value === "string" && /^(?:[0-9a-f]{16}|[0-9a-f]{32})$/i.test(value) && !/^0+$/.test(value);
|
||||
const digest = (bytes: Buffer) => createHash("sha256").update(bytes).digest("hex");
|
||||
const object = (value: unknown): Record<string, unknown> => value && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : {};
|
||||
async function readReceipt(file: string) {
|
||||
if ((await stat(file)).size > 262_144) throw new Error("Oversized gateway receipt");
|
||||
const bytes = await readFile(file);
|
||||
if (bytes.length > 262_144) throw new Error("Oversized gateway receipt");
|
||||
return { bytes, value: object(JSON.parse(bytes.toString("utf8"))) };
|
||||
}
|
||||
|
||||
export async function verifyBraintrustGatewayIndexing(config: EvalBraintrustConfig, input: {
|
||||
ledgerPath: string; campaignId: string; expected: ExpectedExecution[];
|
||||
}, transport: typeof fetch = fetch, options: { rounds?: number; pause?: () => Promise<void>; signal?: AbortSignal } = {}): Promise<BraintrustGatewayIndexReceipt> {
|
||||
if (!uuid(config.projectId) || !config.apiKey?.trim() || /[\r\n]/.test(config.apiKey)
|
||||
|| !path.isAbsolute(input.ledgerPath) || !validEvalLedgerId(input.campaignId)
|
||||
|| !input.expected.length || input.expected.length > 1000 || new Set(input.expected.map(row => row.executionId)).size !== input.expected.length
|
||||
|| input.expected.some(row => !validEvalLedgerId(row.executionId) || !validEvalLedgerId(row.model) || !["openai", "anthropic"].includes(row.provider))) {
|
||||
throw new Error("Invalid gateway index scope");
|
||||
}
|
||||
const rounds = options.rounds ?? 3;
|
||||
if (!Number.isInteger(rounds) || rounds < 1 || rounds > 3) throw new Error("Invalid gateway index verification limit");
|
||||
const ledger = await readEvalCostLedger(input.ledgerPath);
|
||||
if (ledger.state.limits.campaignId !== input.campaignId) throw new Error("Gateway index campaign mismatch");
|
||||
const expected = new Map(input.expected.map(row => [row.executionId, row]));
|
||||
const modelRequests = [...ledger.state.requests].filter(([, request]) => request.identity.phase !== "infrastructure");
|
||||
if (modelRequests.length > 1000 || modelRequests.some(([id, request]) => !uuid(id) || !expected.has(request.identity.executionId))) {
|
||||
throw new Error("Gateway index request selection mismatch");
|
||||
}
|
||||
const receipt: BraintrustGatewayIndexReceipt = { version: 1, projectId: config.projectId, campaignId: input.campaignId,
|
||||
checkedAt: new Date().toISOString(), ledgerCheckpoint: ledger.previous, status: "incomplete",
|
||||
expectedRequests: modelRequests.length, verifiedRequests: 0, requests: [] };
|
||||
for (const [requestId, request] of modelRequests) {
|
||||
const row: RequestIndex = { requestId, ...request.identity, status: "incomplete" };
|
||||
receipt.requests.push(row);
|
||||
if (!request.finished) { row.reason = "request_in_flight"; continue; }
|
||||
try {
|
||||
const { bytes, value } = await readReceipt(path.join(`${input.ledgerPath}.requests`, `${requestId}.json`));
|
||||
row.receiptSha256 = digest(bytes);
|
||||
const planned = expected.get(request.identity.executionId)!;
|
||||
if (value.requestId !== requestId || value.route !== "braintrust"
|
||||
|| request.observations.some(observation => observation.route !== "braintrust" || observation.provider !== planned.provider
|
||||
|| observation.model !== undefined && observation.model !== planned.model)
|
||||
|| object(value.usage).provider !== undefined && object(value.usage).provider !== planned.provider
|
||||
|| object(value.usage).model !== undefined && object(value.usage).model !== planned.model) {
|
||||
row.reason = "receipt_identity_conflict"; continue;
|
||||
}
|
||||
if (!uuid(value.gatewaySpanId)) { row.reason = "missing_gateway_row_id"; continue; }
|
||||
row.gatewayRowId = value.gatewaySpanId;
|
||||
row.reason = "row_not_indexed";
|
||||
} catch { row.status = "unavailable"; row.reason = "receipt_unavailable"; }
|
||||
}
|
||||
const rowIds = new Map<string, RequestIndex[]>();
|
||||
for (const row of receipt.requests) if (row.gatewayRowId) rowIds.set(row.gatewayRowId, [...(rowIds.get(row.gatewayRowId) ?? []), row]);
|
||||
for (const rows of rowIds.values()) if (rows.length !== 1) for (const row of rows) row.reason = "duplicate_gateway_row_id";
|
||||
const signal = AbortSignal.any([AbortSignal.timeout(45_000), ...(options.signal ? [options.signal] : [])]);
|
||||
const pause = options.pause ?? (() => delay(2_000, undefined, { signal }));
|
||||
for (let round = 0; round < rounds && !signal.aborted; round++) {
|
||||
const pending = receipt.requests.filter(row => row.reason === "row_not_indexed" || row.reason === "query_unavailable");
|
||||
if (!pending.length) break;
|
||||
for (let offset = 0; offset < pending.length && !signal.aborted; offset += 100) {
|
||||
const batch = pending.slice(offset, offset + 100);
|
||||
try {
|
||||
const rows = await queryBraintrustRows(config, `SELECT id, span_id, root_span_id, metadata.provider AS provider, metadata.model AS model, metadata.paperclip_eval_request_id AS request_id, metadata.paperclip_eval_campaign_id AS campaign_id, metadata.paperclip_eval_execution_id AS execution_id, metadata.paperclip_eval_attempt_id AS attempt_id, metadata.paperclip_eval_phase AS phase FROM project_logs('${config.projectId}') WHERE id IN (${batch.map(row => `'${row.gatewayRowId}'`).join(",")}) LIMIT ${batch.length + 1}`, transport, signal);
|
||||
if (rows.length > batch.length || rows.some(row => !batch.some(expected => expected.gatewayRowId === row.id))) throw new Error("Unexpected gateway index rows");
|
||||
for (const row of batch) {
|
||||
const matches = rows.filter(value => value.id === row.gatewayRowId);
|
||||
row.status = "incomplete"; row.reason = "row_not_indexed";
|
||||
if (!matches.length) continue;
|
||||
const found = matches[0]!, planned = expected.get(row.executionId)!;
|
||||
const identity = { request_id: row.requestId, campaign_id: input.campaignId,
|
||||
execution_id: row.executionId, attempt_id: row.attemptId, phase: row.phase };
|
||||
if (matches.length !== 1 || !spanIdentity(found.span_id) || !spanIdentity(found.root_span_id)
|
||||
|| found.model !== planned.model || found.provider !== planned.provider
|
||||
|| Object.entries(identity).some(([field, value]) => found[field] != null && found[field] !== value)) {
|
||||
row.reason = "indexed_identity_conflict"; continue;
|
||||
}
|
||||
row.status = "verified"; delete row.reason;
|
||||
row.spanId = found.span_id; row.traceId = found.root_span_id;
|
||||
row.campaignMetadataPresent = Object.entries(identity).every(([field, value]) => found[field] === value);
|
||||
}
|
||||
} catch { for (const row of batch) { row.status = "unavailable"; row.reason = "query_unavailable"; } }
|
||||
}
|
||||
if (round + 1 < rounds && receipt.requests.some(row => ["row_not_indexed", "query_unavailable"].includes(row.reason ?? "")) && !signal.aborted) {
|
||||
try { await pause(); } catch { break; }
|
||||
}
|
||||
}
|
||||
for (const row of receipt.requests) if (row.receiptSha256) {
|
||||
try {
|
||||
const { bytes } = await readReceipt(path.join(`${input.ledgerPath}.requests`, `${row.requestId}.json`));
|
||||
if (digest(bytes) !== row.receiptSha256) throw new Error("Gateway receipt changed");
|
||||
} catch { row.status = "unavailable"; row.reason = "receipt_changed"; }
|
||||
}
|
||||
const after = await readEvalCostLedger(input.ledgerPath);
|
||||
if (after.previous !== ledger.previous || after.sequence !== ledger.sequence) {
|
||||
for (const row of receipt.requests) { row.status = "unavailable"; row.reason = "ledger_changed"; }
|
||||
}
|
||||
receipt.checkedAt = new Date().toISOString();
|
||||
receipt.verifiedRequests = receipt.requests.filter(row => row.status === "verified").length;
|
||||
receipt.status = receipt.expectedRequests > 0 && receipt.verifiedRequests === receipt.expectedRequests ? "verified"
|
||||
: signal.aborted || receipt.requests.some(row => row.status === "unavailable") ? "unavailable" : "incomplete";
|
||||
return receipt;
|
||||
}
|
||||
@@ -1,4 +1,5 @@
|
||||
import { setTimeout as delay } from "node:timers/promises";
|
||||
import { queryBraintrustRows as query } from "./braintrust-query.js";
|
||||
import { buildBraintrustExperiment, experimentEvidenceHash, type BraintrustExperimentReceipt } from "./braintrust-experiment.js";
|
||||
import { exportBraintrustAttemptParent, type EvalBraintrustConfig } from "./braintrust.js";
|
||||
import type { RunnerE2ECampaign, RunnerE2EResult } from "./types.js";
|
||||
@@ -13,25 +14,6 @@ export interface BraintrustIndexReceipt {
|
||||
experiment: { id: string | null; status: IndexStatus; expectedRows: number; indexedRows?: number };
|
||||
}
|
||||
|
||||
async function query(config: EvalBraintrustConfig, sql: string, transport: typeof fetch, signal: AbortSignal): Promise<Record<string, unknown>[]> {
|
||||
const response = await transport("https://api.braintrust.dev/btql", {
|
||||
method: "POST", redirect: "error", signal: AbortSignal.any([signal, AbortSignal.timeout(10_000)]),
|
||||
headers: { authorization: `Bearer ${config.apiKey}`, "content-type": "application/json" },
|
||||
body: JSON.stringify({ query: sql, fmt: "json" }),
|
||||
});
|
||||
const reader = response.body?.getReader(), chunks: Uint8Array[] = []; let size = 0;
|
||||
try {
|
||||
if (!response.ok) throw new Error("query_rejected");
|
||||
while (reader) {
|
||||
const next = await reader.read(); if (next.done) break;
|
||||
size += next.value.length; if (size > 1_048_576) throw new Error("query_oversized");
|
||||
chunks.push(next.value);
|
||||
}
|
||||
const payload = JSON.parse(Buffer.concat(chunks).toString("utf8"));
|
||||
if (!payload || !Array.isArray(payload.data) || payload.data.some((row: unknown) => !row || typeof row !== "object" || Array.isArray(row))) throw new Error("invalid_query_result");
|
||||
return payload.data;
|
||||
} finally { await reader?.cancel().catch(() => {}); }
|
||||
}
|
||||
|
||||
export function braintrustTraceIndexQuery(projectId: string, result: RunnerE2EResult): string {
|
||||
const trace = result.tracing;
|
||||
|
||||
@@ -0,0 +1,22 @@
|
||||
import type { EvalBraintrustConfig } from "./braintrust.js";
|
||||
|
||||
export async function queryBraintrustRows(config: EvalBraintrustConfig, sql: string, transport: typeof fetch, signal: AbortSignal): Promise<Record<string, unknown>[]> {
|
||||
const response = await transport("https://api.braintrust.dev/btql", {
|
||||
method: "POST", redirect: "error", signal: AbortSignal.any([signal, AbortSignal.timeout(10_000)]),
|
||||
headers: { authorization: `Bearer ${config.apiKey}`, "content-type": "application/json" },
|
||||
body: JSON.stringify({ query: sql, fmt: "json" }),
|
||||
});
|
||||
const reader = response.body?.getReader(), chunks: Uint8Array[] = []; let size = 0;
|
||||
try {
|
||||
if (!response.ok) throw new Error("query_rejected");
|
||||
while (reader) {
|
||||
const next = await reader.read(); if (next.done) break;
|
||||
size += next.value.length; if (size > 1_048_576) throw new Error("query_oversized");
|
||||
chunks.push(next.value);
|
||||
}
|
||||
const payload = JSON.parse(Buffer.concat(chunks).toString("utf8"));
|
||||
if (!payload || !Array.isArray(payload.data) || payload.data.some((row: unknown) => !row || typeof row !== "object" || Array.isArray(row))) throw new Error("invalid_query_result");
|
||||
return payload.data;
|
||||
} finally { await reader?.cancel().catch(() => {}); }
|
||||
}
|
||||
|
||||
@@ -8,6 +8,8 @@ import { openExperimentIngestion } from "./experiment-ingestion.js";
|
||||
import { collectScheduledIngestion } from "./ingestion-usage.js";
|
||||
import { importBraintrustExperiment, type BraintrustExperimentReceipt } from "./braintrust-experiment.js";
|
||||
import { verifyBraintrustIndexing } from "./braintrust-indexing.js";
|
||||
import { verifyBraintrustGatewayIndexing } from "./braintrust-gateway-indexing.js";
|
||||
import { runnerMatrix } from "./catalog.js";
|
||||
import { feedbackReviewTemplate, validateCampaignFeedbackReview, type CampaignFeedbackReview } from "./feedback-dispositions.js";
|
||||
import { expectedFixtureFeedback } from "./feedback-persistence.js";
|
||||
import { feedbackReview, summarizeCampaignFeedback, type FeedbackEvidence } from "./feedback.js";
|
||||
@@ -110,8 +112,24 @@ export async function finalizeScheduledCampaign(env: NodeJS.ProcessEnv) {
|
||||
await writeFile(path.join(directory, "indexing-receipt.json"), `${JSON.stringify(receipt, null, 2)}\n`, { mode: 0o600 });
|
||||
} catch {}
|
||||
if (indexing !== "verified") findings.push("trace_or_experiment_indexing_incomplete");
|
||||
let gatewayIndexing = "unavailable";
|
||||
try {
|
||||
if (!config || config.projectId !== manifest.projectId || manifest.route !== "braintrust") throw new Error("Missing gateway index scope");
|
||||
const expected = manifest.executionIds.map(executionId => {
|
||||
const execution = runnerMatrix.find(row => row.id === executionId);
|
||||
if (!execution) throw new Error("Missing gateway execution");
|
||||
const provider = execution.profile.credential === "ANTHROPIC_API_KEY" ? "anthropic" as const : "openai" as const;
|
||||
if (!manifest.models.some(model => model.provider === provider && model.price.model === execution.profile.model)) throw new Error("Missing gateway model");
|
||||
return { executionId, provider, model: execution.profile.model };
|
||||
});
|
||||
const receipt = await verifyBraintrustGatewayIndexing(config, { ledgerPath: manifest.ledgerPath,
|
||||
campaignId: manifest.limits.campaignId, expected });
|
||||
gatewayIndexing = receipt.status;
|
||||
await writeFile(path.join(directory, "gateway-indexing-receipt.json"), `${JSON.stringify(receipt, null, 2)}\n`, { mode: 0o600 });
|
||||
} catch {}
|
||||
if (gatewayIndexing !== "verified") findings.push("gateway_indexing_incomplete");
|
||||
await writeFile(path.join(directory, "findings.json"), `${JSON.stringify({ version: 1, findings,
|
||||
nativeIndexVerification: indexing, invoiceReconciliation: "pending",
|
||||
nativeIndexVerification: indexing, gatewayIndexVerification: gatewayIndexing, invoiceReconciliation: "pending",
|
||||
accountingStatus: campaign.operations ? evalAccountingStatus(campaign.operations) : "unavailable",
|
||||
feedbackReviewMethod: "Only exact synthetic fixture bodies with all corroborating graders are automatically dismissed; all other feedback requires review" }, null, 2)}\n`, { mode: 0o600 });
|
||||
for (const finding of findings) console.error(`::error title=Eval operations::${finding}`);
|
||||
|
||||
@@ -0,0 +1,55 @@
|
||||
import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import path from "node:path";
|
||||
import { afterEach, beforeEach, expect, it, vi } from "vitest";
|
||||
import { finalizeScheduledCampaign } from "./scheduled-finalization.js";
|
||||
import { verifyBraintrustGatewayIndexing } from "./braintrust-gateway-indexing.js";
|
||||
import { runnerMatrix } from "./catalog.js";
|
||||
import { buildRunnerCampaign } from "./history.js";
|
||||
import { emptyFeedbackSummary } from "./feedback.js";
|
||||
import type { RunnerE2EResult } from "./types.js";
|
||||
|
||||
vi.mock("./braintrust-gateway-indexing.js", () => ({ verifyBraintrustGatewayIndexing: vi.fn() }));
|
||||
vi.mock("./braintrust-indexing.js", () => ({ verifyBraintrustIndexing: vi.fn(async () => ({ status: "verified" })) }));
|
||||
vi.mock("./ingestion-usage.js", () => ({ collectScheduledIngestion: vi.fn(async () => ({ status: "complete" })) }));
|
||||
vi.mock("./experiment-ingestion.js", () => ({ openExperimentIngestion: vi.fn(async () => ({ transport: vi.fn(), close: async () => {} })) }));
|
||||
vi.mock("./braintrust-experiment.js", async importOriginal => ({ ...await importOriginal<typeof import("./braintrust-experiment.js")>(), importBraintrustExperiment: vi.fn() }));
|
||||
const directories: string[] = [];
|
||||
beforeEach(() => { vi.clearAllMocks(); });
|
||||
afterEach(async () => { for (const directory of directories.splice(0)) await rm(directory, { recursive: true, force: true }); });
|
||||
|
||||
it.each(["verified", "incomplete", "unavailable"] as const)("retains %s gateway coverage independently of native indexing", async status => {
|
||||
const workspace = await mkdtemp(path.join(tmpdir(), "scheduled-gateway-index-")); directories.push(workspace);
|
||||
const directory = path.join(workspace, "tests/runner-e2e/results/scheduled"), manifestPath = path.join(directory, "manifest.json");
|
||||
const campaignId = "scheduled-gateway-fixture", projectId = "11111111-1111-4111-8111-111111111111";
|
||||
const cell = runnerMatrix.find(row => row.id === "core-compatibility.runner-acpx-claude.local.message-marker")!;
|
||||
const root = path.join(workspace, "tests/runner-e2e/results", campaignId), resultPath = path.join(root, ...cell.id.split("."), "attempt-1");
|
||||
await mkdir(directory, { recursive: true }); await mkdir(resultPath, { recursive: true });
|
||||
const result: RunnerE2EResult = { schema: "paperclip.runner-e2e.result/v2", executionId: cell.id,
|
||||
suiteId: cell.suite.id, suiteDefinitionHash: cell.suiteDefinitionHash,
|
||||
accountingIdentity: { campaignId, attemptId: "one" }, source: { sha: "a".repeat(40), ref: null, workflowRunUrl: null },
|
||||
attempt: 1, status: "passed", cleanup: "passed", profileId: cell.profile.id, caseId: cell.task.id,
|
||||
environmentId: cell.environment.id, provider: cell.profile.provider, model: cell.profile.model, runtimeMode: cell.profile.expectedRuntimeMode,
|
||||
startedAt: "2026-10-24T00:00:00.000Z", finishedAt: "2026-10-24T00:00:01.000Z", durationMs: 1000,
|
||||
feedback: { ...emptyFeedbackSummary(), collection: "complete", invocationCollection: "complete" },
|
||||
tracing: { projectId, traceId: "a".repeat(32), spanId: "b".repeat(16), status: "accepted", verification: "unverified", nativeConfiguration: "configured",
|
||||
finalization: { spanId: "c".repeat(16), status: "accepted", outcome: "passed", verification: "unverified" } } };
|
||||
const campaign = buildRunnerCampaign({ campaignId, generatedAt: result.finishedAt, expected: [cell.id], results: [result], attempts: [result] });
|
||||
const ledgerPath = path.join(directory, "ledger.jsonl");
|
||||
await writeFile(manifestPath, JSON.stringify({ limits: { campaignId }, ledgerPath, executionIds: [cell.id], projectId, route: "braintrust",
|
||||
models: [{ provider: "anthropic", price: { model: cell.profile.model } }] }));
|
||||
await writeFile(path.join(root, "campaign.json"), JSON.stringify(campaign));
|
||||
await writeFile(path.join(resultPath, "result.json"), JSON.stringify(result));
|
||||
const indexReceipt = { version: 1 as const, projectId, campaignId, checkedAt: result.finishedAt,
|
||||
ledgerCheckpoint: "d".repeat(64), status, expectedRequests: 2, verifiedRequests: status === "verified" ? 2 : 1, requests: [] };
|
||||
vi.mocked(verifyBraintrustGatewayIndexing).mockResolvedValue(indexReceipt);
|
||||
const findings = await finalizeScheduledCampaign({ GITHUB_WORKSPACE: workspace, PAPERCLIP_E2E_OPERATIONS_MANIFEST: manifestPath,
|
||||
PAPERCLIP_E2E_BRAINTRUST_PROJECT_ID: projectId, BRAINTRUST_API_KEY: "private-key-canary" });
|
||||
expect(findings.includes("gateway_indexing_incomplete")).toBe(status !== "verified");
|
||||
expect(findings).not.toContain("trace_or_experiment_indexing_incomplete");
|
||||
expect(JSON.parse(await readFile(path.join(directory, "gateway-indexing-receipt.json"), "utf8"))).toEqual(indexReceipt);
|
||||
expect(JSON.parse(await readFile(path.join(directory, "findings.json"), "utf8"))).toMatchObject({ nativeIndexVerification: "verified", gatewayIndexVerification: status });
|
||||
expect(verifyBraintrustGatewayIndexing).toHaveBeenCalledWith(expect.objectContaining({ projectId }), {
|
||||
ledgerPath, campaignId, expected: [{ executionId: cell.id, provider: "anthropic", model: cell.profile.model }],
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user