diff --git a/packages/core/src/receipts.ts b/packages/core/src/receipts.ts index 3d087fe3..1636f463 100644 --- a/packages/core/src/receipts.ts +++ b/packages/core/src/receipts.ts @@ -8,6 +8,15 @@ const ReceiptCheckSchema = z.object({ exit_code: z.number().int().optional(), }); +// Non-gating evidence: named artifacts a run was expected to preserve. Distinct +// from `checks`, which the run's outcome is gated on. Optional so receipts +// sealed before this field existed still parse and digest unchanged. +const ReceiptEvidenceSchema = z.object({ + name: z.string(), + status: z.enum(["passed", "failed", "skipped", "unknown"]), + reason: z.string().optional(), +}); + export const FacilityReceiptSchema = z.object({ schema: z.literal("facility.run.v1"), run_id: z.string().optional(), @@ -71,6 +80,8 @@ export const FacilityReceiptSchema = z.object({ .optional(), checks: z.array(ReceiptCheckSchema).optional(), checks_truncated: z.boolean().optional(), + evidence: z.array(ReceiptEvidenceSchema).optional(), + evidence_truncated: z.boolean().optional(), integrity: z .object({ algorithm: z.literal("sha256"), diff --git a/packages/core/test/core.test.ts b/packages/core/test/core.test.ts index 88f001bc..7dcfeb89 100644 --- a/packages/core/test/core.test.ts +++ b/packages/core/test/core.test.ts @@ -468,4 +468,53 @@ describe("receipts", () => { }), ).toBe(false); }); + + it("seals non-gating evidence without disturbing receipts that carry none", () => { + const base = parseLegacyAgentReceipt({ + schema: "example.agent_sdlc.run.v1", + provider: "claude_code", + mode: "builder", + result: "succeeded", + usage: { input_tokens: 1, output_tokens: 1, cost_source: "provider" }, + activity: {}, + timing: { started_at: "2026-08-16T00:00:00.000Z" }, + }); + // A receipt sealed before `evidence` existed must still verify: both fields + // are optional and the digest skips undefined, so its payload hash is + // unchanged. + const withoutEvidence = sealFacilityReceipt(base, null); + expect(verifyFacilityReceipt(withoutEvidence)).toBe(true); + expect( + receiptContentDigest({ ...base, evidence: undefined, evidence_truncated: undefined }), + ).toBe(receiptContentDigest(base)); + + const withEvidence = sealFacilityReceipt( + { + ...base, + evidence: [{ name: "transcript", status: "failed", reason: "transcript_upload_failed" }], + }, + null, + ); + expect(verifyFacilityReceipt(withEvidence)).toBe(true); + expect(withEvidence.evidence).toEqual([ + { name: "transcript", status: "failed", reason: "transcript_upload_failed" }, + ]); + // Evidence is covered by the seal, so it cannot be edited after the fact. + expect(withEvidence.integrity?.payload_sha256).not.toBe( + withoutEvidence.integrity?.payload_sha256, + ); + expect( + verifyFacilityReceipt({ + ...withEvidence, + evidence: [{ name: "transcript", status: "passed" }], + }), + ).toBe(false); + + // The truncation flag is sealed too, so a receipt cannot be made to look + // complete after the fact. + const truncated = sealFacilityReceipt({ ...withEvidence, evidence_truncated: true }, null); + expect(verifyFacilityReceipt(truncated)).toBe(true); + expect(truncated.evidence_truncated).toBe(true); + expect(verifyFacilityReceipt({ ...truncated, evidence_truncated: false })).toBe(false); + }); }); diff --git a/runner/src/index.ts b/runner/src/index.ts index 7bfc5a9e..cc198d18 100644 --- a/runner/src/index.ts +++ b/runner/src/index.ts @@ -107,8 +107,7 @@ async function main() { const phases = new RunPhaseRecorder(emit); let bundle: RunBundle | null = null; let steerStop: (() => void) | undefined; - let progressStop: (() => Promise) | undefined; - let checkpointStop: (() => Promise) | undefined; + const polls: RunPolls = {}; let preparedSecuritySweep: PreparedSecuritySweepEvidence | null = null; let restoredSessionState = false; secretsToRedact.add(runnerToken()); @@ -193,10 +192,8 @@ async function main() { if (managedProgress) { await emit([{ type: "agent_progress", data: { markdown: managedProgress } }]); } - progressStop = startAgentProgressPoll(cwdFor(bundle)); - const stopProgress = progressStop; - checkpointStop = startSessionCheckpointPoll(activeBundle); - const stopCheckpoint = checkpointStop; + polls.progress = startAgentProgressPoll(cwdFor(bundle)); + polls.checkpoint = startSessionCheckpointPoll(activeBundle); const engineCode = await phases.measure( "agent", () => runEngine(activeBundle, restoredSessionState), @@ -204,69 +201,15 @@ async function main() { outcome: interruptRequested ? "canceled" : code === 0 ? "succeeded" : "failed", }), ); - const captured = await phases.measure("result_capture", async () => { - const agentProgressPublished = await stopProgress(); - progressStop = undefined; - if (managedProgress) { - await emit([ - { - type: "agent_progress", - data: { markdown: readOnlyEngineProgress(engineCode === 0) }, - }, - ]); - } - const progressPublished = managedProgress !== null || agentProgressPublished; - const securityReport = isSecurityMode(activeBundle.mode) - ? await readSecurityReport( - join(cwdFor(activeBundle), ".agent-sdlc", "security-findings.json"), - ) - : undefined; - const securityReportConfigured = - !isSecurityMode(activeBundle.mode) || securityReport !== null; - const securityEvidenceConfigured = - !isSecurityMode(activeBundle.mode) || - (preparedSecuritySweep !== null && - (await verifySecuritySweepEvidence(preparedSecuritySweep))); - if (isSecurityMode(activeBundle.mode)) { - await emit([ - { - type: "check", - data: { - self_reported: false, - name: "deterministic security evidence", - status: securityEvidenceConfigured ? "passed" : "failed", - ...(securityEvidenceConfigured ? {} : { reason: "security_evidence_invalid" }), - }, - }, - { - type: "check", - data: { - self_reported: false, - name: "structured security findings", - status: securityReportConfigured ? "passed" : "failed", - ...(securityReportConfigured ? {} : { reason: "security_report_invalid" }), - }, - }, - ]); - } - if (engineEventTransportDegraded) { - await emit([ - { - type: "artifact_error", - data: { kind: "engine_events_degraded" }, - }, - ]).catch(() => undefined); - } - await uploadTranscript(); - await stopCheckpoint(); - checkpointStop = undefined; - return { - progressPublished, - securityReport, - securityReportConfigured, - securityEvidenceConfigured, - }; - }); + const captured = await phases.measure("result_capture", () => + captureRunResult({ + bundle: activeBundle, + engineCode, + managedProgress, + preparedSecuritySweep, + polls, + }), + ); const { progressPublished, securityReport, @@ -414,12 +357,113 @@ async function main() { ); process.exitCode = 1; } finally { - await progressStop?.().catch(() => false); - await checkpointStop?.().catch(() => undefined); + await polls.progress?.().catch(() => false); + await polls.checkpoint?.().catch(() => undefined); steerStop?.(); } } +/** + * The polls a run starts before the engine. `captureRunResult` clears each + * handle as it stops it, so `main()`'s `finally` cannot stop one a second time. + */ +export type RunPolls = { + progress?: () => Promise; + checkpoint?: () => Promise; +}; + +export type RunResultCapture = { + progressPublished: boolean; + securityReport: Record | null | undefined; + securityReportConfigured: boolean; + securityEvidenceConfigured: boolean; +}; + +/** + * The run's `result_capture` phase: close the progress poll, record the + * security evidence the mode requires, and preserve the transcript. + * + * Exported because this is the only definition of that sequence. `main()` runs + * exactly this and the integration tests drive exactly this, so a step that + * stops happening for real stops happening under test too. Reassembling these + * calls in a test instead would keep the test green while the phase drifted — + * which is how the transcript evidence came to be emitted apart from the upload + * it describes. + */ +export async function captureRunResult({ + bundle, + engineCode, + managedProgress, + preparedSecuritySweep, + polls, + transcriptPath, +}: { + bundle: RunBundle; + engineCode: number; + managedProgress: string | null; + preparedSecuritySweep: PreparedSecuritySweepEvidence | null; + polls: RunPolls; + transcriptPath?: string; +}): Promise { + const agentProgressPublished = (await polls.progress?.()) ?? false; + polls.progress = undefined; + if (managedProgress) { + await emit([ + { + type: "agent_progress", + data: { markdown: readOnlyEngineProgress(engineCode === 0) }, + }, + ]); + } + const progressPublished = managedProgress !== null || agentProgressPublished; + const securityReport = isSecurityMode(bundle.mode) + ? await readSecurityReport(join(cwdFor(bundle), ".agent-sdlc", "security-findings.json")) + : undefined; + const securityReportConfigured = !isSecurityMode(bundle.mode) || securityReport !== null; + const securityEvidenceConfigured = + !isSecurityMode(bundle.mode) || + (preparedSecuritySweep !== null && (await verifySecuritySweepEvidence(preparedSecuritySweep))); + if (isSecurityMode(bundle.mode)) { + await emit([ + { + type: "check", + data: { + self_reported: false, + name: "deterministic security evidence", + status: securityEvidenceConfigured ? "passed" : "failed", + ...(securityEvidenceConfigured ? {} : { reason: "security_evidence_invalid" }), + }, + }, + { + type: "check", + data: { + self_reported: false, + name: "structured security findings", + status: securityReportConfigured ? "passed" : "failed", + ...(securityReportConfigured ? {} : { reason: "security_report_invalid" }), + }, + }, + ]); + } + if (engineEventTransportDegraded) { + await emit([ + { + type: "artifact_error", + data: { kind: "engine_events_degraded" }, + }, + ]).catch(() => undefined); + } + await uploadTranscript({ transcriptPath }); + await polls.checkpoint?.(); + polls.checkpoint = undefined; + return { + progressPublished, + securityReport, + securityReportConfigured, + securityEvidenceConfigured, + }; +} + export async function prepareWorkspace( bundle: RunBundle, virtualKey: string, @@ -873,8 +917,16 @@ function isSecurityReport(value: unknown): value is Record { }); } -async function uploadTranscript() { - const size = await stat(transcriptFile) +// `transcriptPath` stays injectable because the default is an absolute path +// inside the sandbox runtime root, which tests cannot write to. Nothing here is +// exported: `captureRunResult` is the only entry the tests drive, so the upload +// is exercised as part of the phase that runs it or not at all. +async function uploadTranscript({ + transcriptPath = transcriptFile, +}: { + transcriptPath?: string; +} = {}) { + const size = await stat(transcriptPath) .then((info) => info.size) .catch(() => 0); if (size === 0) return; @@ -886,12 +938,23 @@ async function uploadTranscript() { headers: { "content-type": "application/x-ndjson" }, duplex: "half", } as RequestInit & { duplex: "half" }, - () => createReadStream(transcriptFile) as unknown as RequestInit["body"], + () => createReadStream(transcriptPath) as unknown as RequestInit["body"], ); } catch { - await emit([{ type: "artifact_error", data: { kind: "transcript_upload_failed" } }]).catch( - () => undefined, - ); + // Best-effort, and deliberately one batch: `appendRunEvents` inserts a + // batch inside a single transaction, so the anonymous error and the named + // evidence either both land or neither does. Split across two requests, a + // transient failure on the second leaves the receipt holding the anonymous + // half alone — the loss this evidence exists to name. The whole emit stays + // guarded because this evidence is non-gating: a degraded events endpoint + // must not turn an otherwise successful run into a failed one. + await emit([ + { type: "artifact_error", data: { kind: "transcript_upload_failed" } }, + { + type: "evidence", + data: { name: "transcript", status: "failed", reason: "transcript_upload_failed" }, + }, + ]).catch(() => undefined); } } diff --git a/runner/test/transcript-upload.integration.test.ts b/runner/test/transcript-upload.integration.test.ts new file mode 100644 index 00000000..8f4f8b52 --- /dev/null +++ b/runner/test/transcript-upload.integration.test.ts @@ -0,0 +1,312 @@ +import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import { createServer, type IncomingHttpHeaders } from "node:http"; +import type { AddressInfo } from "node:net"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { captureRunResult, type RunPolls } from "../src/index.js"; +import type { RunBundle, RunEvent } from "../src/types.js"; + +const ENV_KEYS = ["FACILITY_API_URL", "RUN_ID", "RUNNER_TOKEN"] as const; + +const RUN_ID = "run_transcript"; +const RUNNER_TOKEN = "runner-transcript-token"; +const TRANSCRIPT_LINE = '{"type":"assistant","message":"hello"}\n'; + +type RecordedRequest = { + method: string; + path: string; + headers: IncomingHttpHeaders; + body: string; +}; + +let cleanups: Array<() => Promise> = []; +let handlerFailures: unknown[] = []; +let previousEnv: Record<(typeof ENV_KEYS)[number], string | undefined>; +let previousFetch: typeof fetch; + +beforeEach(() => { + cleanups = []; + handlerFailures = []; + previousEnv = Object.fromEntries(ENV_KEYS.map((key) => [key, process.env[key]])) as Record< + (typeof ENV_KEYS)[number], + string | undefined + >; + previousFetch = globalThis.fetch; + globalThis.fetch = async (request, init) => { + const url = new URL( + typeof request === "string" ? request : request instanceof URL ? request.href : request.url, + ); + if (url.protocol !== "http:" || url.hostname !== "127.0.0.1") { + throw new Error(`integration test blocked external request to ${url.origin}`); + } + return previousFetch(request, init); + }; +}); + +afterEach(async () => { + const failures: unknown[] = []; + try { + for (const cleanup of cleanups.reverse()) { + try { + await cleanup(); + } catch (error) { + failures.push(error); + } + } + } finally { + for (const key of ENV_KEYS) { + const value = previousEnv[key]; + if (value === undefined) delete process.env[key]; + else process.env[key] = value; + } + globalThis.fetch = previousFetch; + } + failures.push(...handlerFailures); + if (failures.length > 0) throw new AggregateError(failures, "integration fixture cleanup failed"); +}); + +/** + * Stands in for the platform: the transcript route answers with `transcriptStatus`, + * the events route walks `eventsStatuses` one status per request (the last value + * sticking for any further requests), so a test can make one events request + * succeed and the next fail. Every events request is recorded — rejected ones + * included — so a test can assert how many requests were made, not only what + * the platform ended up storing. + */ +async function startPlatform({ + transcriptStatus = 200, + eventsStatuses = [200], +}: { + transcriptStatus?: number; + eventsStatuses?: number[]; +} = {}) { + const requests: RecordedRequest[] = []; + const eventRequests: RunEvent[][] = []; + const eventBatches: RunEvent[][] = []; + let eventsCall = 0; + const server = createServer(async (request, response) => { + try { + const chunks: Buffer[] = []; + // Drain before replying: the transcript upload streams its body, and + // answering early would surface as a socket error instead of the status. + for await (const chunk of request) chunks.push(Buffer.from(chunk)); + const recorded: RecordedRequest = { + method: request.method ?? "GET", + path: request.url ?? "/", + headers: request.headers, + body: Buffer.concat(chunks).toString("utf8"), + }; + requests.push(recorded); + if (recorded.headers.authorization !== `Bearer ${RUNNER_TOKEN}`) { + response.writeHead(401, { "content-type": "application/json" }); + response.end(JSON.stringify({ message: "invalid runner token" })); + return; + } + if (recorded.method === "POST" && recorded.path === `/internal/runs/${RUN_ID}/transcript`) { + response.writeHead(transcriptStatus, { "content-type": "application/json" }); + response.end(JSON.stringify(transcriptStatus === 200 ? {} : { error: "s3_write_failed" })); + return; + } + if (recorded.method === "POST" && recorded.path === `/internal/runs/${RUN_ID}/events`) { + const batch = JSON.parse(recorded.body) as RunEvent[]; + const status = eventsStatuses[Math.min(eventsCall, eventsStatuses.length - 1)] ?? 200; + eventsCall += 1; + eventRequests.push(batch); + // The API inserts a batch in one transaction, so a rejected request + // stores none of it. + if (status === 200) eventBatches.push(batch); + response.writeHead(status, { "content-type": "application/json" }); + response.end(JSON.stringify(status === 200 ? {} : { message: "events degraded" })); + return; + } + response.writeHead(404, { "content-type": "application/json" }); + response.end(JSON.stringify({ message: `unexpected route ${recorded.path}` })); + } catch (error) { + handlerFailures.push(error); + response.writeHead(500, { "content-type": "application/json" }); + response.end(JSON.stringify({ message: String(error) })); + } + }); + await new Promise((resolve, reject) => { + const onError = (error: Error) => reject(error); + server.once("error", onError); + server.listen(0, "127.0.0.1", () => { + server.off("error", onError); + resolve(); + }); + }); + const { port } = server.address() as AddressInfo; + cleanups.push( + () => + new Promise((resolve, reject) => { + server.closeAllConnections(); + server.close((error) => (error ? reject(error) : resolve())); + }), + ); + process.env.FACILITY_API_URL = `http://127.0.0.1:${port}`; + process.env.RUN_ID = RUN_ID; + process.env.RUNNER_TOKEN = RUNNER_TOKEN; + return { + requests, + /** Every event the platform accepted, flattened across batches. */ + events: () => eventBatches.flat(), + /** Every batch the runner sent to the events route, rejected ones included. */ + eventRequests: () => eventRequests, + transcriptRequests: () => + requests.filter((r) => r.path === `/internal/runs/${RUN_ID}/transcript`), + }; +} + +async function writeTranscript(contents: string) { + const dir = await mkdtemp(join(tmpdir(), "facility-transcript-")); + cleanups.push(() => rm(dir, { recursive: true, force: true })); + const path = join(dir, "engine.stream.jsonl"); + await writeFile(path, contents); + return path; +} + +function bundle(overrides: Partial = {}): RunBundle { + return { + runId: RUN_ID, + mode: "builder", + engine: "codex", + contract: "Do the work.", + skills: [], + engineConfig: {}, + repo: { cloneUrl: null, branch: null, expectedHeadSha: null, installationTokenRef: null }, + harness: null, + packageInstallCmd: null, + provisionCmd: null, + checkCmds: [], + gatewayUrls: { anthropic: "https://anthropic.test", openai: "https://openai.test" }, + scope: {}, + timeoutMin: 5, + ...overrides, + }; +} + +/** + * Runs the run's real `result_capture` phase — the same exported function + * `main()` measures — over a transcript this test controls. Nothing about the + * phase is reassembled here: only its inputs (the polls `main()` started + * before the engine, and the transcript path) are supplied, so a step removed + * from the phase is a step removed from these tests. + */ +async function capture(transcriptPath: string) { + const polls: RunPolls = { + progress: async () => false, + checkpoint: async () => undefined, + }; + const captured = await captureRunResult({ + bundle: bundle(), + engineCode: 0, + managedProgress: null, + preparedSecuritySweep: null, + polls, + transcriptPath, + }); + return { captured, polls }; +} + +describe("result capture phase wiring", () => { + it("keeps the capture phase wired into main()", async () => { + // The tests below drive `captureRunResult` directly, which is the only way + // to reach the phase: `main()` is not exported and no test spawns the + // runner. That leaves one gap they cannot see — a `main()` that stops + // calling the phase at all. Asserting the call site closes it, so deleting + // it turns this suite red instead of silently retiring the transcript + // upload and the evidence it emits. + const source = await readFile(new URL("../src/index.ts", import.meta.url), "utf8"); + + expect(source).toMatch( + /phases\.measure\(\s*"result_capture",\s*\(\)\s*=>\s*captureRunResult\(/, + ); + }); +}); + +describe("result capture transcript evidence", () => { + it("streams the transcript to the platform and records nothing when it lands", async () => { + const platform = await startPlatform(); + const path = await writeTranscript(TRANSCRIPT_LINE); + + const { polls } = await capture(path); + + const [upload] = platform.transcriptRequests(); + expect(upload?.headers["content-type"]).toBe("application/x-ndjson"); + expect(upload?.body).toBe(TRANSCRIPT_LINE); + expect(platform.events()).toEqual([]); + // Both polls stopped and cleared, so main()'s `finally` cannot stop them + // a second time. + expect(polls).toEqual({ progress: undefined, checkpoint: undefined }); + }); + + it("does not upload or record anything when the engine wrote no transcript", async () => { + const platform = await startPlatform({ transcriptStatus: 500 }); + const path = await writeTranscript(""); + + await capture(path); + + expect(platform.transcriptRequests()).toEqual([]); + expect(platform.events()).toEqual([]); + }); + + it("records a rejected upload as evidence that the receipt's check query cannot collect", async () => { + const platform = await startPlatform({ transcriptStatus: 500 }); + const path = await writeTranscript(TRANSCRIPT_LINE); + + await capture(path); + + const events = platform.events(); + expect(events).toContainEqual({ + type: "evidence", + data: { name: "transcript", status: "failed", reason: "transcript_upload_failed" }, + }); + // The receipt collects its check list with `where type = 'check'`, so the + // loss is recorded without ever reaching the gate that list feeds. + expect(events.filter((event) => event.type === "check")).toEqual([]); + expect(events).toContainEqual({ + type: "artifact_error", + data: { kind: "transcript_upload_failed" }, + }); + }); + + it("sends the error and the evidence in one request, so a later failure cannot strand the anonymous half", async () => { + const platform = await startPlatform({ transcriptStatus: 500, eventsStatuses: [200, 503] }); + const path = await writeTranscript(TRANSCRIPT_LINE); + + await capture(path); + + const batch: RunEvent[] = [ + { type: "artifact_error", data: { kind: "transcript_upload_failed" } }, + { + type: "evidence", + data: { name: "transcript", status: "failed", reason: "transcript_upload_failed" }, + }, + ]; + // The outcome that matters: the platform stored both halves even though it + // rejects the second request. Emitted separately, the 200 is spent on the + // anonymous error and the named evidence is lost to the 503, leaving the + // receipt with an anonymous +1 on `activity.errors` and nothing saying the + // transcript is what the run lost. + expect(platform.events()).toEqual(batch); + // One request, so the 503 is never reached at all. + expect(platform.eventRequests()).toEqual([batch]); + }); + + it("survives an events endpoint that is degraded at the same time", async () => { + const platform = await startPlatform({ transcriptStatus: 500, eventsStatuses: [503] }); + const path = await writeTranscript(TRANSCRIPT_LINE); + + // Both writes fail, and the phase must still complete: an unguarded emit + // here would reach main()'s outer catch and fail an otherwise successful + // run over a storage blip. + const { captured, polls } = await capture(path); + + expect(captured.progressPublished).toBe(false); + expect(polls).toEqual({ progress: undefined, checkpoint: undefined }); + // The batch is rejected whole, so the receipt never holds the anonymous + // error without the evidence that names what was lost. + expect(platform.events()).toEqual([]); + }); +}); diff --git a/services/api/src/sandbox/orchestrator.ts b/services/api/src/sandbox/orchestrator.ts index 970dc9f5..5bd8d94e 100644 --- a/services/api/src/sandbox/orchestrator.ts +++ b/services/api/src/sandbox/orchestrator.ts @@ -3039,6 +3039,7 @@ async function gatewayAggregate(db: ReturnType["db"], runId: st select count(*)::int as event_count, count(*) filter (where type = 'check')::int as check_count, + count(*) filter (where type = 'evidence')::int as evidence_count, count(*) filter (where type = 'assistant')::int as turns, count(*) filter (where type = 'tool')::int as tool_calls, count(*) filter ( @@ -3063,10 +3064,20 @@ async function gatewayAggregate(db: ReturnType["db"], runId: st .where(and(eq(runEvents.runId, runId), eq(runEvents.type, "check"))) .orderBy(runEvents.seq) .limit(200); + // Non-gating evidence is deliberately not a check, so the query above cannot + // see it. Collected separately, the receipt names what a run failed to + // preserve instead of folding it into the anonymous `activity.errors` count. + const evidenceEvents = await db + .select({ data: runEvents.data }) + .from(runEvents) + .where(and(eq(runEvents.runId, runId), eq(runEvents.type, "evidence"))) + .orderBy(runEvents.seq) + .limit(200); const eventRow = ( events as unknown as Array<{ event_count: number; check_count: number; + evidence_count: number; turns: number; tool_calls: number; shell_commands: number; @@ -3083,6 +3094,7 @@ async function gatewayAggregate(db: ReturnType["db"], runId: st costCents: Number(usage?.costCents ?? 0), eventCount: Number(eventRow?.event_count ?? 0), checkCount: Number(eventRow?.check_count ?? 0), + evidenceCount: Number(eventRow?.evidence_count ?? 0), activity: { turns: Number(eventRow?.turns ?? 0), shell_commands: Number(eventRow?.shell_commands ?? 0), @@ -3092,6 +3104,7 @@ async function gatewayAggregate(db: ReturnType["db"], runId: st errors: Number(eventRow?.errors ?? 0), }, checks: checkEvents.map(({ data }) => receiptCheck(data)), + evidence: evidenceEvents.map(({ data }) => receiptEvidence(data)), }; } @@ -3165,6 +3178,10 @@ async function canonicalRunReceipt( events: { count: aggregate.eventCount, checks: aggregate.checkCount }, checks: aggregate.checks, checks_truncated: aggregate.checkCount > aggregate.checks.length, + ...(aggregate.evidence.length > 0 ? { evidence: aggregate.evidence } : {}), + // Only when true: a receipt that carries no evidence must digest exactly as + // it did before the field existed. + ...(aggregate.evidenceCount > aggregate.evidence.length ? { evidence_truncated: true } : {}), }); return sealFacilityReceipt(receipt, await previousReceiptDigest(db, run)); } @@ -3333,6 +3350,17 @@ function receiptCheck(value: unknown) { }; } +function receiptEvidence(value: unknown) { + const data = objectOrEmpty(value); + const rawStatus = typeof data.status === "string" ? data.status.trim().toLowerCase() : "unknown"; + const status = ["passed", "failed", "skipped"].includes(rawStatus) ? rawStatus : "unknown"; + return { + name: typeof data.name === "string" && data.name.trim() ? data.name : "unnamed evidence", + status, + ...(typeof data.reason === "string" && data.reason.trim() ? { reason: data.reason } : {}), + }; +} + function normalizeDriver(value: string): SandboxDriverName { if (value === "aws") return "aws"; if (value === "vercel") return "vercel"; diff --git a/services/api/test/sandbox.test.ts b/services/api/test/sandbox.test.ts index 475cf4f6..a70f0966 100644 --- a/services/api/test/sandbox.test.ts +++ b/services/api/test/sandbox.test.ts @@ -1609,6 +1609,128 @@ describe("sandbox api", async () => { await expect(verifyStoredReceipts(db, orgId, [run.id])).resolves.toMatchObject({ ok: true }); }); + it("stores normalised non-gating evidence in the run receipt", async () => { + const token = "frt_receipt_evidence"; + const run = await insertRunnerRun(token, "running"); + await appendRunEvents(db, orgId, run.id, [ + { + type: "evidence", + data: { name: "transcript", status: "failed", reason: "transcript_upload_failed" }, + }, + // Malformed: no name, a status needing trim and case folding, and a blank + // reason that must not reach the receipt as an empty string. + { type: "evidence", data: { status: " SKIPPED ", reason: " " } }, + // An unrecognised status must clamp rather than pass through. + { type: "evidence", data: { name: "session state", status: "exploded" } }, + { + type: "check", + data: { command: "pnpm test", status: "passed", exit_code: 0, self_reported: false }, + }, + { type: "artifact_error", data: { kind: "transcript_upload_failed" } }, + ]); + + const response = await app.inject({ + method: "POST", + url: `/internal/runs/${run.id}/result`, + headers: { authorization: `Bearer ${token}` }, + payload: { status: "succeeded" }, + }); + expect(response.statusCode).toBe(200); + const finished = (await db.select().from(runs).where(eq(runs.id, run.id)).limit(1))[0]; + const receipt = finished?.receipt as { + evidence?: unknown[]; + evidence_truncated?: boolean; + checks?: unknown[]; + checks_truncated?: boolean; + activity?: { errors?: number }; + result?: string; + }; + + expect(receipt.evidence).toEqual([ + { name: "transcript", status: "failed", reason: "transcript_upload_failed" }, + { name: "unnamed evidence", status: "skipped" }, + { name: "session state", status: "unknown" }, + ]); + // Evidence is collected separately from checks and must not leak into the + // gated list the run's outcome is read from. + expect(receipt.checks).toEqual([ + { name: "pnpm test", status: "passed", source: "platform", exit_code: 0 }, + ]); + expect(receipt.checks_truncated).toBe(false); + expect(receipt).not.toHaveProperty("evidence_truncated"); + + // The tally is unmoved by the three evidence events. It has exactly two + // contributions: the one artifact_error the aggregate counted, plus one for + // the run resolving to failed. It resolves to failed because + // insertRunnerRun creates a builder run and finishing one with no git + // changes is delivery_no_changes — asserted here so the count reads as + // deliberate rather than as an evidence event that leaked into it. + expect(receipt.result).toBe("failed"); + expect(finished?.error).toBe("delivery_no_changes"); + expect(receipt.activity?.errors).toBe(2); + expect(verifyFacilityReceipt(receipt as never)).toBe(true); + await expect(verifyStoredReceipts(db, orgId, [run.id])).resolves.toMatchObject({ ok: true }); + }); + + it("omits the evidence field entirely from a run that recorded none", async () => { + const token = "frt_receipt_no_evidence"; + const run = await insertRunnerRun(token, "running"); + await appendRunEvents(db, orgId, run.id, [ + { + type: "check", + data: { command: "pnpm test", status: "passed", exit_code: 0, self_reported: false }, + }, + ]); + + const response = await app.inject({ + method: "POST", + url: `/internal/runs/${run.id}/result`, + headers: { authorization: `Bearer ${token}` }, + payload: { status: "succeeded" }, + }); + expect(response.statusCode).toBe(200); + const finished = (await db.select().from(runs).where(eq(runs.id, run.id)).limit(1))[0]; + const receipt = finished?.receipt as Record; + + // Absent, not an empty array: such a receipt has to digest exactly as it + // did before the field existed. + expect(receipt).not.toHaveProperty("evidence"); + expect(receipt).not.toHaveProperty("evidence_truncated"); + expect(verifyFacilityReceipt(receipt as never)).toBe(true); + }); + + it("discloses when a run receipt truncates its evidence list", async () => { + const token = "frt_receipt_evidence_truncated"; + const run = await insertRunnerRun(token, "running"); + await appendRunEvents( + db, + orgId, + run.id, + Array.from({ length: 201 }, (_, index) => ({ + type: "evidence", + data: { name: `artifact ${index + 1}`, status: "failed" }, + })), + ); + + const response = await app.inject({ + method: "POST", + url: `/internal/runs/${run.id}/result`, + headers: { authorization: `Bearer ${token}` }, + payload: { status: "succeeded" }, + }); + expect(response.statusCode).toBe(200); + const finished = (await db.select().from(runs).where(eq(runs.id, run.id)).limit(1))[0]; + const receipt = finished?.receipt as { + evidence?: unknown[]; + evidence_truncated?: boolean; + }; + + expect(receipt.evidence).toHaveLength(200); + expect(receipt.evidence_truncated).toBe(true); + expect(verifyFacilityReceipt(receipt as never)).toBe(true); + await expect(verifyStoredReceipts(db, orgId, [run.id])).resolves.toMatchObject({ ok: true }); + }); + it("delivers run events over the NOTIFY-backed SSE path without safety polling", async () => { const token = "frt_stream"; const run = await insertRunnerRun(token, "running");