From 5cf56df54952fb16a12c2c8005d7aa5f6b2bc19b Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Sun, 6 Sep 2026 17:38:29 -0400 Subject: [PATCH] =?UTF-8?q?=F0=9F=8F=BA=20fix:=20Report=20Artifact=20Deliv?= =?UTF-8?q?ery=20Failures=20(#119)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix: Surface artifact delivery failures * fix: Await Worker Artifact Settlement in Blocking PTC --- api/src/api/v2.ts | 20 +-- api/src/delivery.test.ts | 73 +++++++++++ api/src/delivery.ts | 43 +++++++ api/src/job.ts | 2 + service/src/execution-log.test.ts | 17 ++- service/src/execution-log.ts | 21 +++- service/src/service/blocking-poll.test.ts | 66 ++++++++++ service/src/service/blocking-poll.ts | 76 ++++++++++++ service/src/service/programmatic-router.ts | 138 ++++----------------- service/src/types/service.ts | 11 ++ service/src/workers.ts | 3 + 11 files changed, 347 insertions(+), 123 deletions(-) create mode 100644 api/src/delivery.test.ts create mode 100644 api/src/delivery.ts create mode 100644 service/src/service/blocking-poll.test.ts create mode 100644 service/src/service/blocking-poll.ts diff --git a/api/src/api/v2.ts b/api/src/api/v2.ts index 40ba2517..a883dc72 100644 --- a/api/src/api/v2.ts +++ b/api/src/api/v2.ts @@ -4,6 +4,7 @@ import type { TFile } from '../job'; import { getLatestRuntimeMatchingLanguageVersion, getRuntimes } from '../runtime'; import { logger } from '../logger'; import { config } from '../config'; +import { reconcileArtifactDelivery } from '../delivery'; import { Job, SessionWorkspaceDirtyError, @@ -563,15 +564,20 @@ router.post('/execute', express.json({ limit: config.execute_body_limit }), asyn return new Set(); }); - const generatedIds = new Set(job.getGeneratedFileIds()); - const before = result.files.length; - result.files = result.files.filter( - f => !generatedIds.has(f.id) || uploaded.has(f.id), + const delivery = reconcileArtifactDelivery( + result.files, + job.getGeneratedFileIds(), + uploaded, ); - const dropped = before - result.files.length; - if (dropped > 0) { + result.files = delivery.files; + result.artifact_delivery = delivery.artifact_delivery; + if (delivery.artifact_delivery) { logger.warn( - { job: job.uuid, dropped, kept: result.files.length }, + { + job: job.uuid, + dropped: delivery.artifact_delivery.failed, + kept: result.files.length, + }, 'Pruned files from response because upload did not reach file_server', ); } diff --git a/api/src/delivery.test.ts b/api/src/delivery.test.ts new file mode 100644 index 00000000..eb488e8c --- /dev/null +++ b/api/src/delivery.test.ts @@ -0,0 +1,73 @@ +import { describe, expect, test } from 'bun:test'; +import { reconcileArtifactDelivery } from './delivery'; + +describe('reconcileArtifactDelivery', () => { + test('leaves successful and inherited file references unchanged', () => { + const files = [ + { id: 'generated', name: 'result.txt' }, + { id: 'inherited', name: 'input.txt', inherited: true as const }, + ]; + + expect( + reconcileArtifactDelivery( + files, + ['generated'], + new Set(['generated']), + ), + ).toEqual({ files }); + }); + + test('reports a complete delivery failure without returning phantom references', () => { + const files = [ + { id: 'generated', name: 'result.txt' }, + { id: 'inherited', name: 'input.txt', inherited: true as const }, + ]; + + expect( + reconcileArtifactDelivery(files, ['generated'], new Set()), + ).toEqual({ + files: [{ id: 'inherited', name: 'input.txt', inherited: true }], + artifact_delivery: { + code: 'artifact_delivery_failed', + status: 'failed', + attempted: 1, + delivered: 0, + failed: 1, + }, + }); + }); + + test('reports partial delivery and counts only expected generated ids', () => { + const files = [ + { id: 'first', name: 'first.txt' }, + { id: 'second', name: 'second.txt' }, + ]; + + expect( + reconcileArtifactDelivery( + files, + ['first', 'second'], + new Set(['first', 'unknown']), + ), + ).toEqual({ + files: [{ id: 'first', name: 'first.txt' }], + artifact_delivery: { + code: 'artifact_delivery_failed', + status: 'partial', + attempted: 2, + delivered: 1, + failed: 1, + }, + }); + }); + + test('does not report a failure when there were no generated files', () => { + const files = [ + { id: 'inherited', name: 'input.txt', inherited: true as const }, + ]; + + expect(reconcileArtifactDelivery(files, [], new Set())).toEqual({ + files, + }); + }); +}); diff --git a/api/src/delivery.ts b/api/src/delivery.ts new file mode 100644 index 00000000..ff159d6f --- /dev/null +++ b/api/src/delivery.ts @@ -0,0 +1,43 @@ +export interface ArtifactDeliveryFailure { + code: 'artifact_delivery_failed'; + status: 'partial' | 'failed'; + attempted: number; + delivered: number; + failed: number; +} + +export interface ArtifactDeliveryResult { + files: T[]; + artifact_delivery?: ArtifactDeliveryFailure; +} + +/** Removes unusable generated refs while preserving an explicit delivery failure for callers. */ +export function reconcileArtifactDelivery( + files: T[], + generatedFileIds: Iterable, + uploadedFileIds: ReadonlySet, +): ArtifactDeliveryResult { + const generatedIds = new Set(generatedFileIds); + if (generatedIds.size === 0) return { files }; + + let delivered = 0; + const retained = files.filter(file => { + if (!generatedIds.has(file.id)) return true; + if (!uploadedFileIds.has(file.id)) return false; + delivered++; + return true; + }); + const failed = generatedIds.size - delivered; + if (failed === 0) return { files: retained }; + + return { + files: retained, + artifact_delivery: { + code: 'artifact_delivery_failed', + status: delivered === 0 ? 'failed' : 'partial', + attempted: generatedIds.size, + delivered, + failed, + }, + }; +} diff --git a/api/src/job.ts b/api/src/job.ts index 1e78ae75..610b748b 100644 --- a/api/src/job.ts +++ b/api/src/job.ts @@ -7,6 +7,7 @@ import * as fsp from 'fs/promises'; import { pipeline } from 'stream/promises'; import { Readable, Transform } from 'stream'; import type { Logger } from 'pino'; +import type { ArtifactDeliveryFailure } from './delivery'; import type { NsJailResult } from './nsjail'; import type { Runtime } from './runtime'; import { logger as rootLogger } from './logger'; @@ -676,6 +677,7 @@ interface ExecuteResult { /** Top-level execution session id (one sandbox `/exec` invocation). */ session_id: string; files: FileRef[]; + artifact_delivery?: ArtifactDeliveryFailure; } const jobQueue: Array<() => void> = []; diff --git a/service/src/execution-log.test.ts b/service/src/execution-log.test.ts index d78fafc4..f04bad90 100644 --- a/service/src/execution-log.test.ts +++ b/service/src/execution-log.test.ts @@ -20,6 +20,14 @@ describe('execution log summaries', () => { { id: 'file_1', name: 'a.txt', inherited: true }, { id: 'file_2', name: 'b.txt', modified_from: { id: 'file_1', storage_session_id: 'sess_old' } }, ], + artifact_delivery: { + code: 'artifact_delivery_failed', + status: 'partial', + attempted: 3, + delivered: 2, + failed: 1, + detail: 'private storage failure', + }, run: { code: 0, stdout: 'top secret stdout', @@ -33,9 +41,17 @@ describe('execution log summaries', () => { expect(JSON.stringify(summary)).not.toContain('top secret stdout'); expect(JSON.stringify(summary)).not.toContain('sensitive stderr'); expect(JSON.stringify(summary)).not.toContain('combined output'); + expect(JSON.stringify(summary)).not.toContain('private storage failure'); expect(summary).toMatchObject({ session_id: 'sess_123', files: { count: 2, inheritedCount: 1, modifiedCount: 1 }, + artifact_delivery: { + code: 'artifact_delivery_failed', + status: 'partial', + attempted: 3, + delivered: 2, + failed: 1, + }, run: { stdout: { length: 17, present: true }, stderr: { length: 16, present: true }, @@ -56,4 +72,3 @@ describe('execution log summaries', () => { expect(summary).toEqual({ count: 3, skillCount: 1, agentCount: 1, userCount: 1 }); }); }); - diff --git a/service/src/execution-log.ts b/service/src/execution-log.ts index 311169e8..48a93cce 100644 --- a/service/src/execution-log.ts +++ b/service/src/execution-log.ts @@ -18,9 +18,28 @@ type SandboxResponseLike = { language?: unknown; version?: unknown; files?: unknown; + artifact_delivery?: unknown; run?: RunLike; }; +function summarizeArtifactDelivery(value: unknown): Record | undefined { + if (value == null || typeof value !== 'object' || Array.isArray(value)) return undefined; + const delivery = value as { + code?: unknown; + status?: unknown; + attempted?: unknown; + delivered?: unknown; + failed?: unknown; + }; + return { + code: delivery.code, + status: delivery.status, + attempted: delivery.attempted, + delivered: delivery.delivered, + failed: delivery.failed, + }; +} + export function summarizeText(value: unknown): { length: number; present: boolean } { if (typeof value !== 'string') { return { length: 0, present: false }; @@ -67,6 +86,7 @@ export function summarizeSandboxResponse(data: SandboxResponseLike): Record ({ jobCompleted: tick >= 2 }), + getBlockingResult: async () => result, + getPending: async () => ({ status: 'completed' }), + isNotFound: (error: unknown) => error === 'missing', + sleep: async (): Promise => { tick++; }, + now: () => tick, + }; +} + +describe('blocking worker settlement', () => { + test('completed callback waits for delayed upload reconciliation', async () => { + const deps = fixture(); + expect(await pollBlockingExecution('exec', 5, deps)).toEqual({ + status: 'completed', stdout: result.stdout, stderr: '', files: [], + artifact_delivery: result.artifact_delivery, + }); + expect(deps.now()).toBe(2); + }); + + test('missing callback session still receives the worker result', async () => { + const deps = fixture(); + deps.getPending = async (): Promise => { throw 'missing'; }; + expect((await pollBlockingExecution('exec', 5, deps)).artifact_delivery).toEqual(result.artifact_delivery); + }); + + test('never reports success if uploads remain unsettled at timeout', async () => { + expect(await pollBlockingExecution('exec', 1, fixture())).toEqual({ status: 'error' }); + }); + + test('accepts the legacy inline worker result during rolling deployments', async () => { + const deps = fixture(); + expect(await pollBlockingExecution('exec', 5, { + ...deps, + getExecutionState: async () => ({ jobCompleted: true, jobResult: result }), + getBlockingResult: async () => null, + })).toMatchObject({ status: 'completed', artifact_delivery: result.artifact_delivery }); + }); + + test('preserves waiting calls and worker failures', async () => { + expect(await pollBlockingExecution('exec', 5, { + ...fixture(), + getPending: async () => ({ status: 'waiting', pending_calls: [ + { call_id: 'call', tool_name: 'search', tool_input: { query: 'hello' } }, + ] }), + })).toEqual({ status: 'waiting', pending_calls: [ + { id: 'call', name: 'search', input: { query: 'hello' } }, + ] }); + expect(await pollBlockingExecution('exec', 5, { + ...fixture(), getExecutionState: async () => ({ jobError: 'worker failed' }), + })).toEqual({ status: 'error' }); + }); +}); diff --git a/service/src/service/blocking-poll.ts b/service/src/service/blocking-poll.ts new file mode 100644 index 00000000..8ce07fa8 --- /dev/null +++ b/service/src/service/blocking-poll.ts @@ -0,0 +1,76 @@ +import type * as t from '../types'; + +export interface BlockingPendingState { + status: string; + pending_calls?: Array<{ + call_id: string; + tool_name: string; + tool_input: Record; + }>; +} + +export interface BlockingPollDependencies { + getExecutionState(id: string): Promise<{ + jobCompleted?: boolean; + jobResult?: t.ExecuteResult; + jobError?: string; + } | null>; + getBlockingResult(id: string): Promise; + getPending(id: string): Promise; + isNotFound(error: unknown): boolean; + sleep(): Promise; + now(): number; +} + +/** Tool-call completion precedes upload reconciliation. Only a worker result is final. */ +export async function pollBlockingExecution( + id: string, + timeout: number, + deps: BlockingPollDependencies, +): Promise<{ + status: 'waiting' | 'completed' | 'error'; + pending_calls?: t.ProgrammaticToolCall[]; + stdout?: string; + stderr?: string; + files?: t.FileRefs; + artifact_delivery?: t.ArtifactDeliveryFailure; +}> { + const start = deps.now(); + while (deps.now() - start < timeout) { + const execution = await deps.getExecutionState(id); + if (execution?.jobCompleted === true) { + // Preserve the inline result fallback for in-flight jobs from older binaries. + const result = (await deps.getBlockingResult(id)) ?? execution.jobResult; + if (result) { + return { + status: 'completed', + stdout: result.stdout, + stderr: result.stderr, + files: result.files, + artifact_delivery: result.artifact_delivery, + }; + } + } + if (execution?.jobError != null) return { status: 'error' }; + + try { + const pending = await deps.getPending(id); + if (pending.status === 'waiting' && pending.pending_calls != null && pending.pending_calls.length > 0) { + return { + status: 'waiting', + pending_calls: pending.pending_calls.map(call => ({ + id: call.call_id, + name: call.tool_name, + input: call.tool_input, + })), + }; + } + if (pending.status === 'error') return { status: 'error' }; + // Both completed and missing callback sessions must await the worker result. + } catch (error) { + if (!deps.isNotFound(error)) throw error; + } + await deps.sleep(); + } + return { status: 'error' }; +} diff --git a/service/src/service/programmatic-router.ts b/service/src/service/programmatic-router.ts index 4536270b..9896518c 100644 --- a/service/src/service/programmatic-router.ts +++ b/service/src/service/programmatic-router.ts @@ -40,6 +40,7 @@ import { } from '../sandbox-egress'; import { findUnregisteredToolCall } from '../tool-scope'; import { summarizeRequestedFiles } from '../execution-log'; +import { pollBlockingExecution, type BlockingPendingState } from './blocking-poll'; import { clearSessionOwnership, recordSessionOwnership } from '../session-ownership'; import { FileRefAuthorizationError, authorizeRequestedFiles } from './file-authorization'; import { @@ -225,118 +226,24 @@ function decodeContinuationToken(token: string): { execution_id: string } | null // Blocking mode (legacy path) // --------------------------------------------------------------------------- -async function waitForExecutionState( - execution_id: string, - timeout: number, -): Promise<{ - status: 'waiting' | 'completed' | 'error' | 'running'; - pending_calls?: t.ProgrammaticToolCall[]; - stdout?: string; - stderr?: string; - files?: t.FileRefs; -}> { - const startTime = Date.now(); - - while (Date.now() - startTime < timeout) { - const execution = await getExecutionState(execution_id); - - /** Result lives in the `exec_result:` key (see setBlockingResult). The - * inline `execution.jobResult` branch is kept as a fallback so any - * in-flight executions whose state was written by an older binary - * mid-deploy still complete correctly without rolling back. */ - if (execution?.jobCompleted === true) { - const result = (await getBlockingResult(execution_id)) ?? execution.jobResult; - if (result) { - return { - status: 'completed', - stdout: result.stdout, - stderr: result.stderr, - files: result.files, - }; - } - } - - if (execution?.jobError != null) { - return { status: 'error' }; - } - - try { - const pendingResponse = await retryToolCallServerRequest( - () => axios.get<{ - status: string; - pending_calls?: Array<{ - call_id: string; - tool_name: string; - tool_input: Record; - timestamp: number; - }>; - }>(`${env.TOOL_CALL_SERVER_URL}/sessions/${execution_id}/pending`, { - headers: internalServiceHeaders(), - }), +function waitForExecutionState(execution_id: string, timeout: number): ReturnType { + return pollBlockingExecution(execution_id, timeout, { + getExecutionState, + getBlockingResult, + getPending: async (id) => { + const response = await retryToolCallServerRequest( + () => axios.get( + `${env.TOOL_CALL_SERVER_URL}/sessions/${id}/pending`, + { headers: internalServiceHeaders() }, + ), 'Get pending tool calls', ); - - const { status, pending_calls } = pendingResponse.data; - - if (status === 'waiting' && pending_calls && pending_calls.length > 0) { - return { - status: 'waiting', - pending_calls: pending_calls.map(call => ({ - id: call.call_id, - name: call.tool_name, - input: call.tool_input, - })), - }; - } - - if (status === 'completed') { - const statusResponse = await retryToolCallServerRequest( - () => axios.get<{ - status: string; - stdout?: string; - stderr?: string; - files?: t.FileRefs; - }>(`${env.TOOL_CALL_SERVER_URL}/sessions/${execution_id}/status`, { - headers: internalServiceHeaders(), - }), - 'Get execution status', - ); - - return { - status: 'completed', - stdout: statusResponse.data.stdout, - stderr: statusResponse.data.stderr, - files: statusResponse.data.files, - }; - } - - if (status === 'error') { - return { status: 'error' }; - } - - await new Promise(resolve => setTimeout(resolve, POLL_INTERVAL)); - } catch (error) { - if (axios.isAxiosError(error) && error.response?.status === 404) { - const exec = await getExecutionState(execution_id); - if (exec?.jobCompleted === true) { - const result = (await getBlockingResult(execution_id)) ?? exec.jobResult; - if (result) { - return { - status: 'completed', - stdout: result.stdout, - stderr: result.stderr, - files: result.files, - }; - } - } - await new Promise(resolve => setTimeout(resolve, POLL_INTERVAL)); - continue; - } - throw error; - } - } - - return { status: 'error' }; + return response.data; + }, + isNotFound: (error) => axios.isAxiosError(error) && error.response?.status === 404, + sleep: () => new Promise(resolve => setTimeout(resolve, POLL_INTERVAL)), + now: Date.now, + }); } // --------------------------------------------------------------------------- @@ -403,10 +310,10 @@ async function runReplayIteration( (state.bridgeWorkerId != null ? 'remote-bridge' : resolveQueuedSandboxBackend( - env.EXECUTION_PROFILE, - env.SANDBOX_BACKEND, - env.EXECUTION_PROFILE_SOURCE, - )); + env.EXECUTION_PROFILE, + env.SANDBOX_BACKEND, + env.EXECUTION_PROFILE_SOURCE, + )); const { queue, events, language } = getExecutionQueueBinding( state.language ?? 'python', replayBackend, @@ -1025,6 +932,7 @@ async function runAndRespond( stdout: cleanStdout, stderr: result.stderr, files: result.files, + artifact_delivery: result.artifact_delivery, session_id: state.session_id, }); } @@ -1243,6 +1151,7 @@ async function handleBlocking( stdout: state.stdout ?? '', stderr: state.stderr ?? '', files: state.files ?? [], + artifact_delivery: state.artifact_delivery, session_id: execution.session_id, }); } @@ -1495,6 +1404,7 @@ async function handleBlocking( stdout: state.stdout ?? '', stderr: state.stderr ?? '', files: state.files ?? [], + artifact_delivery: state.artifact_delivery, session_id, }); } diff --git a/service/src/types/service.ts b/service/src/types/service.ts index 555056d5..0f97a532 100644 --- a/service/src/types/service.ts +++ b/service/src/types/service.ts @@ -103,6 +103,14 @@ export type RequestFile = { export type FileRefs = FileRef[]; +export interface ArtifactDeliveryFailure { + code: 'artifact_delivery_failed'; + status: 'partial' | 'failed'; + attempted: number; + delivered: number; + failed: number; +} + export type ExecuteResponse = { run?: { stdout: string; @@ -121,6 +129,7 @@ export type ExecuteResponse = { /** Top-level execution session id (one sandbox `/exec` invocation). */ session_id: string; files: FileRefs; + artifact_delivery?: ArtifactDeliveryFailure; }; export interface RequestBody { @@ -221,6 +230,7 @@ export type ExecuteResult = { stdout: string; stderr: string; files: FileRefs; + artifact_delivery?: ArtifactDeliveryFailure; code?: number | null; signal?: string | null; message?: string | null; @@ -358,6 +368,7 @@ export interface ProgrammaticResponse { stdout?: string; stderr?: string; files?: FileRefs; + artifact_delivery?: ArtifactDeliveryFailure; /** Top-level execution session id (one sandbox PTC invocation). */ session_id?: string; tool_calls_made?: number; diff --git a/service/src/workers.ts b/service/src/workers.ts index 215a4d1b..ad20fd0f 100644 --- a/service/src/workers.ts +++ b/service/src/workers.ts @@ -179,6 +179,9 @@ async function processJobInner(job: t.ExecuteJob): Promise { * `[]` so the strictened response type from Phase B doesn't * surface a regression that wasn't there before. */ files: files ?? [], + ...(responseData.artifact_delivery != null + ? { artifact_delivery: responseData.artifact_delivery } + : {}), stdout, stderr, };