diff --git a/docs/superpowers/specs/2026-07-22-cloud-agent-e2e-health-design.md b/docs/superpowers/specs/2026-07-22-cloud-agent-e2e-health-design.md new file mode 100644 index 0000000000..71fae61952 --- /dev/null +++ b/docs/superpowers/specs/2026-07-22-cloud-agent-e2e-health-design.md @@ -0,0 +1,87 @@ +# Cloud Agent E2E Health Reliability Design + +## Goal + +Make the local fake-LLM smoke matrix a reliable health check for the real Cloud Agent path: Worker, Durable Object, sandbox container, wrapper, Kilo, and persisted run reporting. The harness must distinguish genuine product regressions from expected warm-session preparation chatter and from local container-engine cleanup races. + +## Scope + +This change is limited to the local E2E harness under `services/cloud-agent-next/test/e2e` and its focused unit tests and documentation. It does not change Worker, wrapper, or UI behavior. It does not add retries. + +## Warm-Reuse Health Contract + +The `hot` and `cold-hot` scenarios must continue to prove all of the following: + +- the follow-up reaches its expected terminal state; +- the first Kilo event arrives within the existing observation window; +- the original primary sandbox remains present; +- no new sandbox container appears; and +- no live preparation step that represents cold workspace work starts. + +Preparation snapshots replayed when a stream connects are historical state, not evidence of new work. Live `attempt_started`, `attempt_completed`, progress, and completion frames are also not sufficient by themselves to prove reprovisioning. The harness will therefore classify only version 2 `preparing` frames with `action: "step_started"`. + +The following steps are expected during a warm verification and do not indicate reprovisioning: + +- `sandbox_provision` +- `sandbox_boot` +- `kilo_server` + +Any other live `step_started` preparation step is cold-path work and fails warm reuse. This deliberately fails closed for newly introduced step names: a new step must be explicitly classified before the health check accepts it. Legacy unversioned `preparing` frames also fail warm reuse because the harness cannot prove that they are harmless snapshots or warm verification. + +Failure output will report the unexpected preparation step names instead of only printing `noPrepare=false`. + +## Matrix Isolation Contract + +After every matrix row, the harness will: + +1. identify sandbox families created after the matrix baseline; +2. kill their currently running primary/proxy containers using the existing exact-family matching; +3. wait for each selected family to disappear; and +4. poll all non-baseline sandbox containers until none exist continuously for five seconds. + +The continuous absence window covers the observed local container-engine `destroy -> create -> destroy` churn after an external kill. If a new non-baseline sandbox appears during the window, the stability timer resets. The total quiescence wait is bounded. + +If cleanup does not quiesce, the matrix records an explicit cleanup failure and stops before running another scenario in contaminated state. It does not silently warn and continue, sleep for a fixed interval, or retry a failed product scenario. + +Baseline containers are never killed or treated as cleanup failures. + +## Components + +### Preparation classification + +A small pure helper in `lifecycle.ts` will return the unexpected cold preparation steps observed in a stream event list. Both `hot` and `cold-hot` will use it. + +### Sandbox quiescence + +`sandbox-control.ts` will own a bounded stable-absence polling helper because it already owns Docker discovery and family operations. Its Docker executor and timing inputs will be injectable for deterministic unit tests while production callers use defaults. + +`smoke.ts` will use the helper after its existing family cleanup and convert failure into an explicit matrix result before stopping. + +### Documentation + +The E2E README will describe the warm-reuse and between-row isolation assertions so operators know what a failure means. + +## Tests + +Focused unit tests will cover: + +- snapshots and warm verification steps do not count as cold preparation; +- cold-path and legacy preparation events do count; +- the stable-absence window succeeds only after uninterrupted absence; +- a transient recreation resets the stability window; +- persistent non-baseline containers time out; and +- baseline containers are ignored. + +Focused verification will run the new unit tests, service typecheck, formatting checks, and `git diff --check`. + +Runtime verification will run the complete 15-row matrix against the current local stack with `WORKER_URL=http://localhost:10594`. Completion requires 15/15 scenario results and no matrix-cleanup failure. + +## Deferred Work + +The following remain separate follow-ups: + +- suppressing no-op preparation attempts in production to remove the warm UI flash; +- retrying matrix scenarios; +- removing stopped Docker container corpses; +- adding provisioning elapsed-time landmarks; and +- cleaning pre-existing proxy containers. diff --git a/packages/worker-utils/src/cloud-agent-queue-report.ts b/packages/worker-utils/src/cloud-agent-queue-report.ts index 81595ca45a..8a2b0623ca 100644 --- a/packages/worker-utils/src/cloud-agent-queue-report.ts +++ b/packages/worker-utils/src/cloud-agent-queue-report.ts @@ -35,6 +35,8 @@ export const CloudAgentRunFailureClassifications = [ { failureStage: 'agent_activity', failureCode: 'payment_required' }, { failureStage: 'agent_activity', failureCode: 'model_missing' }, { failureStage: 'agent_activity', failureCode: 'wrapper_error_after_activity' }, + { failureStage: 'agent_activity', failureCode: 'payment_required' }, + { failureStage: 'agent_activity', failureCode: 'model_missing' }, { failureStage: 'interruption', failureCode: 'user_interrupt' }, { failureStage: 'interruption', failureCode: 'container_shutdown' }, { failureStage: 'interruption', failureCode: 'system_interrupt' }, diff --git a/services/cloud-agent-next/src/agent-sandbox/cloudflare/cloudflare-agent-sandbox.test.ts b/services/cloud-agent-next/src/agent-sandbox/cloudflare/cloudflare-agent-sandbox.test.ts index 19a11f7f3a..12e7be8e10 100644 --- a/services/cloud-agent-next/src/agent-sandbox/cloudflare/cloudflare-agent-sandbox.test.ts +++ b/services/cloud-agent-next/src/agent-sandbox/cloudflare/cloudflare-agent-sandbox.test.ts @@ -1666,16 +1666,20 @@ describe('CloudflareAgentSandbox', () => { it('deletes session resources without destroying a shared sandbox', async () => { const destroy = vi.fn(); const deleteSession = vi.fn().mockResolvedValue(undefined); - const sandbox = new CloudflareAgentSandbox({} as Env, metadata(), { - resolveSandbox: () => - ({ - getSession: vi.fn().mockResolvedValue({ - exec: vi.fn().mockResolvedValue({ exitCode: 0, stdout: '', stderr: '' }), - }), - deleteSession, - destroy, - }) as unknown as SandboxInstance, - }); + const sandbox = new CloudflareAgentSandbox( + {} as Env, + metadata({ sandboxId: `usr-${'a'.repeat(48)}` as TestSandboxId }), + { + resolveSandbox: () => + ({ + getSession: vi.fn().mockResolvedValue({ + exec: vi.fn().mockResolvedValue({ exitCode: 0, stdout: '', stderr: '' }), + }), + deleteSession, + destroy, + }) as unknown as SandboxInstance, + } + ); await sandbox.delete('explicit'); @@ -1683,6 +1687,19 @@ describe('CloudflareAgentSandbox', () => { expect(destroy).not.toHaveBeenCalled(); }); + it('destroys an isolated sandbox when deleting its only session', async () => { + const destroy = vi.fn().mockResolvedValue(undefined); + const deleteSession = vi.fn(); + const sandbox = new CloudflareAgentSandbox({} as Env, metadata(), { + resolveSandbox: () => ({ destroy, deleteSession }) as unknown as SandboxInstance, + }); + + await sandbox.delete('explicit'); + + expect(destroy).toHaveBeenCalledOnce(); + expect(deleteSession).not.toHaveBeenCalled(); + }); + it('maps infrastructure recovery to destructive Cloudflare sandbox replacement', async () => { const destroy = vi.fn().mockResolvedValue(undefined); const sandbox = new CloudflareAgentSandbox({} as Env, metadata(), { diff --git a/services/cloud-agent-next/src/agent-sandbox/cloudflare/cloudflare-agent-sandbox.ts b/services/cloud-agent-next/src/agent-sandbox/cloudflare/cloudflare-agent-sandbox.ts index 9f8b5322ac..f6527820ba 100644 --- a/services/cloud-agent-next/src/agent-sandbox/cloudflare/cloudflare-agent-sandbox.ts +++ b/services/cloud-agent-next/src/agent-sandbox/cloudflare/cloudflare-agent-sandbox.ts @@ -24,6 +24,7 @@ import { SANDBOX_SLEEP_AFTER_SECONDS } from '../../core/lease.js'; import { generateSandboxId, getSandboxNamespace, + isGeneratedSharedSandboxId, isOrgInList, MANAGED_SCM_OUTBOUND_HANDLER, } from '../../sandbox-id.js'; @@ -845,8 +846,9 @@ export class CloudflareAgentSandbox implements AgentSandbox { } async delete(reason: SandboxDeleteReason): Promise { - const sandbox = await this.getSandbox(); - if (reason === 'recovery') { + const sandboxId = await this.resolveSandboxId(); + const sandbox = this.resolveSandbox(sandboxId); + if (reason === 'recovery' || !isGeneratedSharedSandboxId(sandboxId)) { await sandbox.destroy(); return; } diff --git a/services/cloud-agent-next/src/kilo/wrapper-client.test.ts b/services/cloud-agent-next/src/kilo/wrapper-client.test.ts index 77f03d082d..d40b59d792 100644 --- a/services/cloud-agent-next/src/kilo/wrapper-client.test.ts +++ b/services/cloud-agent-next/src/kilo/wrapper-client.test.ts @@ -360,7 +360,13 @@ describe('WrapperClient', () => { ); expect(result.kiloSessionId).toBe('kilo_sess_1'); - expect(transport.request).toHaveBeenNthCalledWith(1, 'POST', '/session/ready', readyRequest); + expect(transport.request).toHaveBeenNthCalledWith( + 1, + 'POST', + '/session/ready', + readyRequest, + expect.any(String) + ); expect(transport.request).toHaveBeenNthCalledWith( 2, 'POST', @@ -381,7 +387,8 @@ describe('WrapperClient', () => { condenseOnComplete: false, }, session: binding, - }) + }), + expect.any(String) ); expect('executeSession' in client).toBe(false); expect(session.exec).not.toHaveBeenCalled(); @@ -2282,6 +2289,91 @@ describe('WrapperClient', () => { await expect(client.health()).rejects.toThrow(WrapperError); }); + it('retains the actual curl status and content type for malformed health diagnostics', async () => { + const session = createMockSession(command => { + const marker = command.match(/__KILO_WRAPPER_RESPONSE_[A-Za-z0-9-]+__/)?.[0]; + if (!marker) throw new Error('Missing curl response metadata marker'); + return { + exitCode: 0, + stdout: `private-wrapper-error\n${marker}500\ttext/html`, + }; + }); + const withFields = vi.spyOn(logger, 'withFields').mockReturnValue(logger); + const logError = vi.spyOn(logger, 'error').mockImplementation(() => logger); + const client = new WrapperClient({ session, port: defaultPort }); + + await expect(client.health()).rejects.toMatchObject({ code: 'PARSE_ERROR' }); + + expect(withFields).toHaveBeenCalledWith( + expect.objectContaining({ + path: '/health', + statusCode: 500, + contentType: 'text/html', + }) + ); + expect(JSON.stringify(withFields.mock.calls)).not.toContain('private-wrapper-error'); + withFields.mockRestore(); + logError.mockRestore(); + }); + + it('fingerprints malformed health responses without retaining their body', async () => { + const body = 'token=secret-health-body'; + const transport: WrapperTransport = { + request: vi.fn().mockResolvedValue( + new Response(body, { + status: 500, + headers: { 'Content-Type': 'text/html; charset=utf-8' }, + }) + ), + }; + const withFields = vi.spyOn(logger, 'withFields').mockReturnValue(logger); + const logError = vi.spyOn(logger, 'error').mockImplementation(() => logger); + const client = new WrapperClient({ + session: createMockSession(createSuccessResponse({})), + port: defaultPort, + transport, + }); + + const expectedHash = Array.from( + new Uint8Array(await crypto.subtle.digest('SHA-256', new TextEncoder().encode(body))), + byte => byte.toString(16).padStart(2, '0') + ).join(''); + let thrown: unknown; + try { + await client.health(); + } catch (error) { + thrown = error; + } + + expect(thrown).toMatchObject({ + message: 'Failed to parse wrapper response', + code: 'PARSE_ERROR', + }); + expect(String(thrown)).not.toContain(body); + expect(withFields).toHaveBeenCalledWith( + expect.objectContaining({ + method: 'GET', + path: '/health', + requestId: expect.any(String), + contentType: 'text/html; charset=utf-8', + responseBytes: new TextEncoder().encode(body).byteLength, + responseSha256: expectedHash, + statusCode: 500, + }) + ); + expect(JSON.stringify(withFields.mock.calls)).not.toContain('secret-health-body'); + expect(logError).toHaveBeenCalledWith('Failed to parse wrapper HTTP response'); + expect(transport.request).toHaveBeenCalledWith( + 'GET', + '/health', + undefined, + expect.any(String) + ); + + withFields.mockRestore(); + logError.mockRestore(); + }); + it('handles curl exit codes', async () => { const session = createMockSession(createCurlError(28, 'Operation timed out')); const client = new WrapperClient({ session, port: defaultPort }); diff --git a/services/cloud-agent-next/src/kilo/wrapper-client.ts b/services/cloud-agent-next/src/kilo/wrapper-client.ts index ad8356abf9..b2d59178a7 100644 --- a/services/cloud-agent-next/src/kilo/wrapper-client.ts +++ b/services/cloud-agent-next/src/kilo/wrapper-client.ts @@ -34,6 +34,7 @@ import { } from '../shared/wrapper-bootstrap.js'; import { TOOL_CGROUP_ENV_KEYS, type ToolCgroupEnv } from '../shared/tool-cgroup-env.js'; import { parseWrapperSessionReadyErrorResponse } from './wrapper-ready-error.js'; +import { WRAPPER_REQUEST_ID_HEADER } from '../shared/wrapper-http.js'; // --------------------------------------------------------------------------- // Types @@ -169,7 +170,12 @@ export type WrapperContainerClientOptions = { }; export type WrapperTransport = { - request(method: 'GET' | 'POST', path: string, body?: unknown): Promise; + request( + method: 'GET' | 'POST', + path: string, + body?: unknown, + requestId?: string + ): Promise; }; // --------------------------------------------------------------------------- @@ -253,6 +259,11 @@ const ERROR_STATUS_CODES: Record = { const MAX_PORT_ATTEMPTS = 3; const TOOL_CGROUP_ENV_KEY_SET = new Set(TOOL_CGROUP_ENV_KEYS); +async function sha256Hex(bytes: Uint8Array): Promise { + const digest = await crypto.subtle.digest('SHA-256', bytes); + return Array.from(new Uint8Array(digest), byte => byte.toString(16).padStart(2, '0')).join(''); +} + function healthMatchesLease( health: WrapperHealthResponse, leasedInstance: WrapperInstanceLease | undefined, @@ -324,14 +335,25 @@ class ExecCurlWrapperTransport implements WrapperTransport { this.shellQuote = options.shellQuote; } - async request(method: 'GET' | 'POST', path: string, body?: unknown): Promise { + async request( + method: 'GET' | 'POST', + path: string, + body?: unknown, + requestId?: string + ): Promise { const url = `${this.baseUrl}${path}`; let command = `curl -s -X ${method} -H 'Content-Type: application/json'`; + const responseMetadataMarker = `__KILO_WRAPPER_RESPONSE_${requestId ?? crypto.randomUUID()}__`; + + if (requestId) { + command += ` -H ${this.shellQuote(`${WRAPPER_REQUEST_ID_HEADER}: ${requestId}`)}`; + } if (body) { command += ` -d ${this.shellQuote(JSON.stringify(body))}`; } + command += ` -w ${this.shellQuote(`\n${responseMetadataMarker}%{http_code}\t%{content_type}`)}`; command += ` ${this.shellQuote(url)}`; const result = await this.session.exec(command); @@ -340,9 +362,27 @@ class ExecCurlWrapperTransport implements WrapperTransport { throw new WrapperError(`Request failed: ${stderr || 'curl error'}`, 'REQUEST_FAILED', 500); } - return new Response(result.stdout ?? '', { - status: 200, - headers: { 'Content-Type': 'application/json' }, + const stdout = result.stdout ?? ''; + const markerIndex = stdout.lastIndexOf(responseMetadataMarker); + if (markerIndex === -1) { + return new Response(stdout, { + status: 200, + headers: { 'Content-Type': 'application/json' }, + }); + } + + const responseBody = stdout.slice(0, markerIndex).replace(/\n$/, ''); + const [statusText, contentType] = stdout + .slice(markerIndex + responseMetadataMarker.length) + .split('\t', 2); + const parsedStatus = Number(statusText); + const status = + Number.isInteger(parsedStatus) && parsedStatus >= 200 && parsedStatus <= 599 + ? parsedStatus + : 500; + return new Response(responseBody, { + status, + ...(contentType ? { headers: { 'Content-Type': contentType.trim() } } : {}), }); } } @@ -356,11 +396,19 @@ export class ContainerFetchWrapperTransport implements WrapperTransport { this.port = options.port; } - async request(method: 'GET' | 'POST', path: string, body?: unknown): Promise { + async request( + method: 'GET' | 'POST', + path: string, + body?: unknown, + requestId?: string + ): Promise { const url = new URL(`http://localhost:${this.port}${path}`); const request = new Request(url, { method, - headers: { 'Content-Type': 'application/json' }, + headers: { + 'Content-Type': 'application/json', + ...(requestId ? { [WRAPPER_REQUEST_ID_HEADER]: requestId } : {}), + }, ...(body !== undefined ? { body: JSON.stringify(body) } : {}), }); return this.sandbox.containerFetch(request, this.port); @@ -522,8 +570,10 @@ export class WrapperClient { * Make an HTTP request to the wrapper. */ private async request(method: 'GET' | 'POST', path: string, body?: unknown): Promise { - const response = await this.transport.request(method, path, body); - const responseText = await response.text(); + const requestId = crypto.randomUUID(); + const response = await this.transport.request(method, path, body, requestId); + const responseBytes = new Uint8Array(await response.arrayBuffer()); + const responseText = new TextDecoder().decode(responseBytes); if (!responseText.trim()) { // Some endpoints return empty body @@ -531,11 +581,23 @@ export class WrapperClient { } try { - const parsed = JSON.parse(responseText) as T & { + const raw: unknown = JSON.parse(responseText); + if (typeof raw !== 'object' || raw === null || Array.isArray(raw)) { + throw new Error('Invalid wrapper response object'); + } + const parsed = raw as T & { error?: string; message?: string; wrapperRunId?: string; }; + if ( + (parsed.error !== undefined && typeof parsed.error !== 'string') || + (parsed.message !== undefined && typeof parsed.message !== 'string') || + (parsed.wrapperRunId !== undefined && typeof parsed.wrapperRunId !== 'string') || + (!response.ok && parsed.error === undefined) + ) { + throw new Error('Invalid wrapper application-error response'); + } // Check for error response if (parsed.error || !response.ok) { @@ -551,6 +613,7 @@ export class WrapperClient { method, path, ...this.protocolDiagnosticFields(), + requestId, errorCode, statusCode, }) @@ -589,7 +652,10 @@ export class WrapperClient { method, path, ...this.protocolDiagnosticFields(), - responseBytes: responseText.length, + requestId, + contentType: response.headers.get('content-type'), + responseBytes: responseBytes.byteLength, + responseSha256: await sha256Hex(responseBytes), statusCode: response.status, }) .error('Failed to parse wrapper HTTP response'); diff --git a/services/cloud-agent-next/src/session/queries/events.ts b/services/cloud-agent-next/src/session/queries/events.ts index 92394dddb1..678e1a1d75 100644 --- a/services/cloud-agent-next/src/session/queries/events.ts +++ b/services/cloud-agent-next/src/session/queries/events.ts @@ -1,4 +1,4 @@ -import { count, max, eq, and, gt, gte, lte, lt, inArray, asc, like } from 'drizzle-orm'; +import { count, max, eq, and, gt, gte, lte, lt, inArray, asc, sql } from 'drizzle-orm'; import type { DrizzleSqliteDODatabase } from 'drizzle-orm/durable-sqlite'; import * as z from 'zod'; import type { StoredEvent } from '../../websocket/types.js'; @@ -317,7 +317,7 @@ export function createEventQueries(db: DrizzleSqliteDODatabase, rawSql: SqlStora timestamp: events.timestamp, }) .from(events) - .where(like(events.entity_id, `${prefix}%`)) + .where(sql`substr(${events.entity_id}, 1, length(${prefix})) = ${prefix}`) .orderBy(asc(events.timestamp), asc(events.id)) .all() satisfies StoredEvent[]; }, diff --git a/services/cloud-agent-next/src/session/safe-failure-projection.test.ts b/services/cloud-agent-next/src/session/safe-failure-projection.test.ts index f73dbaf649..448e7ec716 100644 --- a/services/cloud-agent-next/src/session/safe-failure-projection.test.ts +++ b/services/cloud-agent-next/src/session/safe-failure-projection.test.ts @@ -126,6 +126,39 @@ describe('classifyAssistantFailureMessage', () => { }) ).toBe('Assistant request timed out'); }); + + it.each([ + [ + 'Payment Required: {"error":{"message":"Insufficient credits","balance":-0.01,"url":"https://gateway.test/private"}}', + 'payment_required', + 'Assistant request failed: insufficient credits', + ], + [ + 'Account balance is -$0.25 for tenant secret', + 'payment_required', + 'Assistant request failed: insufficient credits', + ], + ['load balance is -1 for shard secret', undefined, 'Assistant request failed'], + [ + 'model_not_found: provider/private-model', + 'model_missing', + 'Assistant request failed: model not found', + ], + ['429 Too Many Requests', undefined, 'Assistant request was rate limited'], + ['upstream timeout', undefined, 'Assistant request timed out'], + ['403 Forbidden', undefined, 'Assistant request was not authorized'], + ['400 malformed request', undefined, 'Assistant request was invalid'], + ['504 upstream unavailable', undefined, 'Assistant service is unavailable'], + ['unknown provider failure', undefined, 'Assistant request failed'], + ] as const)( + 'returns a structured terminal classification for %s', + (source, terminalCode, safeMessage) => { + const result = classifyAssistantFailure(source); + + expect(result.safeMessage).toBe(safeMessage); + expect(result.terminalCode).toBe(terminalCode); + } + ); }); describe('classifyAssistantFailure', () => { diff --git a/services/cloud-agent-next/src/session/safe-failure-projection.ts b/services/cloud-agent-next/src/session/safe-failure-projection.ts index 3fe456be41..27825d58e3 100644 --- a/services/cloud-agent-next/src/session/safe-failure-projection.ts +++ b/services/cloud-agent-next/src/session/safe-failure-projection.ts @@ -84,7 +84,10 @@ export function classifyAssistantFailure( ): AssistantFailureClassification { const message = extractErrorMessage(source).toLocaleLowerCase(); const providerOwnership = /\[byok\]/i.test(message) ? 'byok' : defaultProviderOwnership; - if (/\b(payment required|insufficient (?:credits?|balance|funds))\b/.test(message)) { + if ( + /\b(payment required|insufficient (?:credits?|balance|funds))\b/.test(message) || + /\b(?:account|credit)s?\b[\s\S]{0,80}?\bbalance(?: is|:)?\s*-\s*\$?\d/.test(message) + ) { return { reason: 'insufficient_credits', safeMessage: 'Assistant request failed: insufficient credits', @@ -92,7 +95,11 @@ export function classifyAssistantFailure( terminalCode: 'payment_required', }; } - if (/\b(model (?:was )?not found|unknown model|invalid model)\b/.test(message)) { + if ( + /\b(model (?:was )?not found|model[_ -]?not[_ -]?found|unknown model|invalid model|model .+ does not exist)\b/.test( + message + ) + ) { return { reason: 'model_unavailable', safeMessage: 'Assistant request failed: model not found', diff --git a/services/cloud-agent-next/src/shared/wrapper-http.ts b/services/cloud-agent-next/src/shared/wrapper-http.ts new file mode 100644 index 0000000000..30bd1d5c37 --- /dev/null +++ b/services/cloud-agent-next/src/shared/wrapper-http.ts @@ -0,0 +1,2 @@ +export const WRAPPER_REQUEST_ID_HEADER = 'x-kilo-request-id'; +export const WRAPPER_HEALTH_CHECK_FAILED = 'HEALTH_CHECK_FAILED'; diff --git a/services/cloud-agent-next/src/telemetry/queue-reports.test.ts b/services/cloud-agent-next/src/telemetry/queue-reports.test.ts index aed4ca3263..54cbb0128d 100644 --- a/services/cloud-agent-next/src/telemetry/queue-reports.test.ts +++ b/services/cloud-agent-next/src/telemetry/queue-reports.test.ts @@ -205,19 +205,25 @@ describe('Cloud Agent report emitter', () => { expect(JSON.stringify(reports)).not.toContain('unrelated secret-bearing failure text'); }); - it('emits a safe insufficient-credit diagnostic for the wrapper terminal text', async () => { + it('emits a safe insufficient-credit diagnostic for an appended gateway body', async () => { const reports: CloudAgentQueueReport[] = []; + const rawGatewayError = + 'Payment Required: {"balance":-1.25,"url":"https://gateway.test/private","token":"secret"}'; await emitRunStateReport({ queue: { send: async report => void reports.push(report) }, cloudAgentSessionId: 'agent_report', - state: { ...state, error: 'Insufficient credits' }, + state: { ...state, error: rawGatewayError }, }); expect(reports[0]?.run.diagnostic).toEqual({ errorMessageRedacted: 'Model request failed: insufficient credits', errorExpiresAt: new Date(5 + 30 * 24 * 60 * 60 * 1000).toISOString(), }); - expect(JSON.stringify(reports)).not.toContain('Insufficient credits'); + const serialized = JSON.stringify(reports); + expect(serialized).not.toContain(rawGatewayError); + expect(serialized).not.toContain('-1.25'); + expect(serialized).not.toContain('gateway.test'); + expect(serialized).not.toContain('secret'); }); it.each(['Payment Required', 'pAyMeNt ReQuIrEd'])( diff --git a/services/cloud-agent-next/src/telemetry/queue-reports.ts b/services/cloud-agent-next/src/telemetry/queue-reports.ts index 76cfb5ff69..f99c34a98a 100644 --- a/services/cloud-agent-next/src/telemetry/queue-reports.ts +++ b/services/cloud-agent-next/src/telemetry/queue-reports.ts @@ -5,7 +5,10 @@ import { type CloudAgentRunStateReport, } from '@kilocode/worker-utils/cloud-agent-queue-report'; import { logger } from '../logger.js'; -import { workspaceFailureMessage } from '../session/safe-failure-projection.js'; +import { + classifyAssistantFailure, + workspaceFailureMessage, +} from '../session/safe-failure-projection.js'; import { admittedAgentModel, type SessionMessageState } from '../session/session-message-state.js'; import { classifyCloudAgentFailure, @@ -22,12 +25,6 @@ type ReportLogContext = { status: string; }; -const INSUFFICIENT_CREDIT_TERMINAL_ERRORS = new Set([ - 'insufficient credits', - 'insufficient credits: payment_required', - 'insufficient credits: insufficient_funds', - 'payment required', -]); const FAILED_RUN_DIAGNOSTIC_MESSAGES: Partial< Record, string> > = { @@ -68,7 +65,7 @@ function isKnownInsufficientCreditFailure(state: SessionMessageState): boolean { return false; } if (state.error === undefined) return false; - return INSUFFICIENT_CREDIT_TERMINAL_ERRORS.has(state.error.trim().toLowerCase()); + return classifyAssistantFailure(state.error).terminalCode === 'payment_required'; } function diagnosticForFailedRun( diff --git a/services/cloud-agent-next/src/telemetry/report-store.test.ts b/services/cloud-agent-next/src/telemetry/report-store.test.ts index 81f7aa0069..647cbed61e 100644 --- a/services/cloud-agent-next/src/telemetry/report-store.test.ts +++ b/services/cloud-agent-next/src/telemetry/report-store.test.ts @@ -315,6 +315,40 @@ describe('cloud agent reporting store', () => { }); }); + it.each([ + ['agent_activity', 'payment_required'], + ['agent_activity', 'model_missing'], + ['post_dispatch_no_activity', 'payment_required'], + ['post_dispatch_no_activity', 'model_missing'], + ] as const)('persists the explicit %s/%s classification', async (failureStage, failureCode) => { + const fake = makeDb([[{ createdAt: occurredAt }], []]); + const store = createCloudAgentReportStore(fake.db as never); + + await store.saveReport( + { + version: 1, + type: 'run.state', + occurredAt, + session: { cloudAgentSessionId }, + run: { + messageId: `msg_${failureStage}_${failureCode}`, + status: 'failed', + terminalAt: occurredAt, + failureStage, + failureCode, + }, + }, + occurredAt + ); + + expect( + fake.inserts.find(call => call.table === cloud_agent_session_runs)?.values + ).toMatchObject({ + failure_stage: failureStage, + failure_code: failureCode, + }); + }); + it('keeps established terminal outcomes and earliest observed dispatch on replay', async () => { const fake = makeDb([ [{ createdAt: occurredAt }], diff --git a/services/cloud-agent-next/src/websocket/ingest.test.ts b/services/cloud-agent-next/src/websocket/ingest.test.ts index 310a17029b..0e98a7d685 100644 --- a/services/cloud-agent-next/src/websocket/ingest.test.ts +++ b/services/cloud-agent-next/src/websocket/ingest.test.ts @@ -1051,6 +1051,53 @@ describe('createIngestHandler', () => { ); }); + it.each([ + { + source: 'Payment Required: {"balance":-0.25,"token":"secret"}', + failureCode: 'payment_required' as const, + safeFailureMessage: 'Assistant request failed: insufficient credits', + }, + { + source: 'model_not_found: provider/private-model', + failureCode: 'model_missing' as const, + safeFailureMessage: 'Assistant request failed: model not found', + }, + ])('preserves $failureCode from message.updated ingest', async classification => { + const doContext = createNewPathDOContext(); + const handler = createIngestHandler( + createFakeState(), + createFakeEventQueries(), + SESSION_ID, + vi.fn(), + doContext + ); + + await handler.handleIngestMessage( + createFakeWebSocket(makeNewPathAttachment()), + makeStreamMessage('kilocode', { + event: 'message.updated', + properties: { + info: { + id: 'asst_classified', + role: 'assistant', + parentID: 'msg_classified', + error: classification.source, + }, + }, + }) + ); + + expect(doContext.terminalizeSessionMessageOnce).toHaveBeenCalledWith( + 'msg_classified', + expect.objectContaining({ + kind: 'failed', + failureCode: classification.failureCode, + safeFailureMessage: classification.safeFailureMessage, + }), + WRAPPER_RUN_ID + ); + }); + it('publishes and persists a safe session error with session correlation intact', async () => { const eventQueries = createFakeEventQueries(); const broadcast = vi.fn(); @@ -1378,6 +1425,7 @@ describe('createIngestHandler', () => { status: 'failed', error: rawError, errorSource: 'assistant', + failureCode: 'payment_required', }); }); @@ -1439,6 +1487,7 @@ describe('createIngestHandler', () => { status: 'failed', error: 'Model not found: kilo/retired-model', errorSource: 'assistant', + failureCode: 'model_missing', modelNotFoundRuntimeDiagnostics: diagnostics, }); }); diff --git a/services/cloud-agent-next/src/websocket/ingest.ts b/services/cloud-agent-next/src/websocket/ingest.ts index cd031d374e..19f137d170 100644 --- a/services/cloud-agent-next/src/websocket/ingest.ts +++ b/services/cloud-agent-next/src/websocket/ingest.ts @@ -909,10 +909,11 @@ export function createIngestHandler( const errorData = parsedError.data; if (errorData.fatal) { const fatalMessage = errorData.error ?? errorData.message ?? 'Fatal error'; - const safeFatalMessage = + const assistantClassification = errorData.errorSource === 'assistant' - ? classifyAssistantFailureMessage(fatalMessage) - : 'Agent wrapper failed'; + ? classifyAssistantFailure(fatalMessage) + : undefined; + const safeFatalMessage = assistantClassification?.safeMessage ?? 'Agent wrapper failed'; const shouldForwardModelNotFoundDiagnostics = errorData.errorSource === 'assistant' && safeFatalMessage === MODEL_NOT_FOUND_SAFE_ERROR_MESSAGE; @@ -963,7 +964,7 @@ export function createIngestHandler( ...(parsedDiagnostics?.success ? { modelNotFoundRuntimeDiagnostics: parsedDiagnostics.data } : {}), - failureCode: errorData.failureCode, + failureCode: errorData.failureCode ?? assistantClassification?.terminalCode, }); logger .withFields({ diff --git a/services/cloud-agent-next/test/e2e/README.md b/services/cloud-agent-next/test/e2e/README.md index d9d25d2b4f..62597c5af0 100644 --- a/services/cloud-agent-next/test/e2e/README.md +++ b/services/cloud-agent-next/test/e2e/README.md @@ -140,12 +140,37 @@ Matrix (runs the default regression suite): tsx services/cloud-agent-next/test/e2e/smoke.ts ``` +The dev `SandboxSmall` capacity is 16 because this matrix creates fourteen +isolated sessions. Lowering it below the matrix demand turns later cold rows +into local capacity probes rather than Cloud Agent health checks. + The matrix starts with `cold-hot`, which pays one cold sandbox boot and then runs several hot same-session turns. Fresh sessions use per-session sandboxes in local dev, so the harness identifies each newly-created sandbox instead of killing every sandbox between cases. Kill scenarios only terminate the sandbox family created for that scenario. +Warm-reuse scenarios allow preparation snapshots and the live sandbox/Kilo +verification steps that occur on every delivery. They fail only when a live +cold-path preparation step starts or the sandbox identity changes, so a warm +turn's expected preparation chatter does not hide a real reprovisioning +regression or create a false failure. + +Between matrix rows, the harness deletes every session created by that row +through the trusted `cleanupSession` control plane. For local per-session +`ses-*` sandboxes, deletion destroys the isolated sandbox Durable Object; a +shared sandbox would retain the container and delete only the named session. +Pool ownership is released first through trusted Worker/Sandbox cleanup. Only +retryable `physical wrapper cleanup pending` responses are polled for up to 30 +seconds so intentional kill scenarios can converge; product scenarios are not +rerun. Only after cleanup succeeds does the harness remove the row's exact post-baseline +Docker primary/proxy family, which contains local runtime leftovers without +bypassing live pool bookkeeping. It then requires five seconds of stable +absence. A cold session may reuse a warm container, so readiness is proven by +finding that session's wrapper marker rather than accepting any new Docker ID. +If trusted cleanup or Docker quiescence fails, the matrix reports +`matrix-cleanup` and stops. + Per-run overrides via env vars. Defaults assume a zero-offset session; for any other offset, compute the real ports from `pnpm dev:status --json` (worker = `8794 + portOffset`, fake-LLM = `8811 + portOffset`): @@ -240,6 +265,7 @@ These are wrapped by `releaseGate()`, `waitForGateEngaged()`, | `queue-overflow` | Block on `gate:overflow`, fill the pending queue until enqueue fails with HTTP 429, release gate, drain. | | `queue-interrupt-clears` | Block on `gate:`, enqueue two, `interruptSession`, assert `cloud.message.failed` with `reason: 'interrupted'` for each. | | `llm-error` | Return fake provider HTTP 402 and assert the turn reaches a failed terminal event instead of hanging. | +| `classified-failure-report` | With `payment` or `model`, assert the terminal classification and safe diagnostic are persisted in the run report. | | `chunked-streaming` | Stream delayed fake chunks and assert multiple downstream `message.part.delta` events survive. | | `empty-response` | Run `idle`, assert completion, and assert no downstream `message.part.delta` is emitted. | | `interrupt-mid-stream` | Interrupt an actively gated fake request and assert the active message is interrupted, not a queued message. | diff --git a/services/cloud-agent-next/test/e2e/auth.ts b/services/cloud-agent-next/test/e2e/auth.ts index 51501f0a8e..88c82be0e9 100644 --- a/services/cloud-agent-next/test/e2e/auth.ts +++ b/services/cloud-agent-next/test/e2e/auth.ts @@ -16,7 +16,14 @@ import { createHash, randomUUID } from 'node:crypto'; import path from 'node:path'; import process from 'node:process'; import jwt from 'jsonwebtoken'; -import { computeDatabaseUrl, createDrizzleClient, kilocode_users, sql } from '@kilocode/db'; +import { and, eq } from 'drizzle-orm'; +import { + cloud_agent_session_runs, + computeDatabaseUrl, + createDrizzleClient, + kilocode_users, + sql, +} from '@kilocode/db'; export const DRIVER_USER_EMAIL_SUFFIX = '@cloud-agent-next-e2e.example.com'; export const FUNDED_DRIVER_BALANCE_MICRODOLLARS = 10_000_000; @@ -148,6 +155,50 @@ export async function ensureTestUser( } } +export type StoredRunFailure = { + status: string; + failureStage: string | null; + failureCode: string | null; + errorMessageRedacted: string | null; +}; + +export async function waitForStoredRunFailure( + databaseUrl: string | undefined, + cloudAgentSessionId: string, + messageId: string, + timeoutMs = 20_000 +): Promise { + const driver = createDrizzleClient({ + connectionString: databaseUrl ?? computeDatabaseUrl(), + poolConfig: { application_name: 'cloud-agent-next-e2e-driver', max: 1 }, + }); + const deadline = Date.now() + timeoutMs; + try { + while (Date.now() < deadline) { + const [row] = await driver.db + .select({ + status: cloud_agent_session_runs.status, + failureStage: cloud_agent_session_runs.failure_stage, + failureCode: cloud_agent_session_runs.failure_code, + errorMessageRedacted: cloud_agent_session_runs.error_message_redacted, + }) + .from(cloud_agent_session_runs) + .where( + and( + eq(cloud_agent_session_runs.cloud_agent_session_id, cloudAgentSessionId), + eq(cloud_agent_session_runs.message_id, messageId) + ) + ) + .limit(1); + if (row?.status === 'failed') return row; + await new Promise(resolve => setTimeout(resolve, 250)); + } + return null; + } finally { + await driver.pool.end().catch(() => {}); + } +} + // --------------------------------------------------------------------------- // JWT minting // --------------------------------------------------------------------------- diff --git a/services/cloud-agent-next/test/e2e/client.ts b/services/cloud-agent-next/test/e2e/client.ts index d00060404d..1b5bee6610 100644 --- a/services/cloud-agent-next/test/e2e/client.ts +++ b/services/cloud-agent-next/test/e2e/client.ts @@ -30,6 +30,8 @@ export type CallbackTarget = { export type DriverConfig = { workerUrl: string; + /** Postgres URL used by scenarios that verify persisted report rows. */ + databaseUrl?: string; user: TestUser; nextAuthSecret: string; /** @@ -55,6 +57,8 @@ export type DriverConfig = { * `/test/release`, `/test/gate-status`, and `/test/requests`. */ fakeLlmUrl: string; + /** Matrix-only hook for row-scoped Durable Object cleanup. */ + onSessionCreated?: (cloudAgentSessionId: string) => void; }; export const DEFAULT_CONFIG: Omit = { @@ -146,8 +150,12 @@ export async function startSession( args: StartSessionArgs, api: ApiVersion = 'unified' ): Promise { - if (api === 'legacy') return startSessionLegacy(config, args); - return startSessionUnified(config, args); + const result = + api === 'legacy' + ? await startSessionLegacy(config, args) + : await startSessionUnified(config, args); + config.onSessionCreated?.(result.cloudAgentSessionId); + return result; } async function startSessionUnified( @@ -300,6 +308,51 @@ export async function interruptSession( return trpcCall(config, 'interruptSession', { sessionId }); } +export async function cleanupSession( + config: DriverConfig, + sessionId: string +): Promise<{ success: true; message?: string }> { + if (!config.internalApiSecret) { + throw new Error('cleanupSession requires INTERNAL_API_SECRET from .dev.vars'); + } + return trpcCall( + config, + 'cleanupSession', + { sessionId }, + { + internalApiSecret: config.internalApiSecret, + } + ); +} + +export async function cleanupSessionUntilSettled( + config: DriverConfig, + sessionId: string, + options: { + timeoutMs?: number; + pollIntervalMs?: number; + now?: () => number; + sleep?: (ms: number) => Promise; + } = {} +): Promise<{ success: true; message?: string }> { + const { + timeoutMs = 30_000, + pollIntervalMs = 1_000, + now = Date.now, + sleep = ms => new Promise(resolve => setTimeout(resolve, ms)), + } = options; + const deadline = now() + timeoutMs; + while (true) { + try { + return await cleanupSession(config, sessionId); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + if (!message.includes('"retryable":true') || now() >= deadline) throw error; + await sleep(Math.max(0, Math.min(pollIntervalMs, deadline - now()))); + } + } +} + export async function answerPermission( config: DriverConfig, sessionId: string, diff --git a/services/cloud-agent-next/test/e2e/fake-llm-server.ts b/services/cloud-agent-next/test/e2e/fake-llm-server.ts index 94393a9d74..ce4f2a392d 100644 --- a/services/cloud-agent-next/test/e2e/fake-llm-server.ts +++ b/services/cloud-agent-next/test/e2e/fake-llm-server.ts @@ -371,6 +371,10 @@ export const scenarioRegistry: Record = { writeJsonError(ctx.res, 402, message, 'insufficient_quota'); }, + 'model-missing'(_args, ctx) { + writeJsonError(ctx.res, 404, 'Model not found: kilo/fake-deterministic', 'model_not_found'); + }, + gate(args, ctx) { const rawArg = args[0] ?? ''; // Kilo augments the user message with `...` and other diff --git a/services/cloud-agent-next/test/e2e/lifecycle.ts b/services/cloud-agent-next/test/e2e/lifecycle.ts index e4a89b6587..e6ed419acb 100644 --- a/services/cloud-agent-next/test/e2e/lifecycle.ts +++ b/services/cloud-agent-next/test/e2e/lifecycle.ts @@ -18,9 +18,11 @@ import { type DriverConfig, type StreamEvent, } from './client.js'; +import { waitForStoredRunFailure } from './auth.js'; import { startCallbackServer, type CallbackServerHandle } from './callback-server.js'; import { killSandboxFamily, + listSandboxesForAgentSession, listSandboxContainers, readKiloCliLog, readWrapperLog, @@ -79,6 +81,34 @@ function hasEventOfType(events: StreamEvent[], type: string): boolean { return events.some(e => e.streamEventType === type); } +const WARM_VERIFICATION_PREPARATION_STEPS = new Set([ + 'sandbox_provision', + 'sandbox_boot', + 'kilo_server', +]); + +/** + * Return live preparation steps that prove a supposedly warm turn did cold work. + * Snapshot and attempt-level events are expected on warm stream connections. + */ +export function unexpectedColdPreparationSteps(events: StreamEvent[]): string[] { + const unexpected = new Set(); + for (const event of events) { + if (event.streamEventType !== 'preparing') continue; + + const { version, action, step } = event.data; + if (version !== 2) { + unexpected.add(`legacy:${typeof step === 'string' ? step : 'unknown'}`); + continue; + } + if (action !== 'step_started') continue; + + const stepName = typeof step === 'string' ? step : 'unknown'; + if (!WARM_VERIFICATION_PREPARATION_STEPS.has(stepName)) unexpected.add(stepName); + } + return [...unexpected]; +} + async function snapshotSandboxIds(): Promise> { const containers = await listSandboxContainers(); return new Set(containers.map(container => container.id)); @@ -103,7 +133,11 @@ export async function lifecycleCold(args: LifecycleArgs): Promise sandbox.id === warmupSandbox.id) && - sandboxesAfter.every(sandbox => sandboxIdsBeforeFollowup.has(sandbox.id)); + sessionSandboxesAfter.length === 1 && sessionSandboxesAfter[0]?.id === warmupSandbox.id; const terminalName = terminal?.streamEventType ?? 'none'; - const ok = (conversation === 'hang' ? !terminal : !!terminal) && noPrepare && sameContainers; + const ok = + (conversation === 'hang' ? !terminal : !!terminal) && + coldPreparationSteps.length === 0 && + sameContainers; return { name: 'hot', conversation, ok, - message: `terminal=${terminalName}, firstKilocode=${firstKilocode ? `${firstKilocodeLatency}ms` : 'none'}, noPrepare=${noPrepare}, sameContainers=${sameContainers}`, + message: `terminal=${terminalName}, firstKilocode=${firstKilocode ? `${firstKilocodeLatency}ms` : 'none'}, coldPreparation=${coldPreparationSteps.join(',') || 'none'}, sameContainers=${sameContainers}`, events, durationMs: Date.now() - start, }; @@ -282,7 +321,11 @@ export async function lifecycleColdHot(args: LifecycleArgs): Promise candidate.id === sandbox.id) && - sandboxesAfter.every(candidate => sandboxIdsBeforeFollowup.has(candidate.id)); - if (!completed || !noPrepare || !sameContainers) { + sessionSandboxesAfter.length === 1 && sessionSandboxesAfter[0]?.id === sandbox.id; + if (!completed || coldPreparationSteps.length > 0 || !sameContainers) { return { name: 'cold-hot', conversation, ok: false, - message: `${directive}: terminal=${hotResult.terminal?.streamEventType ?? 'none'}, firstKilocode=${firstKilocode ? `${firstKilocodeLatency}ms` : 'none'}, noPrepare=${noPrepare}, sameContainers=${sameContainers}`, + message: `${directive}: terminal=${hotResult.terminal?.streamEventType ?? 'none'}, firstKilocode=${firstKilocode ? `${firstKilocodeLatency}ms` : 'none'}, coldPreparation=${coldPreparationSteps.join(',') || 'none'}, sameContainers=${sameContainers}`, events, durationMs: Date.now() - start, }; } hotSummaries.push( - `${directive}:${hotResult.terminal.streamEventType}/${firstKilocode ? `${firstKilocodeLatency}ms` : 'no-kilocode'}` + `${directive}:${hotResult.terminal?.streamEventType ?? 'none'}/${firstKilocode ? `${firstKilocodeLatency}ms` : 'no-kilocode'}` ); } @@ -382,7 +423,11 @@ export async function lifecycleExternalKill(args: LifecycleArgs): Promise { + const start = Date.now(); + const { config, conversation, timeoutMs = 60_000, api = 'unified' } = args; + const expected = + conversation === 'payment' + ? { + directive: 'error:Payment Required: balance is -$12.34', + code: 'payment_required', + diagnostic: 'Model request failed: insufficient credits', + } + : conversation === 'model' + ? { + directive: 'model-missing', + code: 'model_missing', + diagnostic: 'No model is available for this run', + } + : null; + if (!expected) { + return { + name: 'classified-failure-report', + conversation, + ok: false, + message: 'conversation must be payment or model', + events: [], + durationMs: Date.now() - start, + }; + } + + try { + const session = await startSession(config, { prompt: fakeDirective(expected.directive) }, api); + const stream = openStream(config, session.cloudAgentSessionId, { replay: false }); + const terminal = await stream.waitFor( + event => + event.streamEventType === 'cloud.message.failed' && + messageIdFromEvent(event) === session.messageId, + timeoutMs + ); + const events = [...stream.events]; + stream.close(); + if (!terminal) { + return { + name: 'classified-failure-report', + conversation, + ok: false, + message: `no matching cloud.message.failed within ${timeoutMs}ms`, + events, + durationMs: Date.now() - start, + }; + } + + const row = await waitForStoredRunFailure( + config.databaseUrl, + session.cloudAgentSessionId, + session.messageId + ); + const ok = + row?.failureStage === 'agent_activity' && + row.failureCode === expected.code && + row.errorMessageRedacted === expected.diagnostic; + return { + name: 'classified-failure-report', + conversation, + ok, + message: row + ? `stored stage=${row.failureStage} code=${row.failureCode} diagnostic=${JSON.stringify(row.errorMessageRedacted)}` + : 'failed run report was not persisted within 20s', + events, + durationMs: Date.now() - start, + }; + } catch (err) { + return { + name: 'classified-failure-report', + conversation, + ok: false, + message: `threw: ${err instanceof Error ? err.message : String(err)}`, + events: [], + durationMs: Date.now() - start, + }; + } +} + /** * chunked-streaming: drives `__fake__:slow::` (defaults 5:50). The fake * emits assistant content chunks separated by ms delays. Assert @@ -1204,7 +1357,11 @@ export async function lifecycleChunkedStreaming(args: LifecycleArgs): Promise
  • { nextAuthSecret: devVars.NEXTAUTH_SECRET ?? '', internalApiSecret: devVars.INTERNAL_API_SECRET, workerUrl: process.env.WORKER_URL ?? DEFAULT_CONFIG.workerUrl, + databaseUrl: process.env.DATABASE_URL, gitUrl: process.env.E2E_GIT_URL ?? DEFAULT_CONFIG.gitUrl, model: process.env.E2E_MODEL ?? DEFAULT_CONFIG.model, }; diff --git a/services/cloud-agent-next/test/e2e/sandbox-control.ts b/services/cloud-agent-next/test/e2e/sandbox-control.ts index 9671f105ab..358d9518a7 100644 --- a/services/cloud-agent-next/test/e2e/sandbox-control.ts +++ b/services/cloud-agent-next/test/e2e/sandbox-control.ts @@ -17,6 +17,17 @@ const execFileAsync = promisify(execFile); export type DockerCommandExecutor = (args: string[]) => Promise<{ stdout: string }>; +export type SandboxCleanupQuiescenceOptions = { + timeoutMs: number; + stableMs: number; + /** Reap delayed recreations only after the Sandbox runtime has released ownership. */ + reapPostBaseline?: boolean; + pollIntervalMs?: number; + executeDocker?: DockerCommandExecutor; + now?: () => number; + sleep?: (ms: number) => Promise; +}; + const executeDockerCommand: DockerCommandExecutor = async args => { const { stdout } = await execFileAsync('docker', args); return { stdout }; @@ -34,9 +45,15 @@ export type SandboxContainer = { * callers can kill them together with their primary. */ export async function listSandboxContainers( - executeDocker: DockerCommandExecutor = executeDockerCommand + executeDocker: DockerCommandExecutor = executeDockerCommand, + includeStopped = false ): Promise { - const { stdout } = await executeDocker(['ps', '--format', '{{.ID}}\t{{.Names}}\t{{.Image}}']); + const { stdout } = await executeDocker([ + 'ps', + ...(includeStopped ? ['-a'] : []), + '--format', + '{{.ID}}\t{{.Names}}\t{{.Image}}', + ]); const result: SandboxContainer[] = []; for (const line of stdout.trim().split('\n')) { if (!line) continue; @@ -73,16 +90,36 @@ export async function killContainer( } } +async function removeContainer( + idOrName: string, + executeDocker: DockerCommandExecutor +): Promise { + try { + await executeDocker(['rm', '-f', idOrName]); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + if (message.includes('No such container')) return; + throw error; + } +} + /** Block until a primary sandbox appears that was not present in `knownIds`. */ export async function waitForNewSandboxPresent( knownIds: Set, - timeoutMs: number + timeoutMs: number, + agentSessionId?: string, + executeDocker: DockerCommandExecutor = executeDockerCommand ): Promise { const deadline = Date.now() + timeoutMs; while (Date.now() < deadline) { - const containers = await listSandboxContainers(); - const primary = containers.find(c => !c.isProxy && !knownIds.has(c.id)); - if (primary) return primary; + if (agentSessionId) { + const [reused] = await listSandboxesForAgentSession(agentSessionId, executeDocker); + if (reused) return reused; + } else { + const containers = await listSandboxContainers(executeDocker); + const primary = containers.find(c => !c.isProxy && !knownIds.has(c.id)); + if (primary) return primary; + } await new Promise(r => setTimeout(r, 500)); } return null; @@ -170,6 +207,28 @@ export async function killSandboxFamily( return killed; } +/** Remove a matrix-owned sandbox family, including stopped container corpses. */ +export async function removeSandboxFamily( + sandbox: SandboxContainer, + executeDocker: DockerCommandExecutor = executeDockerCommand +): Promise { + const familyNames = sandboxFamilyNames(sandbox); + const containers = await listSandboxContainers(executeDocker, true); + const removed: string[] = []; + for (const container of containers) { + if (!familyNames.has(container.name)) continue; + try { + await executeDocker(['rm', '-f', container.id]); + removed.push(container.name); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + if (message.includes('No such container')) continue; + throw error; + } + } + return removed; +} + /** Block until a sandbox container and its proxy sibling are gone. */ export async function waitForSandboxFamilyGone( sandbox: SandboxContainer, @@ -185,6 +244,53 @@ export async function waitForSandboxFamilyGone( return false; } +/** + * Wait until post-baseline sandboxes remain absent for one continuous window. + * Optional reaping is reserved for containment after supported teardown fails. + */ +export async function waitForSandboxCleanupQuiescence( + baselineSandboxIds: Set, + options: SandboxCleanupQuiescenceOptions +): Promise { + const { + timeoutMs, + stableMs, + reapPostBaseline = false, + pollIntervalMs = 250, + executeDocker = executeDockerCommand, + now = Date.now, + sleep = ms => new Promise(resolve => setTimeout(resolve, ms)), + } = options; + const deadline = now() + timeoutMs; + let absentSince: number | undefined; + + while (true) { + const containers = await listSandboxContainers(executeDocker); + const observedAt = now(); + if (observedAt > deadline) return false; + + const postBaselineSandboxes = containers.filter( + container => !baselineSandboxIds.has(container.id) + ); + if (postBaselineSandboxes.length > 0) { + absentSince = undefined; + if (reapPostBaseline) { + for (const container of postBaselineSandboxes) { + await killContainer(container.id, executeDocker); + await removeContainer(container.id, executeDocker); + } + } + } else { + absentSince ??= observedAt; + if (observedAt - absentSince >= stableMs) return true; + } + + const remainingMs = deadline - now(); + if (remainingMs <= 0) return false; + await sleep(Math.min(pollIntervalMs, remainingMs)); + } +} + /** * Read the wrapper log file inside a running sandbox container. Used for * smoke tests to assert "using fake kilo client" is present after boot. diff --git a/services/cloud-agent-next/test/e2e/smoke.ts b/services/cloud-agent-next/test/e2e/smoke.ts index 4f1b446d30..b858742242 100644 --- a/services/cloud-agent-next/test/e2e/smoke.ts +++ b/services/cloud-agent-next/test/e2e/smoke.ts @@ -16,13 +16,19 @@ import path from 'node:path'; import { fileURLToPath } from 'node:url'; import { ensureTestUser, loadDevVars, loadRepoEnvFiles, DRIVER_USER_EMAIL_SUFFIX } from './auth.js'; -import { DEFAULT_CONFIG, type ApiVersion, type DriverConfig } from './client.js'; +import { + cleanupSessionUntilSettled, + DEFAULT_CONFIG, + type ApiVersion, + type DriverConfig, +} from './client.js'; import { LIFECYCLE_SCENARIOS, type LifecycleResult } from './lifecycle.js'; import { printResult } from './run.js'; import { - killSandboxFamily, listSandboxContainers, - waitForSandboxFamilyGone, + removeSandboxFamily, + sandboxFamilyKey, + waitForSandboxCleanupQuiescence, type SandboxContainer, } from './sandbox-control.js'; @@ -50,6 +56,8 @@ const DEFAULT_MATRIX: Case[] = [ // Failure, streaming, and fake-server cleanup edge cases. { lifecycle: 'llm-error', conversation: 'boom' }, + { lifecycle: 'classified-failure-report', conversation: 'payment' }, + { lifecycle: 'classified-failure-report', conversation: 'model' }, { lifecycle: 'chunked-streaming', conversation: 'slow:5:50' }, { lifecycle: 'empty-response', conversation: '_' }, { lifecycle: 'interrupt-mid-stream', conversation: '_' }, @@ -69,11 +77,7 @@ const DEFAULT_MATRIX: Case[] = [ { lifecycle: 'kill-mid-flight', conversation: 'hang' }, ]; -function sandboxFamilyKey(container: SandboxContainer): string { - return container.isProxy ? container.name.replace(/-proxy$/, '') : container.name; -} - -async function cleanupMatrixSandboxes(baselineSandboxIds: Set): Promise { +async function cleanupMatrixSandboxes(baselineSandboxIds: Set): Promise { const createdSandboxes = (await listSandboxContainers()).filter( container => !baselineSandboxIds.has(container.id) ); @@ -81,18 +85,24 @@ async function cleanupMatrixSandboxes(baselineSandboxIds: Set): Promise< for (const container of createdSandboxes) { const key = sandboxFamilyKey(container); const existing = sandboxFamilies.get(key); - if (!existing || (existing.isProxy && !container.isProxy)) { - sandboxFamilies.set(key, container); - } + if (!existing || (existing.isProxy && !container.isProxy)) sandboxFamilies.set(key, container); } + for (const sandbox of sandboxFamilies.values()) await removeSandboxFamily(sandbox); - for (const sandbox of sandboxFamilies.values()) { - await killSandboxFamily(sandbox); - const gone = await waitForSandboxFamilyGone(sandbox, 30_000); - if (!gone) { - console.warn(`smoke: sandbox family ${sandboxFamilyKey(sandbox)} remained after cleanup`); - } - } + const quiescent = await waitForSandboxCleanupQuiescence(baselineSandboxIds, { + timeoutMs: 30_000, + stableMs: 5_000, + reapPostBaseline: true, + }); + if (quiescent) return null; + + const remainingSandboxes = (await listSandboxContainers()).filter( + container => !baselineSandboxIds.has(container.id) + ); + const remaining = remainingSandboxes.map(container => container.name); + return remaining.length > 0 + ? `post-cleanup Docker state did not quiesce: ${remaining.join(', ')}` + : 'sandbox cleanup did not remain stable for 5s'; } async function main(): Promise { @@ -109,6 +119,7 @@ async function main(): Promise { nextAuthSecret: devVars.NEXTAUTH_SECRET ?? '', internalApiSecret: devVars.INTERNAL_API_SECRET, workerUrl: process.env.WORKER_URL ?? DEFAULT_CONFIG.workerUrl, + databaseUrl: process.env.DATABASE_URL, gitUrl: process.env.E2E_GIT_URL ?? DEFAULT_CONFIG.gitUrl, model: process.env.E2E_MODEL ?? DEFAULT_CONFIG.model, }; @@ -141,12 +152,54 @@ async function main(): Promise { continue; } console.log(`\n=== ${lifecycle}/${conversation} [api=${api}] ===`); + const rowSessionIds = new Set(); + const rowConfig: DriverConfig = { + ...config, + onSessionCreated: sessionId => rowSessionIds.add(sessionId), + }; + let cleanupFailure: LifecycleResult | null = null; try { - const result = await scenarioFn({ config, conversation, api }); + const result = await scenarioFn({ config: rowConfig, conversation, api }); printResult(result); results.push(result); } finally { - await cleanupMatrixSandboxes(baselineSandboxIds); + const cleanupStartedAt = Date.now(); + const cleanupErrors: string[] = []; + for (const sessionId of rowSessionIds) { + try { + await cleanupSessionUntilSettled(rowConfig, sessionId); + } catch (error) { + cleanupErrors.push( + `cleanup ${sessionId} threw: ${error instanceof Error ? error.message : String(error)}` + ); + } + } + try { + const message = await cleanupMatrixSandboxes(baselineSandboxIds); + if (message) cleanupErrors.push(message); + } catch (error) { + cleanupErrors.push( + `sandbox cleanup threw: ${error instanceof Error ? error.message : String(error)}` + ); + } + if (cleanupErrors.length > 0) { + cleanupFailure = { + name: 'matrix-cleanup', + conversation: `${lifecycle}/${conversation}`, + ok: false, + message: cleanupErrors.join('; '), + events: [], + durationMs: Date.now() - cleanupStartedAt, + }; + } + } + if (cleanupFailure) { + printResult(cleanupFailure); + results.push(cleanupFailure); + console.error( + 'smoke: stopping because the next row would inherit contaminated sandbox state' + ); + break; } } diff --git a/services/cloud-agent-next/test/integration/session/events.test.ts b/services/cloud-agent-next/test/integration/session/events.test.ts index 0633dcbb72..8c409f954d 100644 --- a/services/cloud-agent-next/test/integration/session/events.test.ts +++ b/services/cloud-agent-next/test/integration/session/events.test.ts @@ -136,6 +136,46 @@ describe('Event Storage', () => { expect(result.combined).toHaveLength(2); }); + it('finds entity prefixes literally without SQLite pattern matching', async () => { + const id = env.CLOUD_AGENT_SESSION.idFromName('user_1:literal_entity_prefix'); + const stub = env.CLOUD_AGENT_SESSION.get(id); + + const result = await runInDurableObject(stub, async (_instance, state) => { + const events = createEventQueries( + drizzle(state.storage, { logger: false }), + state.storage.sql + ); + const longPrefix = `long/${'x'.repeat(2048)}`; + const entityIds = [ + 'preparation/attempt/one', + 'literal%/one', + 'literalX/one', + 'literal_/one', + 'literalY/one', + `${longPrefix}/one`, + ]; + entityIds.forEach((entityId, index) => { + events.upsert({ + executionId: 'exc_prefix', + sessionId: 'sess_prefix', + streamEventType: 'preparing', + payload: JSON.stringify({ entityId }), + timestamp: index, + entityId, + }); + }); + + return { + normal: events.findByEntityPrefix('preparation/').map(event => event.timestamp), + percent: events.findByEntityPrefix('literal%/').map(event => event.timestamp), + underscore: events.findByEntityPrefix('literal_/').map(event => event.timestamp), + long: events.findByEntityPrefix(longPrefix).map(event => event.timestamp), + }; + }); + + expect(result).toEqual({ normal: [0], percent: [1], underscore: [3], long: [5] }); + }); + it('should delete events older than timestamp', async () => { const id = env.CLOUD_AGENT_SESSION.idFromName('user_1:sess_3'); const stub = env.CLOUD_AGENT_SESSION.get(id); diff --git a/services/cloud-agent-next/test/unit/client-session-tracking.test.ts b/services/cloud-agent-next/test/unit/client-session-tracking.test.ts new file mode 100644 index 0000000000..36a58ae434 --- /dev/null +++ b/services/cloud-agent-next/test/unit/client-session-tracking.test.ts @@ -0,0 +1,121 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; + +import { + cleanupSession, + cleanupSessionUntilSettled, + startSession, + type DriverConfig, + type StartSessionResult, +} from '../e2e/client.js'; + +function config(onSessionCreated: (sessionId: string) => void): DriverConfig { + return { + workerUrl: 'http://worker.test', + user: { + id: 'user_test', + email: 'test@example.com', + api_token_pepper: 'pepper', + }, + nextAuthSecret: 'test-secret', + gitUrl: 'https://example.com/repo.git', + model: 'kilo/fake-deterministic', + fakeLlmUrl: 'http://fake-llm.test', + onSessionCreated, + }; +} + +afterEach(() => vi.unstubAllGlobals()); + +describe('startSession row tracking', () => { + it('reports a successfully created session to the matrix cleanup hook', async () => { + const result: StartSessionResult = { + cloudAgentSessionId: 'agent_created', + kiloSessionId: 'ses_created', + messageId: 'msg_created', + delivery: 'queued', + }; + vi.stubGlobal( + 'fetch', + vi.fn().mockResolvedValue( + new Response(JSON.stringify({ result: { data: result } }), { + status: 200, + headers: { 'Content-Type': 'application/json' }, + }) + ) + ); + const onSessionCreated = vi.fn(); + + await expect(startSession(config(onSessionCreated), { prompt: 'hello' })).resolves.toEqual( + result + ); + expect(onSessionCreated).toHaveBeenCalledExactlyOnceWith('agent_created'); + }); + + it('does not report a session when creation fails', async () => { + vi.stubGlobal('fetch', vi.fn().mockResolvedValue(new Response('unavailable', { status: 503 }))); + const onSessionCreated = vi.fn(); + + await expect(startSession(config(onSessionCreated), { prompt: 'hello' })).rejects.toThrow( + 'tRPC start failed: 503' + ); + expect(onSessionCreated).not.toHaveBeenCalled(); + }); +}); + +describe('cleanupSession', () => { + it('uses the trusted cleanup endpoint with the internal API secret', async () => { + const fetchMock = vi.fn().mockResolvedValue( + new Response(JSON.stringify({ result: { data: { success: true } } }), { + status: 200, + headers: { 'Content-Type': 'application/json' }, + }) + ); + vi.stubGlobal('fetch', fetchMock); + const cleanupConfig = { ...config(vi.fn()), internalApiSecret: 'internal-secret' }; + + await expect(cleanupSession(cleanupConfig, 'agent_created')).resolves.toEqual({ + success: true, + }); + expect(fetchMock).toHaveBeenCalledWith( + new URL('http://worker.test/trpc/cleanupSession'), + expect.objectContaining({ + method: 'POST', + headers: expect.objectContaining({ 'x-internal-api-key': 'internal-secret' }), + body: JSON.stringify({ sessionId: 'agent_created' }), + }) + ); + }); + + it('polls a retryable physical-cleanup state until deletion settles', async () => { + const retryable = new Response(JSON.stringify({ error: { data: { retryable: true } } }), { + status: 500, + }); + const settled = new Response(JSON.stringify({ result: { data: { success: true } } }), { + status: 200, + headers: { 'Content-Type': 'application/json' }, + }); + const fetchMock = vi.fn().mockResolvedValueOnce(retryable).mockResolvedValueOnce(settled); + vi.stubGlobal('fetch', fetchMock); + const sleep = vi.fn().mockResolvedValue(undefined); + + await expect( + cleanupSessionUntilSettled( + { ...config(vi.fn()), internalApiSecret: 'internal-secret' }, + 'agent_created', + { sleep } + ) + ).resolves.toEqual({ success: true }); + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(sleep).toHaveBeenCalledWith(1_000); + }); + + it('fails before the request when the internal API secret is unavailable', async () => { + const fetchMock = vi.fn(); + vi.stubGlobal('fetch', fetchMock); + + await expect(cleanupSession(config(vi.fn()), 'agent_created')).rejects.toThrow( + 'cleanupSession requires INTERNAL_API_SECRET' + ); + expect(fetchMock).not.toHaveBeenCalled(); + }); +}); diff --git a/services/cloud-agent-next/test/unit/fake-llm-server.test.ts b/services/cloud-agent-next/test/unit/fake-llm-server.test.ts index c9dd5d8551..cd8e780c99 100644 --- a/services/cloud-agent-next/test/unit/fake-llm-server.test.ts +++ b/services/cloud-agent-next/test/unit/fake-llm-server.test.ts @@ -291,6 +291,15 @@ describe('fake-llm-server HTTP', () => { expect(body.error.type).toBe('insufficient_quota'); }); + it('model-missing scenario returns a terminal provider error', async () => { + const h = await start(); + const res = await postChat(h.url, '__fake__:model-missing'); + expect(res.status).toBe(404); + const body = (await res.json()) as { error: { message: string; type: string } }; + expect(body.error.message).toBe('Model not found: kilo/fake-deterministic'); + expect(body.error.type).toBe('model_not_found'); + }); + it('unknown scenario returns HTTP 402', async () => { const h = await start(); const res = await postChat(h.url, '__fake__:nosuch'); diff --git a/services/cloud-agent-next/test/unit/lifecycle-preparation.test.ts b/services/cloud-agent-next/test/unit/lifecycle-preparation.test.ts new file mode 100644 index 0000000000..da4f369a01 --- /dev/null +++ b/services/cloud-agent-next/test/unit/lifecycle-preparation.test.ts @@ -0,0 +1,56 @@ +import { describe, expect, it } from 'vitest'; + +import { unexpectedColdPreparationSteps } from '../e2e/lifecycle.js'; +import type { StreamEvent } from '../e2e/client.js'; + +function preparing(data: Record): StreamEvent { + return { + eventId: 0, + executionId: null, + sessionId: 'agent_test', + streamEventType: 'preparing', + timestamp: new Date(0).toISOString(), + data, + }; +} + +describe('unexpectedColdPreparationSteps', () => { + it('ignores snapshots, attempt events, and live warm verification steps', () => { + const events = [ + preparing({ version: 2, action: 'attempt_snapshot', step: 'workspace_setup' }), + preparing({ version: 2, action: 'step_snapshot', step: 'cloning' }), + preparing({ version: 2, action: 'attempt_started', step: 'workspace_setup' }), + preparing({ version: 2, action: 'step_started', step: 'sandbox_provision' }), + preparing({ version: 2, action: 'step_started', step: 'sandbox_boot' }), + preparing({ version: 2, action: 'step_started', step: 'kilo_server' }), + preparing({ version: 2, action: 'attempt_completed', step: 'ready' }), + ]; + + expect(unexpectedColdPreparationSteps(events)).toEqual([]); + }); + + it('reports unique live cold-path steps', () => { + const events = [ + preparing({ version: 2, action: 'step_started', step: 'workspace_setup' }), + preparing({ version: 2, action: 'step_started', step: 'cloning' }), + preparing({ version: 2, action: 'step_progress', step: 'cloning' }), + preparing({ version: 2, action: 'step_started', step: 'cloning' }), + preparing({ version: 2, action: 'step_started', step: 'kilo_session' }), + ]; + + expect(unexpectedColdPreparationSteps(events)).toEqual([ + 'workspace_setup', + 'cloning', + 'kilo_session', + ]); + }); + + it('fails closed for legacy and malformed live preparation events', () => { + const events = [ + preparing({ step: 'cloning', message: 'Cloning repository' }), + preparing({ version: 2, action: 'step_started' }), + ]; + + expect(unexpectedColdPreparationSteps(events)).toEqual(['legacy:cloning', 'unknown']); + }); +}); diff --git a/services/cloud-agent-next/test/unit/sandbox-control.test.ts b/services/cloud-agent-next/test/unit/sandbox-control.test.ts index 3c235c62b0..75fe2fa693 100644 --- a/services/cloud-agent-next/test/unit/sandbox-control.test.ts +++ b/services/cloud-agent-next/test/unit/sandbox-control.test.ts @@ -3,6 +3,9 @@ import { describe, expect, it, vi } from 'vitest'; import { killSandboxFamily, listSandboxesForAgentSession, + removeSandboxFamily, + waitForSandboxCleanupQuiescence, + waitForNewSandboxPresent, type DockerCommandExecutor, type SandboxContainer, } from '../e2e/sandbox-control.js'; @@ -41,6 +44,7 @@ function createDockerExecutor( return vi.fn(async args => { if (args[0] === 'ps') return { stdout: dockerPsOutput(containers) }; if (args[0] === 'kill') return { stdout: args[1] ?? '' }; + if (args[0] === 'rm') return { stdout: args[2] ?? '' }; if (args[0] === 'exec' && args[1] && markerContainerIds.has(args[1])) return { stdout: '' }; if (args[0] === 'exec') throw Object.assign(new Error('wrapper marker not found'), { code: 1 }); throw new Error(`Unexpected docker command: ${args.join(' ')}`); @@ -85,6 +89,19 @@ describe('listSandboxesForAgentSession', () => { }); }); +describe('waitForNewSandboxPresent', () => { + it('accepts a warm-pool container only when its wrapper marker owns the session', async () => { + const executeDocker = createDockerExecutor( + [unrelatedPrimary, ownedPrimary], + new Set([ownedPrimary.id]) + ); + + await expect( + waitForNewSandboxPresent(new Set([ownedPrimary.id]), 100, 'agent_owned', executeDocker) + ).resolves.toEqual(ownedPrimary); + }); +}); + describe('killSandboxFamily', () => { it('kills only the selected family exact primary and proxy containers', async () => { const similarlyNamedPrimary: SandboxContainer = { @@ -110,3 +127,132 @@ describe('killSandboxFamily', () => { expect(executeDocker).not.toHaveBeenCalledWith(['kill', similarlyNamedPrimary.id]); }); }); + +describe('removeSandboxFamily', () => { + it('force-removes only exact primary and proxy names, including stopped containers', async () => { + const similarlyNamedPrimary: SandboxContainer = { + id: 'similarly-named-primary-id', + name: `${ownedPrimary.name}-replacement`, + image: 'cloudflare/sandbox:latest', + isProxy: false, + }; + const executeDocker = createDockerExecutor([ + ownedPrimary, + ownedProxy, + unrelatedPrimary, + similarlyNamedPrimary, + ]); + + await expect(removeSandboxFamily(ownedPrimary, executeDocker)).resolves.toEqual([ + ownedPrimary.name, + ownedProxy.name, + ]); + expect(executeDocker).toHaveBeenCalledWith([ + 'ps', + '-a', + '--format', + '{{.ID}}\t{{.Names}}\t{{.Image}}', + ]); + expect(executeDocker).toHaveBeenCalledWith(['rm', '-f', ownedPrimary.id]); + expect(executeDocker).toHaveBeenCalledWith(['rm', '-f', ownedProxy.id]); + expect(executeDocker).not.toHaveBeenCalledWith(['rm', '-f', unrelatedPrimary.id]); + expect(executeDocker).not.toHaveBeenCalledWith(['rm', '-f', similarlyNamedPrimary.id]); + }); +}); + +function createSequencedDockerExecutor(observations: SandboxContainer[][]): DockerCommandExecutor { + let index = 0; + return vi.fn(async args => { + if (args[0] === 'ps') { + const containers = observations[Math.min(index, observations.length - 1)] ?? []; + index += 1; + return { stdout: dockerPsOutput(containers) }; + } + if (args[0] === 'kill' || args[0] === 'rm') return { stdout: args.at(-1) ?? '' }; + throw new Error(`Unexpected docker command: ${args.join(' ')}`); + }); +} + +function controlledTime(): { now: () => number; sleep: (ms: number) => Promise } { + let timestamp = 0; + return { + now: () => timestamp, + sleep: async ms => { + timestamp += ms; + }, + }; +} + +describe('waitForSandboxCleanupQuiescence', () => { + it('ignores baseline containers and requires an uninterrupted absence window', async () => { + const time = controlledTime(); + const executeDocker = createSequencedDockerExecutor([[unrelatedPrimary]]); + + await expect( + waitForSandboxCleanupQuiescence(new Set([unrelatedPrimary.id]), { + timeoutMs: 2_000, + stableMs: 1_000, + pollIntervalMs: 250, + executeDocker, + ...time, + }) + ).resolves.toBe(true); + expect(time.now()).toBe(1_000); + }); + + it('resets the absence window when a post-baseline sandbox transiently reappears', async () => { + const time = controlledTime(); + const executeDocker = createSequencedDockerExecutor([ + [unrelatedPrimary], + [unrelatedPrimary, ownedPrimary], + [unrelatedPrimary], + ]); + + await expect( + waitForSandboxCleanupQuiescence(new Set([unrelatedPrimary.id]), { + timeoutMs: 2_000, + stableMs: 500, + reapPostBaseline: true, + pollIntervalMs: 250, + executeDocker, + ...time, + }) + ).resolves.toBe(true); + expect(time.now()).toBe(1_000); + expect(executeDocker).toHaveBeenCalledWith(['kill', ownedPrimary.id]); + expect(executeDocker).toHaveBeenCalledWith(['rm', '-f', ownedPrimary.id]); + }); + + it('does not bypass runtime teardown while waiting on the normal path', async () => { + const time = controlledTime(); + const executeDocker = createSequencedDockerExecutor([[ownedPrimary], []]); + + await expect( + waitForSandboxCleanupQuiescence(new Set(), { + timeoutMs: 1_000, + stableMs: 500, + pollIntervalMs: 250, + executeDocker, + ...time, + }) + ).resolves.toBe(true); + expect(executeDocker).not.toHaveBeenCalledWith(['kill', ownedPrimary.id]); + expect(executeDocker).not.toHaveBeenCalledWith(['rm', '-f', ownedPrimary.id]); + }); + + it('times out while a post-baseline sandbox remains present', async () => { + const time = controlledTime(); + const executeDocker = createSequencedDockerExecutor([[ownedPrimary]]); + + await expect( + waitForSandboxCleanupQuiescence(new Set(), { + timeoutMs: 1_000, + stableMs: 500, + pollIntervalMs: 250, + executeDocker, + ...time, + }) + ).resolves.toBe(false); + expect(time.now()).toBe(1_000); + }); +}); diff --git a/services/cloud-agent-next/wrangler.jsonc b/services/cloud-agent-next/wrangler.jsonc index 5361132883..038412a3b9 100644 --- a/services/cloud-agent-next/wrangler.jsonc +++ b/services/cloud-agent-next/wrangler.jsonc @@ -468,7 +468,7 @@ "image_vars": { "KILOCODE_CLI_VERSION": "7.3.54", }, - "max_instances": 2, + "max_instances": 16, "rollout_active_grace_period": 60, }, { diff --git a/services/cloud-agent-next/wrapper/src/server.test.ts b/services/cloud-agent-next/wrapper/src/server.test.ts index acdd135cdf..142b979968 100644 --- a/services/cloud-agent-next/wrapper/src/server.test.ts +++ b/services/cloud-agent-next/wrapper/src/server.test.ts @@ -1,5 +1,6 @@ import { afterEach, describe, expect, it } from 'bun:test'; import { createServer as createNetServer } from 'node:net'; +import { readFileSync, rmSync } from 'node:fs'; import { WrapperState } from './state'; import { bindSessionContext, @@ -45,6 +46,7 @@ function createTestFetch(overrides?: { deleteCalls?: string[]; runtimeEnvironmentUpdates?: Array>; resizeError?: Error; + toolCgroupHealth?: () => never; }) { const ptyCalls = overrides?.ptyCalls ?? []; const resizeCalls = overrides?.resizeCalls ?? []; @@ -98,6 +100,7 @@ function createTestFetch(overrides?: { updateRuntimeEnvironment: async env => { runtimeEnvironmentUpdates.push(env); }, + ...(overrides?.toolCgroupHealth ? { toolCgroupHealth: overrides.toolCgroupHealth } : {}), }, () => {} ); @@ -338,6 +341,44 @@ describe('wrapper health', () => { wrapperInstanceGeneration: 8, }); }); + + it('returns a bounded correlated error when a health dependency throws', async () => { + const previousLogPath = process.env.WRAPPER_LOG_PATH; + const logPath = `/tmp/wrapper-health-${crypto.randomUUID()}.log`; + process.env.WRAPPER_LOG_PATH = logPath; + const localError = 'tool cgroup probe failed locally'; + + try { + const { fetchHandler } = createTestFetch({ + toolCgroupHealth: () => { + throw new Error(localError); + }, + }); + const requestId = 'health-request-correlation-123'; + const response = await fetchHandler( + new Request('http://wrapper.test/health', { + headers: { 'x-kilo-request-id': requestId }, + }) + ); + if (!response) throw new Error('Expected health response'); + + expect(response.status).toBe(500); + const body = await response.json(); + expect(body).toEqual({ + error: 'HEALTH_CHECK_FAILED', + message: 'Wrapper health check failed', + requestId, + }); + expect(JSON.stringify(body)).not.toContain(localError); + const log = readFileSync(logPath, 'utf8'); + expect(log).toContain(`HTTP GET /health requestId=${requestId}`); + expect(log).toContain(`requestId=${requestId} path=/health error=${localError}`); + } finally { + if (previousLogPath === undefined) delete process.env.WRAPPER_LOG_PATH; + else process.env.WRAPPER_LOG_PATH = previousLogPath; + rmSync(logPath, { force: true }); + } + }); }); describe('wrapper PTY routes', () => { diff --git a/services/cloud-agent-next/wrapper/src/server.ts b/services/cloud-agent-next/wrapper/src/server.ts index c933657018..f673571fb8 100644 --- a/services/cloud-agent-next/wrapper/src/server.ts +++ b/services/cloud-agent-next/wrapper/src/server.ts @@ -35,6 +35,10 @@ import { createProxyRequest } from '../../src/shared/http-proxy.js'; import { PNPM_STORE_DIR, PNPM_STORE_ENV_VAR } from '../../src/shared/runtime-environment.js'; import type { SessionBoundFeedPolicy } from './global-feed-manager.js'; import type { ToolCgroupHealth } from './tool-cgroup.js'; +import { + WRAPPER_HEALTH_CHECK_FAILED, + WRAPPER_REQUEST_ID_HEADER, +} from '../../src/shared/wrapper-http.js'; // --------------------------------------------------------------------------- // Types @@ -173,6 +177,22 @@ function errorResponse(error: string, message: string, status: number): Response return jsonResponse({ error, message }, status); } +function healthErrorResponse(requestId: string): Response { + return jsonResponse( + { + error: WRAPPER_HEALTH_CHECK_FAILED, + message: 'Wrapper health check failed', + requestId, + }, + 500 + ); +} + +function requestCorrelationId(req: Request): string { + const supplied = req.headers.get(WRAPPER_REQUEST_ID_HEADER); + return supplied && /^[A-Za-z0-9_-]{1,128}$/.test(supplied) ? supplied : crypto.randomUUID(); +} + function wrapperFinalizingResponse(state: WrapperState): Response { return jsonResponse( { @@ -1183,8 +1203,9 @@ export function createFetchHandler( const url = new URL(req.url); const method = req.method; const path = url.pathname; + const requestId = requestCorrelationId(req); - logToFile(`HTTP ${method} ${path}`); + logToFile(`HTTP ${method} ${path} requestId=${requestId}`); const ptyPath = parsePtyPath(path); if (ptyPath) { @@ -1233,12 +1254,14 @@ export function createFetchHandler( try { return Promise.resolve(handler(req)).catch(error => { const msg = error instanceof Error ? error.message : String(error); - logToFile(`HTTP handler error: ${msg}`); + logToFile(`HTTP handler error: requestId=${requestId} path=${path} error=${msg}`); + if (path === '/health') return healthErrorResponse(requestId); return errorResponse('INTERNAL_ERROR', msg, 500); }); } catch (error) { const msg = error instanceof Error ? error.message : String(error); - logToFile(`HTTP handler error: ${msg}`); + logToFile(`HTTP handler error: requestId=${requestId} path=${path} error=${msg}`); + if (path === '/health') return healthErrorResponse(requestId); return errorResponse('INTERNAL_ERROR', msg, 500); } };