From a1762eba4942fdb596c497827a98e68f5683c39b Mon Sep 17 00:00:00 2001 From: Evgeny Shurakov Date: Fri, 11 Sep 2026 08:50:04 +0200 Subject: [PATCH 1/2] feat(cloud-agent-next): report pre-dispatch failures durably A message admitted through the control-plane API could fail during preparation, attachment, or dispatch without ever appearing in cloud_agent_session_runs, because only the legacy DO emitted run.state reports and its emitter swallowed the send. Operators could not identify the accepted session/message that had failed before the wrapper started. Add a durable, alarm-driven report obligation to the control-plane SandboxSession DO and commit it in the same transaction as admission, the accepted/terminal common commit, and deletion snapshots. The obligation carries the latest cumulative lifecycle snapshot and is delivered through the existing report queue; a bounded outbox handles retry, capacity, and expiry without changing dispatch or alarm timing or failing a healthy run. Failures are classified from bounded cause facts at the committing transition, observed acceptance is never fabricated, and a trusted first-admission anchor lets the consumer create the reporting parent for empty/worktree chats. Report snapshots and alarms are preserved through both deletion purge paths. Extend focused queue/lifecycle, report-store, report-consumer, and control-plane failure tests, plus a real DO reporting integration test. --- .../src/cloud-agent-queue-report.test.ts | 47 ++ .../src/cloud-agent-queue-report.ts | 22 +- .../src/sandbox-session/SandboxSession.ts | 250 +++++++++- .../src/sandbox-session/report-outbox.test.ts | 281 +++++++++++ .../src/sandbox-session/report-outbox.ts | 267 +++++++++++ .../session-message-queue.test.ts | 1 + .../sandbox-session/session-message-queue.ts | 1 + .../src/sandbox-session/terminal-lifecycle.ts | 11 +- .../telemetry/control-plane-failure.test.ts | 77 ++++ .../src/telemetry/control-plane-failure.ts | 87 ++++ .../src/telemetry/queue-reports.ts | 95 ++-- .../src/telemetry/report-consumer.test.ts | 22 + .../src/telemetry/report-consumer.ts | 8 + .../src/telemetry/report-store.test.ts | 107 +++++ .../src/telemetry/report-store.ts | 96 +++- .../test/integration/sandbox-control.test.ts | 43 +- .../sandbox-session-reports.test.ts | 435 ++++++++++++++++++ .../integration/worktree-deletion.test.ts | 34 +- 18 files changed, 1805 insertions(+), 79 deletions(-) create mode 100644 services/cloud-agent-next/src/sandbox-session/report-outbox.test.ts create mode 100644 services/cloud-agent-next/src/sandbox-session/report-outbox.ts create mode 100644 services/cloud-agent-next/src/telemetry/control-plane-failure.test.ts create mode 100644 services/cloud-agent-next/src/telemetry/control-plane-failure.ts create mode 100644 services/cloud-agent-next/test/integration/sandbox-session-reports.test.ts diff --git a/packages/worker-utils/src/cloud-agent-queue-report.test.ts b/packages/worker-utils/src/cloud-agent-queue-report.test.ts index f6b3506115..610cd4b0b7 100644 --- a/packages/worker-utils/src/cloud-agent-queue-report.test.ts +++ b/packages/worker-utils/src/cloud-agent-queue-report.test.ts @@ -242,4 +242,51 @@ describe('CloudAgentQueueReportSchema', () => { ).success ).toBe(false); }); + + it('round-trips a complete reporting anchor for a session without an initial turn', () => { + const parsed = CloudAgentQueueReportSchema.parse({ + ...reportWithRun({ status: 'queued', queuedAt: '2026-05-26T08:01:00.000Z' }), + session: { + cloudAgentSessionId: 'agent_reporting_session', + kiloSessionId: 'ses_12345678901234567890123456', + initialMessageId: 'msg_anchor_first', + reportingCreatedAt: '2026-05-26T07:59:00.000Z', + }, + }); + + expect(parsed.session).toEqual({ + cloudAgentSessionId: 'agent_reporting_session', + kiloSessionId: 'ses_12345678901234567890123456', + initialMessageId: 'msg_anchor_first', + reportingCreatedAt: '2026-05-26T07:59:00.000Z', + }); + }); + + it('keeps the legacy anchor-free session shape valid', () => { + expect( + CloudAgentQueueReportSchema.safeParse({ + ...reportWithRun({ status: 'queued' }), + session: { cloudAgentSessionId: 'agent_reporting_session' }, + }).success + ).toBe(true); + }); + + it('rejects a partial reporting anchor', () => { + for (const partial of [ + { kiloSessionId: 'ses_12345678901234567890123456' }, + { initialMessageId: 'msg_anchor_first' }, + { reportingCreatedAt: '2026-05-26T07:59:00.000Z' }, + { + kiloSessionId: 'ses_12345678901234567890123456', + initialMessageId: 'msg_anchor_first', + }, + ]) { + expect( + CloudAgentQueueReportSchema.safeParse({ + ...reportWithRun({ status: 'queued' }), + session: { cloudAgentSessionId: 'agent_reporting_session', ...partial }, + }).success + ).toBe(false); + } + }); }); diff --git a/packages/worker-utils/src/cloud-agent-queue-report.ts b/packages/worker-utils/src/cloud-agent-queue-report.ts index a0d44e5bc1..1c94665950 100644 --- a/packages/worker-utils/src/cloud-agent-queue-report.ts +++ b/packages/worker-utils/src/cloud-agent-queue-report.ts @@ -51,6 +51,7 @@ export const DIAGNOSTIC_RETENTION_MS = 30 * 24 * 60 * 60 * 1000; const IsoTimestampSchema = z.string().datetime({ offset: true }); const OperationalIdentifierSchema = z.string().min(1).max(MAX_OPERATIONAL_IDENTIFIER_LENGTH); +const kiloSessionIdSchema = z.string().startsWith('ses_').length(30); const WrapperRunIdentifierSchema = OperationalIdentifierSchema.regex(/^wr_[A-Za-z0-9_-]+$/); const validFailureClassifications = new Set( CloudAgentRunFailureClassifications.map( @@ -59,8 +60,25 @@ const validFailureClassifications = new Set( ); const CloudAgentQueueSessionIdentitySchema = z - .object({ cloudAgentSessionId: OperationalIdentifierSchema }) - .strict(); + .object({ + cloudAgentSessionId: OperationalIdentifierSchema, + kiloSessionId: kiloSessionIdSchema.optional(), + initialMessageId: OperationalIdentifierSchema.optional(), + reportingCreatedAt: IsoTimestampSchema.optional(), + }) + .strict() + .superRefine((session, ctx) => { + const present = [session.kiloSessionId, session.initialMessageId, session.reportingCreatedAt].filter( + value => value !== undefined + ).length; + if (present !== 0 && present !== 3) { + ctx.addIssue({ + code: 'custom', + message: 'Reporting anchor fields must be provided together', + path: ['kiloSessionId'], + }); + } + }); const CloudAgentFailedRunDiagnosticSchema = z .object({ diff --git a/services/cloud-agent-next/src/sandbox-session/SandboxSession.ts b/services/cloud-agent-next/src/sandbox-session/SandboxSession.ts index 8a114ee4b2..b7f4b984ce 100644 --- a/services/cloud-agent-next/src/sandbox-session/SandboxSession.ts +++ b/services/cloud-agent-next/src/sandbox-session/SandboxSession.ts @@ -188,6 +188,25 @@ import { type SessionMessageRecord, } from './session-message-queue.js'; import { createMessageCallbacks, type MessageCallbacks } from './message-callbacks.js'; +import { + createReportOutbox, + readReportAnchor, + writeReportAnchor, + type ReportAnchor, + type ReportOutbox, +} from './report-outbox.js'; +import { + CloudAgentQueueReportSchema, + DIAGNOSTIC_RETENTION_MS, + type CloudAgentQueueReport, + type CloudAgentRunStateReport, +} from '@kilocode/worker-utils/cloud-agent-queue-report'; +import { + buildRunStateReport, + FAILED_RUN_DIAGNOSTIC_MESSAGES, +} from '../telemetry/queue-reports.js'; +import { classifyControlPlaneFailure } from '../telemetry/control-plane-failure.js'; +import { classifyCloudAgentFailure } from '@kilocode/worker-utils/cloud-agent-failure'; import { PENDING_SESSION_MESSAGE_LIMIT } from '../session/pending-messages.js'; import { commitSessionOperationResult, @@ -292,6 +311,7 @@ export class SandboxSession extends DurableObject { private readonly sessionId: SessionId | undefined; private readonly eventQueries: EventQueries; private readonly messageCallbacks: MessageCallbacks; + private readonly reportOutbox: ReportOutbox; private readonly terminalLifecycle: ReturnType; private readonly terminalBridge: ReturnType; private readonly dispatches = new Map>(); @@ -304,6 +324,7 @@ export class SandboxSession extends DurableObject { private readonly worktreeChanges: ReturnType; private readonly interactionRefresh: InteractionRefresh; private callbackRepairRequired = false; + private reportRepairRequired = false; constructor(ctx: DurableObjectState, env: Env) { super(ctx, env); @@ -347,6 +368,10 @@ export class SandboxSession extends DurableObject { parentMessageId ), }); + this.reportOutbox = createReportOutbox({ + storage: ctx.storage, + getQueue: () => env.CLOUD_AGENT_REPORT_QUEUE, + }); this.worktreeChanges = createWorktreeChanges({ storage: ctx.storage, saveSnapshotEvent: snapshot => { @@ -1669,9 +1694,11 @@ export class SandboxSession extends DurableObject { this.snapshotDeletedMessages(metadata); }); if (this.messageCallbacks.pendingCallbackCount() > 0) this.scheduleCallbackRepair(); + if (this.reportOutbox.pendingCount() > 0) this.scheduleReportRepair(); this.deletedWorktreeId = worktreeId; for (const socket of this.ctx.getWebSockets()) socket.close(1001, 'Worktree deleted'); - if (this.messageCallbacks.pendingCallbackCount() === 0) await this.ctx.storage.deleteAlarm(); + if (this.messageCallbacks.pendingCallbackCount() === 0 && this.reportOutbox.pendingCount() === 0) + await this.ctx.storage.deleteAlarm(); if (!metadata) return null; return cloudAgentWorktreeLocationSchema.parse({ sandboxId: metadata.workspace?.sandboxId, @@ -1743,7 +1770,8 @@ export class SandboxSession extends DurableObject { } for (const socket of this.ctx.getWebSockets()) socket.close(1001, 'Worktree deleted'); const callbacksPending = this.messageCallbacks.pendingCallbackCount() > 0; - if (!callbacksPending) await this.ctx.storage.deleteAlarm(); + const reportsPending = this.reportOutbox.pendingCount() > 0; + if (!callbacksPending && !reportsPending) await this.ctx.storage.deleteAlarm(); const db = drizzle(this.ctx.storage, { logger: false }); this.ctx.storage.transactionSync(() => { db.delete(events).where(eq(events.session_id, this.requireSessionId())).run(); @@ -1753,8 +1781,13 @@ export class SandboxSession extends DurableObject { this.ctx.storage.kv.put(DELETED_WORKTREE_KEY, worktreeId); this.ctx.storage.kv.put(DELETION_COMPLETED_KEY, true); }); - if (callbacksPending) { - await this.armQueueRetry(this.messageCallbacks.nextCallbackDueAt() ?? Date.now()); + if (callbacksPending || reportsPending) { + await this.armQueueRetry( + Math.min( + this.messageCallbacks.nextCallbackDueAt() ?? Date.now(), + this.reportOutbox.nextDueAt() ?? Date.now() + ) + ); } } @@ -1763,12 +1796,151 @@ export class SandboxSession extends DurableObject { return operation.finally(() => this.activeOperations.delete(operation)); } + private ensureReportAnchor( + metadata: SessionMetadata, + firstMessageId: string, + isFirstMessage: boolean + ): ReportAnchor | undefined { + const existing = readReportAnchor(this.ctx.storage); + if (existing) return existing; + // Never fabricate a first message or creation time for a session that + // already had messages; the anchor may only be written by the first admission. + if (!isFirstMessage) return undefined; + // Public `start` already has a reporting parent created by registration. + if (metadata.initialMessage?.id !== undefined) return undefined; + const kiloSessionId = metadata.auth.kiloSessionId; + if (kiloSessionId === undefined || !/^ses_.{26}$/.test(kiloSessionId)) return undefined; + const createdAt = Date.now(); + const anchor: ReportAnchor = { + version: 1, + kiloSessionId, + initialMessageId: firstMessageId, + createdAt, + }; + writeReportAnchor(this.ctx.storage, { kiloSessionId, initialMessageId: firstMessageId, createdAt }); + return anchor; + } + + private buildMessageReport( + message: SessionMessageRecord, + anchor: ReportAnchor | undefined, + acceptanceObserved: boolean + ): CloudAgentQueueReport | undefined { + const sessionId = this.sessionId; + if (!sessionId) return undefined; + const status: CloudAgentRunStateReport['run']['status'] = + message.state === 'cancelled' ? 'interrupted' : message.state; + // `applyMessageOutcome` fills an inferred `acceptedAt` for a terminal + // outcome that arrived before the ACK. Only a transition out of the + // accepted state is observable dispatch acceptance. + const dispatchAcceptedAt = + acceptanceObserved && message.acceptedAt !== undefined ? message.acceptedAt : undefined; + const run: CloudAgentRunStateReport['run'] = { + messageId: message.messageId, + status, + ...(message.queuedAt === undefined + ? {} + : { queuedAt: new Date(message.queuedAt).toISOString() }), + ...(dispatchAcceptedAt === undefined + ? {} + : { dispatchAcceptedAt: new Date(dispatchAcceptedAt).toISOString() }), + ...(message.terminalAt === undefined + ? {} + : { terminalAt: new Date(message.terminalAt).toISOString() }), + }; + if (status === 'failed' || status === 'interrupted') { + const dispatchState = acceptanceObserved ? ('accepted' as const) : ('pre_dispatch' as const); + // Coordinator failures carry a bounded cause. Wrapper outcomes and + // operation results copy arbitrary text, so they are not treated as a + // known coordinator cause. + const coordinatorOriginated = + message.terminalSource === undefined || message.terminalSource === 'coordinator'; + const classification = classifyControlPlaneFailure( + coordinatorOriginated ? message.failedReason : undefined, + dispatchState, + status + ); + run.failureStage = classification.stage; + run.failureCode = classification.code; + if (status === 'failed') { + const orchestrator = classifyCloudAgentFailure({ + source: 'run', + stage: classification.stage, + code: classification.code, + }); + run.failureResponsibility = orchestrator.responsibility; + run.failureReason = orchestrator.reason; + if (message.terminalAt !== undefined) { + run.diagnostic = { + errorMessageRedacted: + FAILED_RUN_DIAGNOSTIC_MESSAGES[classification.code] ?? + 'Run failed without a classified cause', + errorExpiresAt: new Date( + message.terminalAt + DIAGNOSTIC_RETENTION_MS + ).toISOString(), + }; + } + } + } + const report = buildRunStateReport({ + cloudAgentSessionId: sessionId, + ...(anchor === undefined + ? {} + : { + anchor: { + kiloSessionId: anchor.kiloSessionId, + initialMessageId: anchor.initialMessageId, + reportingCreatedAt: new Date(anchor.createdAt).toISOString(), + }, + }), + run, + occurredAt: Date.now(), + }); + const parsed = CloudAgentQueueReportSchema.safeParse(report); + if (!parsed.success) { + logger + .withFields({ sessionId: this.sessionId, messageId: message.messageId, status }) + .error('Invalid Cloud Agent report snapshot aborts the lifecycle commit'); + return undefined; + } + return parsed.data; + } + + /** + * Persists the local report obligation for a committed transition. This runs + * inside the caller's state-write transaction: an unbuildable snapshot or a + * failed KV write must abort it rather than be swallowed, so message state + * and obligation commit atomically. Transport/downstream failures remain + * asynchronous and nonfatal in the outbox repair path. + */ + private recordMessageReport(message: SessionMessageRecord, acceptanceObserved: boolean): void { + const report = this.buildMessageReport( + message, + readReportAnchor(this.ctx.storage), + acceptanceObserved + ); + if (!report) { + throw new Error( + `Could not build Cloud Agent report obligation for message ${message.messageId} (${message.state})` + ); + } + this.reportOutbox.record(report); + } + private snapshotDeletedMessages(metadata: SessionMetadata | null): void { const messages = this.ctx.storage.kv.get(MESSAGES_KEY) ?? []; + const now = Date.now(); const cancelled = messages.map(message => { if (message.state !== 'queued' && message.state !== 'accepted') return message; - const next = { ...message, state: 'cancelled' as const }; + const acceptanceObserved = message.state === 'accepted'; + const next = { + ...message, + state: 'cancelled' as const, + terminalAt: now, + terminalSource: 'coordinator' as const, + }; this.messageCallbacks.persistTerminalCallback(next, metadata); + this.recordMessageReport(next, acceptanceObserved); return next; }); this.ctx.storage.kv.put(MESSAGES_KEY, cancelled); @@ -1822,6 +1994,7 @@ export class SandboxSession extends DurableObject { return records; }); if (this.messageCallbacks.pendingCallbackCount() > 0) this.scheduleCallbackRepair(); + if (this.reportOutbox.pendingCount() > 0) this.scheduleReportRepair(); for (const ws of this.ctx.getWebSockets('stream')) { ws.close(1000, 'session access revoked'); } @@ -1831,7 +2004,8 @@ export class SandboxSession extends DurableObject { await this.ingestPublicationChain.catch(() => undefined); if (this.deletedWorktreeId) throw new Error('worktree_deleting'); const callbacksPending = this.messageCallbacks.pendingCallbackCount() > 0; - if (!callbacksPending) await this.ctx.storage.deleteAlarm(); + const reportsPending = this.reportOutbox.pendingCount() > 0; + if (!callbacksPending && !reportsPending) await this.ctx.storage.deleteAlarm(); this.ctx.storage.transactionSync(() => { if (this.deletedWorktreeId) throw new Error('worktree_deleting'); const pendingCleanup = this.pendingRuntimeCleanup(); @@ -1840,8 +2014,13 @@ export class SandboxSession extends DurableObject { if (pendingCleanup) this.ctx.storage.kv.put(PENDING_RUNTIME_CLEANUP_KEY, pendingCleanup); }); if (this.pendingRuntimeCleanup()) await this.armQueueRetry(); - else if (callbacksPending) - await this.armQueueRetry(this.messageCallbacks.nextCallbackDueAt() ?? Date.now()); + else if (callbacksPending || reportsPending) + await this.armQueueRetry( + Math.min( + this.messageCallbacks.nextCallbackDueAt() ?? Date.now(), + this.reportOutbox.nextDueAt() ?? Date.now() + ) + ); } async registerSession(input: SandboxSessionRegistrationInput): Promise { @@ -2163,8 +2342,13 @@ export class SandboxSession extends DurableObject { for (const stop of this.stopLifecycle.pending()) void this.scheduleStopProgress(stop.request.operationId); await this.messageCallbacks.repair(); + await this.reportOutbox.repair(); const callbackDueAt = this.messageCallbacks.nextCallbackDueAt(); - if (callbackDueAt !== undefined) await this.armQueueRetry(callbackDueAt); + const reportDueAt = this.reportOutbox.nextDueAt(); + if (callbackDueAt !== undefined || reportDueAt !== undefined) + await this.armQueueRetry( + Math.min(callbackDueAt ?? Number.MAX_SAFE_INTEGER, reportDueAt ?? Number.MAX_SAFE_INTEGER) + ); const epoch = this.terminalLifecycle.captureEpoch(); if (epoch === null || this.deletedWorktreeId) return; const now = Date.now(); @@ -2387,7 +2571,11 @@ export class SandboxSession extends DurableObject { latestMetadata.agent, latestMetadata.workspace?.worktreeId ? latestMetadata.finalization : undefined ); - nextMessages.push(createSessionMessageRecord(intent)); + const queuedMessage: SessionMessageRecord = { + ...createSessionMessageRecord(intent), + queuedAt: Date.now(), + }; + nextMessages.push(queuedMessage); const nextMetadata = intent.agent.model === undefined ? null @@ -2399,6 +2587,10 @@ export class SandboxSession extends DurableObject { if (nextMetadata) { this.ctx.storage.kv.put(METADATA_KEY, serializeSessionMetadata(nextMetadata)); } + // Local obligation persistence is part of this transaction: a throw here + // rolls back the message/metadata writes and fails admission. + this.ensureReportAnchor(latestMetadata, messageId, latestMessages.length === 0); + this.recordMessageReport(queuedMessage, false); admitted = true; return { success: true, outcome: 'queued', messageId, compatibilityDelivery: 'queued' }; }); @@ -3274,25 +3466,44 @@ export class SandboxSession extends DurableObject { this.ctx.waitUntil(this.messageCallbacks.repair()); } + private scheduleReportRepair(): void { + this.ctx.waitUntil(this.armQueueRetry()); + this.ctx.waitUntil(this.reportOutbox.repair()); + } + private scheduleCallbackRepairIfRequired(): void { - if (!this.callbackRepairRequired) return; - this.callbackRepairRequired = false; - this.scheduleCallbackRepair(); + if (this.callbackRepairRequired) { + this.callbackRepairRequired = false; + this.scheduleCallbackRepair(); + } + if (this.reportRepairRequired) { + this.reportRepairRequired = false; + this.scheduleReportRepair(); + } } private async armQueueRetry(when = Date.now() + QUEUE_RETRY_MS): Promise { const epoch = this.terminalLifecycle.captureEpoch(); const hasPendingStop = this.stopLifecycle.pending().length > 0; const callbackDueAt = this.messageCallbacks.nextCallbackDueAt(); + const reportDueAt = this.reportOutbox.nextDueAt(); const hasPendingCallbacks = callbackDueAt !== undefined; - if (epoch === null && !this.pendingRuntimeCleanup() && !hasPendingStop && !hasPendingCallbacks) + const hasPendingReports = reportDueAt !== undefined; + if ( + epoch === null && + !this.pendingRuntimeCleanup() && + !hasPendingStop && + !hasPendingCallbacks && + !hasPendingReports + ) return; const existing = await this.ctx.storage.getAlarm(); if ( (epoch === null || !this.terminalLifecycle.isCurrent(epoch)) && !this.pendingRuntimeCleanup() && !hasPendingStop && - !hasPendingCallbacks + !hasPendingCallbacks && + !hasPendingReports ) return; const requested = Math.min(when, callbackDueAt ?? Number.MAX_SAFE_INTEGER); @@ -3968,6 +4179,7 @@ export class SandboxSession extends DurableObject { const events: StoredEvent[] = []; const committed: ControlDiagnosticFields[] = []; let callbackPersisted = false; + let reportPersisted = false; let persisted = false; let disposition: ControlEventDisposition = 'epoch_changed'; const write = () => { @@ -3989,6 +4201,8 @@ export class SandboxSession extends DurableObject { if (previous?.state !== 'accepted') { const event = this.persistMessageLifecycleEvent(message); if (event) events.push(event); + this.recordMessageReport(message, true); + reportPersisted = true; committed.push({ messageId: message.messageId, wrapperInstanceId: message.wrapperInstanceId, @@ -4015,6 +4229,8 @@ export class SandboxSession extends DurableObject { const event = this.persistMessageLifecycleEvent(terminal); if (event) events.push(event); if (this.messageCallbacks.persistTerminalCallback(terminal)) callbackPersisted = true; + this.recordMessageReport(terminal, previous?.state === 'accepted'); + reportPersisted = true; committed.push({ messageId: terminal.messageId, wrapperInstanceId: terminal.wrapperInstanceId, @@ -4055,6 +4271,10 @@ export class SandboxSession extends DurableObject { if (scheduleCallbackRepair) this.scheduleCallbackRepair(); else this.callbackRepairRequired = true; } + if (reportPersisted) { + if (scheduleCallbackRepair) this.scheduleReportRepair(); + else this.reportRepairRequired = true; + } for (const fields of committed) { logControlDiagnostic('session_message_committed', { sessionId: this.sessionId, diff --git a/services/cloud-agent-next/src/sandbox-session/report-outbox.test.ts b/services/cloud-agent-next/src/sandbox-session/report-outbox.test.ts new file mode 100644 index 0000000000..aee939319f --- /dev/null +++ b/services/cloud-agent-next/src/sandbox-session/report-outbox.test.ts @@ -0,0 +1,281 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import type { CloudAgentQueueReport } from '@kilocode/worker-utils/cloud-agent-queue-report'; +import { + REPORT_ENQUEUE_MAX_ATTEMPTS, + REPORT_ENQUEUE_RETRY_MS, + REPORT_OUTBOX_MAX_ENTRIES, + REPORT_OUTBOX_PREFIX, + createReportOutbox, + parsePendingRunReport, + reportOutboxKey, +} from './report-outbox.js'; + +afterEach(() => { + vi.restoreAllMocks(); +}); + +type MemoryKv = { + get(key: string): T | undefined; + put(key: string, value: T): void; + delete(key: string): boolean; + list(options?: { prefix?: string }): Iterable<[string, T]>; +}; + +function memoryKv(): MemoryKv { + const values = new Map(); + return { + get: (key: string) => structuredClone(values.get(key)) as T | undefined, + put: (key: string, value: T) => values.set(key, structuredClone(value)), + delete: key => values.delete(key), + list: (options?: { prefix?: string }) => + [...values.entries()] + .filter(([key]) => options?.prefix === undefined || key.startsWith(options.prefix)) + .map(([key, value]) => [key, structuredClone(value) as T] as [string, T]), + }; +} + +function queuedReport(messageId: string): CloudAgentQueueReport { + return { + version: 1, + type: 'run.state', + occurredAt: '2026-05-26T08:00:00.000Z', + session: { cloudAgentSessionId: 'agent_report_outbox' }, + run: { + messageId, + status: 'queued', + queuedAt: '2026-05-26T08:00:00.000Z', + }, + }; +} + +function failedReport(messageId: string): CloudAgentQueueReport { + return { + version: 1, + type: 'run.state', + occurredAt: '2026-05-26T08:04:00.000Z', + session: { cloudAgentSessionId: 'agent_report_outbox' }, + run: { + messageId, + status: 'failed', + terminalAt: '2026-05-26T08:04:00.000Z', + failureStage: 'unknown', + failureCode: 'unclassified', + }, + }; +} + +function createHarness(options: { + queue?: { send: (report: CloudAgentQueueReport) => Promise }; +} = {}) { + const kv = memoryKv(); + const outbox = createReportOutbox({ + storage: { kv } as never, + getQueue: () => options.queue as never, + }); + return { kv, outbox }; +} + +describe('createReportOutbox', () => { + it('keeps one latest cumulative snapshot per message under a fresh obligation identity', () => { + const { kv, outbox } = createHarness(); + + outbox.record(queuedReport('msg_one')); + const first = parsePendingRunReport(kv.get(reportOutboxKey('msg_one'))); + outbox.record(queuedReport('msg_one')); + const second = parsePendingRunReport(kv.get(reportOutboxKey('msg_one'))); + + expect(second?.obligationId).toEqual(expect.any(String)); + expect(second?.obligationId).not.toBe(first?.obligationId); + expect(outbox.pendingCount()).toBe(1); + }); + + it('evicts the entry with the greatest dueAt when the cap is reached', () => { + const { kv, outbox } = createHarness(); + for (let index = 0; index < REPORT_OUTBOX_MAX_ENTRIES; index++) { + outbox.record(queuedReport(`msg_${index}`)); + } + const oldestKey = reportOutboxKey('msg_0'); + const stored = kv.get>(oldestKey); + expect(stored).toBeDefined(); + kv.put(oldestKey, { ...stored, dueAt: Date.now() + 10_000_000 }); + + outbox.record(queuedReport('msg_new')); + + expect(outbox.pendingCount()).toBe(REPORT_OUTBOX_MAX_ENTRIES); + expect(kv.get(oldestKey)).toBeUndefined(); + expect(kv.get(reportOutboxKey('msg_new'))).toBeDefined(); + }); + + it('keeps a newer obligation that replaced the one being sent', async () => { + const { kv, outbox } = createHarness({ + queue: { + send: async () => { + outbox.record(queuedReport('msg_one')); + }, + }, + }); + outbox.record(queuedReport('msg_one')); + const sentObligationId = parsePendingRunReport(kv.get(reportOutboxKey('msg_one')))?.obligationId; + + await outbox.repair(); + + const stored = parsePendingRunReport(kv.get(reportOutboxKey('msg_one'))); + expect(stored?.obligationId).toEqual(expect.any(String)); + expect(stored?.obligationId).not.toBe(sentObligationId); + expect(stored?.attempts).toBe(0); + }); + + it('deletes the obligation after a successful send', async () => { + const sent: CloudAgentQueueReport[] = []; + const { kv, outbox } = createHarness({ + queue: { send: async report => void sent.push(report) }, + }); + outbox.record(queuedReport('msg_one')); + + await outbox.repair(); + + expect(sent).toHaveLength(1); + expect(kv.get(reportOutboxKey('msg_one'))).toBeUndefined(); + }); + + it('retries a failed send and abandons after the attempt budget', async () => { + const send = vi.fn(async () => { + throw new Error('queue unavailable'); + }); + const { kv, outbox } = createHarness({ queue: { send } }); + outbox.record(queuedReport('msg_one')); + + let now = Date.now(); + for (let attempt = 1; attempt <= REPORT_ENQUEUE_MAX_ATTEMPTS; attempt++) { + await outbox.repair(now); + if (attempt < REPORT_ENQUEUE_MAX_ATTEMPTS) { + expect(parsePendingRunReport(kv.get(reportOutboxKey('msg_one')))?.attempts).toBe(attempt); + now += REPORT_ENQUEUE_RETRY_MS; + } + } + + expect(send).toHaveBeenCalledTimes(REPORT_ENQUEUE_MAX_ATTEMPTS); + expect(kv.get(reportOutboxKey('msg_one'))).toBeUndefined(); + }); + + it('keeps a reserved obligation while no queue binding is available', async () => { + const { kv, outbox } = createHarness(); + outbox.record(queuedReport('msg_one')); + + const now = Date.now(); + await outbox.repair(now); + + expect(parsePendingRunReport(kv.get(reportOutboxKey('msg_one')))?.attempts).toBe(1); + expect(outbox.nextDueAt()).toBe(now + REPORT_ENQUEUE_RETRY_MS); + }); + + it('deletes invalid persisted entries instead of sending them', async () => { + const sent: CloudAgentQueueReport[] = []; + const { kv, outbox } = createHarness({ + queue: { send: async report => void sent.push(report) }, + }); + kv.put(`${REPORT_OUTBOX_PREFIX}msg_bad`, { attempts: 0, dueAt: Date.now() }); + kv.put(`${REPORT_OUTBOX_PREFIX}msg_older_shape`, { job: {}, attempts: 0, dueAt: Date.now() }); + + expect(outbox.pendingCount()).toBe(2); + await outbox.repair(); + + expect(sent).toHaveLength(0); + expect(outbox.pendingCount()).toBe(0); + }); + + it('keeps a newer snapshot that lands while an older entry send is in flight', async () => { + const kv = memoryKv(); + const firstSend = Promise.withResolvers(); + const secondSend = Promise.withResolvers(); + const sent: string[] = []; + const outbox = createReportOutbox({ + storage: { kv } as never, + getQueue: () => + ({ + send: async (report: CloudAgentQueueReport) => { + sent.push(report.run.messageId); + return sent.length === 1 ? firstSend.promise : secondSend.promise; + }, + }) as never, + }); + outbox.record(queuedReport('msg_a')); + outbox.record(queuedReport('msg_b')); + + const repair = outbox.repair(Date.now() + 1_000_000); + // Replaces B with a newer terminal obligation while A's send is still pending. + outbox.record(failedReport('msg_b')); + const newer = parsePendingRunReport(kv.get(reportOutboxKey('msg_b'))); + firstSend.resolve(); + await vi.waitFor(() => expect(sent).toHaveLength(2)); + + const stored = parsePendingRunReport(kv.get(reportOutboxKey('msg_b'))); + expect(stored?.obligationId).toBe(newer?.obligationId); + expect(stored?.report.run.status).toBe('failed'); + + secondSend.resolve(); + await repair; + }); + + it('does not delete a newer obligation when an exhausted send fails', async () => { + const kv = memoryKv(); + const reject = Promise.withResolvers(); + const outbox = createReportOutbox({ + storage: { kv } as never, + getQueue: () => ({ send: async () => reject.promise }) as never, + }); + const key = reportOutboxKey('msg_b'); + outbox.record(queuedReport('msg_b')); + kv.put(key, { + ...(parsePendingRunReport(kv.get(key)) as object), + attempts: REPORT_ENQUEUE_MAX_ATTEMPTS - 1, + dueAt: Date.now(), + }); + + const repair = outbox.repair(Date.now() + 1_000_000); + // A newer obligation lands while the final attempt's send is pending. + outbox.record(failedReport('msg_b')); + const newer = parsePendingRunReport(kv.get(key)); + reject.reject(new Error('queue send failed')); + await repair; + + const stored = parsePendingRunReport(kv.get(key)); + expect(stored?.obligationId).toBe(newer?.obligationId); + expect(stored?.report.run.status).toBe('failed'); + }); + + it('does not let an evicted in-flight send delete a recreated obligation', async () => { + const kv = memoryKv(); + let sequence = 0; + const firstSend = Promise.withResolvers(); + const outbox = createReportOutbox({ + storage: { kv } as never, + createObligationId: () => `ob_${++sequence}`, + getQueue: () => ({ send: async () => firstSend.promise }) as never, + }); + const key = reportOutboxKey('msg_a'); + outbox.record(queuedReport('msg_a')); + + // Repair reserves A (greatest dueAt) and holds its send open. + const repair = outbox.repair(Date.now() + 1_000_000); + for (let index = 0; index < REPORT_OUTBOX_MAX_ENTRIES - 1; index++) { + outbox.record(queuedReport(`msg_filler_${index}`)); + } + // The 51st entry forces enforceEntryCap to evict A while its send is in flight. + outbox.record(queuedReport('msg_new')); + expect(kv.get(key)).toBeUndefined(); + + // A newer terminal snapshot re-creates A with a fresh obligation identity. + outbox.record(failedReport('msg_a')); + const recreated = parsePendingRunReport(kv.get(key)); + expect(recreated?.report.run.status).toBe('failed'); + + firstSend.resolve(); + await repair; + + const stored = parsePendingRunReport(kv.get(key)); + expect(stored?.obligationId).toBe(recreated?.obligationId); + expect(stored?.report.run.status).toBe('failed'); + expect(stored?.attempts).toBe(0); + }); +}); diff --git a/services/cloud-agent-next/src/sandbox-session/report-outbox.ts b/services/cloud-agent-next/src/sandbox-session/report-outbox.ts new file mode 100644 index 0000000000..5ecab30dea --- /dev/null +++ b/services/cloud-agent-next/src/sandbox-session/report-outbox.ts @@ -0,0 +1,267 @@ +import { + CloudAgentQueueReportSchema, + type CloudAgentQueueReport, +} from '@kilocode/worker-utils/cloud-agent-queue-report'; +import { logger } from '../logger.js'; + +export const REPORT_OUTBOX_PREFIX = 'report_outbox:'; +export const REPORT_ANCHOR_KEY = 'report_anchor'; +export const REPORT_ENQUEUE_MAX_ATTEMPTS = 5; +export const REPORT_ENQUEUE_RETRY_MS = 30_000; +export const REPORT_OUTBOX_MAX_ENTRIES = 50; + +type ReportQueue = Pick, 'send'>; +type ReportStorage = Pick; + +export type PendingRunReport = { + report: CloudAgentQueueReport; + obligationId: string; + attempts: number; + dueAt: number; +}; + +export type ReportAnchor = { + version: 1; + kiloSessionId: string; + initialMessageId: string; + createdAt: number; +}; + +export type ReportOutboxDependencies = { + storage: ReportStorage; + getQueue: () => ReportQueue | undefined; + createObligationId?: () => string; +}; + +export type ReportOutbox = { + record(report: CloudAgentQueueReport): void; + pendingCount(): number; + nextDueAt(): number | undefined; + repair(now?: number): Promise; +}; + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value); +} + +export function parsePendingRunReport(value: unknown): PendingRunReport | undefined { + if (!isRecord(value)) return undefined; + const parsed = CloudAgentQueueReportSchema.safeParse(value.report); + const obligationId = value.obligationId; + const attempts = value.attempts; + const dueAt = value.dueAt; + if ( + !parsed.success || + typeof obligationId !== 'string' || + obligationId.length === 0 || + typeof attempts !== 'number' || + !Number.isInteger(attempts) || + typeof dueAt !== 'number' || + !Number.isFinite(dueAt) || + attempts < 0 || + attempts > REPORT_ENQUEUE_MAX_ATTEMPTS + ) { + return undefined; + } + return { report: parsed.data, obligationId, attempts, dueAt }; +} + +export function reportOutboxKey(messageId: string): string { + return `${REPORT_OUTBOX_PREFIX}${messageId}`; +} + +export function readReportAnchor(storage: ReportStorage): ReportAnchor | undefined { + const raw = storage.kv.get(REPORT_ANCHOR_KEY); + if (!isRecord(raw)) return undefined; + const { version, kiloSessionId, initialMessageId, createdAt } = raw; + if ( + version !== 1 || + typeof kiloSessionId !== 'string' || + typeof initialMessageId !== 'string' || + typeof createdAt !== 'number' || + !Number.isFinite(createdAt) + ) { + return undefined; + } + return { version: 1, kiloSessionId, initialMessageId, createdAt }; +} + +export function writeReportAnchor( + storage: ReportStorage, + anchor: Omit +): void { + storage.kv.put(REPORT_ANCHOR_KEY, { version: 1, ...anchor }); +} + +export function createReportOutbox(dependencies: ReportOutboxDependencies): ReportOutbox { + const { storage, getQueue } = dependencies; + const createObligationId = dependencies.createObligationId ?? (() => crypto.randomUUID()); + let repairInFlight: Promise | undefined; + + function pendingEntries(): Array<[string, PendingRunReport | undefined]> { + return Array.from(storage.kv.list({ prefix: REPORT_OUTBOX_PREFIX })).map( + ([key, value]) => [key, parsePendingRunReport(value)] + ); + } + + function enforceEntryCap(targetKey: string): void { + const entries = Array.from(storage.kv.list({ prefix: REPORT_OUTBOX_PREFIX })); + if (entries.length < REPORT_OUTBOX_MAX_ENTRIES) return; + if (entries.some(([key]) => key === targetKey)) return; + const evicted = entries + .filter(([key]) => key !== targetKey) + .map(([key, value]) => ({ + key, + dueAt: parsePendingRunReport(value)?.dueAt ?? Number.MAX_SAFE_INTEGER, + })) + .reduce((greatest, entry) => (entry.dueAt > greatest.dueAt ? entry : greatest)); + logger.withFields({ key: evicted.key }).warn('Cloud Agent report outbox entry evicted'); + storage.kv.delete(evicted.key); + } + + function record(report: CloudAgentQueueReport): void { + const key = reportOutboxKey(report.run.messageId); + enforceEntryCap(key); + storage.kv.put(key, { + report: structuredClone(report), + obligationId: createObligationId(), + attempts: 0, + dueAt: Date.now(), + }); + } + + function pendingCount(): number { + return pendingEntries().length; + } + + function nextDueAt(): number | undefined { + let next: number | undefined; + for (const [, pending] of pendingEntries()) { + if (!pending || pending.attempts >= REPORT_ENQUEUE_MAX_ATTEMPTS) { + next = Math.min(next ?? Date.now(), Date.now()); + continue; + } + next = Math.min(next ?? pending.dueAt, pending.dueAt); + } + return next; + } + + function logFailure( + pending: PendingRunReport | undefined, + attempts: number, + abandoned: boolean + ): void { + logger + .withFields({ + messageId: pending?.report.run.messageId, + status: pending?.report.run.status, + attempts, + }) + .error( + abandoned + ? 'Cloud Agent report enqueue abandoned' + : 'Cloud Agent report enqueue failed; retry scheduled' + ); + } + + /** + * Reserves the next attempt only when the stored obligation identity still + * matches the one being processed, so a newer snapshot recorded meanwhile — + * even one that re-created the key after eviction — is never overwritten. + */ + function reserveIfUnchanged( + key: string, + parsed: PendingRunReport, + attempts: number, + now: number + ): boolean { + const current = parsePendingRunReport(storage.kv.get(key)); + if (current?.obligationId !== parsed.obligationId) return false; + storage.kv.put(key, { + ...parsed, + attempts, + dueAt: now + REPORT_ENQUEUE_RETRY_MS, + }); + return true; + } + + /** + * Deletes only when the stored obligation identity still matches what was + * processed. Passing `undefined` deletes only an entry that is still + * invalid/absent. + */ + function deleteIfUnchanged(key: string, expected: PendingRunReport | undefined): boolean { + const current = parsePendingRunReport(storage.kv.get(key)); + if (current === undefined) { + if (expected !== undefined) return false; + storage.kv.delete(key); + return true; + } + if (expected !== undefined && current.obligationId === expected.obligationId) { + storage.kv.delete(key); + return true; + } + return false; + } + + async function runRepair(now: number): Promise { + const keys = Array.from( + storage.kv.list({ prefix: REPORT_OUTBOX_PREFIX }), + ([key]) => key + ); + for (const key of keys) { + // Re-read each iteration: a previous entry's `await send` can span a + // `record` that replaces this key with a newer obligation. + const parsed = parsePendingRunReport(storage.kv.get(key)); + if (parsed === undefined) { + if (deleteIfUnchanged(key, undefined)) { + logger + .withFields({ keyPrefix: REPORT_OUTBOX_PREFIX }) + .error('Invalid report outbox entry'); + } + continue; + } + if (parsed.attempts >= REPORT_ENQUEUE_MAX_ATTEMPTS) { + logFailure(parsed, parsed.attempts, true); + deleteIfUnchanged(key, parsed); + continue; + } + if (parsed.dueAt > now) continue; + + const attempts = parsed.attempts + 1; + const abandoned = attempts >= REPORT_ENQUEUE_MAX_ATTEMPTS; + if (!reserveIfUnchanged(key, parsed, attempts, now)) continue; + + const queue = getQueue(); + if (!queue) { + logFailure(parsed, attempts, abandoned); + if (abandoned) deleteIfUnchanged(key, parsed); + continue; + } + + try { + await queue.send(parsed.report); + deleteIfUnchanged(key, parsed); + } catch { + logFailure(parsed, attempts, abandoned); + if (abandoned) deleteIfUnchanged(key, parsed); + } + } + } + + function repair(now = Date.now()): Promise { + if (repairInFlight) return repairInFlight; + const pending = runRepair(now).finally(() => { + repairInFlight = undefined; + }); + repairInFlight = pending; + return pending; + } + + return { + record, + pendingCount, + nextDueAt, + repair, + }; +} diff --git a/services/cloud-agent-next/src/sandbox-session/session-message-queue.test.ts b/services/cloud-agent-next/src/sandbox-session/session-message-queue.test.ts index 650510c356..2498132643 100644 --- a/services/cloud-agent-next/src/sandbox-session/session-message-queue.test.ts +++ b/services/cloud-agent-next/src/sandbox-session/session-message-queue.test.ts @@ -1209,6 +1209,7 @@ function sessionFixture( const env = { SANDBOX_CONTROL: { getByName: () => sharedControl ?? control }, CALLBACK_QUEUE: callbackQueue, + CLOUD_AGENT_REPORT_QUEUE: { send: async () => undefined }, CLOUD_AGENT_CONTAINER_BILLING_ENABLED: 'true', CLOUD_AGENT_CONTAINER_BILLING_ORG_IDS: 'org_1', CLOUD_AGENT_CONTAINER_BILLING_USER_IDS: 'user_1', diff --git a/services/cloud-agent-next/src/sandbox-session/session-message-queue.ts b/services/cloud-agent-next/src/sandbox-session/session-message-queue.ts index 6b4dcbefac..7bb8827d34 100644 --- a/services/cloud-agent-next/src/sandbox-session/session-message-queue.ts +++ b/services/cloud-agent-next/src/sandbox-session/session-message-queue.ts @@ -52,6 +52,7 @@ export type SessionOperationProof = { type SessionMessageLifecycle = { messageId: string; state: SessionMessageState; + queuedAt?: number; acceptedAt?: number; lastActivityAt?: number; deliveryDeadlineAt?: number; diff --git a/services/cloud-agent-next/src/sandbox-session/terminal-lifecycle.ts b/services/cloud-agent-next/src/sandbox-session/terminal-lifecycle.ts index 8f9e8da381..54fbda7e1c 100644 --- a/services/cloud-agent-next/src/sandbox-session/terminal-lifecycle.ts +++ b/services/cloud-agent-next/src/sandbox-session/terminal-lifecycle.ts @@ -22,6 +22,7 @@ import { isTerminalSessionPlatform } from '../terminal/access.js'; import type { sandboxControlRpc } from './control-rpc.js'; import type { SandboxTerminalRecord } from './terminal-bridge.js'; import { CALLBACK_OUTBOX_PREFIX } from './message-callbacks.js'; +import { REPORT_ANCHOR_KEY, REPORT_OUTBOX_PREFIX } from './report-outbox.js'; export const SANDBOX_SESSION_METADATA_KEY = 'session_metadata'; export const SANDBOX_SESSION_LIFECYCLE_KEY = 'session_lifecycle_fence'; @@ -820,8 +821,14 @@ export function createSandboxTerminalLifecycle(deps: TerminalLifecycleDeps) { if (readFence()?.state !== 'deleted') return; const keys = Array.from(storage.kv.list(), ([key]) => key); for (const key of keys) { - if (key !== SANDBOX_SESSION_LIFECYCLE_KEY && !key.startsWith(CALLBACK_OUTBOX_PREFIX)) - storage.kv.delete(key); + if ( + key === SANDBOX_SESSION_LIFECYCLE_KEY || + key === REPORT_ANCHOR_KEY || + key.startsWith(CALLBACK_OUTBOX_PREFIX) || + key.startsWith(REPORT_OUTBOX_PREFIX) + ) + continue; + storage.kv.delete(key); } } diff --git a/services/cloud-agent-next/src/telemetry/control-plane-failure.test.ts b/services/cloud-agent-next/src/telemetry/control-plane-failure.test.ts new file mode 100644 index 0000000000..1678114af7 --- /dev/null +++ b/services/cloud-agent-next/src/telemetry/control-plane-failure.test.ts @@ -0,0 +1,77 @@ +import { describe, expect, it } from 'vitest'; +import { CloudAgentRunFailureClassifications } from '@kilocode/worker-utils/cloud-agent-queue-report'; +import { + classifyControlPlaneFailure, + type ControlPlaneDispatchState, +} from './control-plane-failure.js'; + +const validPairs = new Set( + CloudAgentRunFailureClassifications.map( + classification => `${classification.failureStage}:${classification.failureCode}` + ) +); + +type Status = 'failed' | 'interrupted'; + +const cases: ReadonlyArray< + readonly [string | undefined, ControlPlaneDispatchState, Status, string, string] +> = [ + // failed: coordinator reasons keep their bounded mapping. + ['missing_metadata', 'pre_dispatch', 'failed', 'pre_dispatch', 'session_metadata_missing'], + ['missing_metadata', 'accepted', 'failed', 'pre_dispatch', 'session_metadata_missing'], + ['preparation_timeout', 'pre_dispatch', 'failed', 'pre_dispatch', 'wrapper_start_failed'], + ['attach_exhausted', 'pre_dispatch', 'failed', 'pre_dispatch', 'wrapper_start_failed'], + ['prompt_exhausted', 'accepted', 'failed', 'post_dispatch_no_activity', 'wrapper_disconnected'], + ['prompt_exhausted', 'pre_dispatch', 'failed', 'pre_dispatch', 'invalid_delivery_request'], + ['environment_failed', 'accepted', 'failed', 'post_dispatch_no_activity', 'wrapper_disconnected'], + ['environment_failed', 'pre_dispatch', 'failed', 'pre_dispatch', 'sandbox_connect_failed'], + ['provider_unknown', 'accepted', 'failed', 'pre_dispatch', 'sandbox_connect_failed'], + ['runtime_unhealthy', 'accepted', 'failed', 'post_dispatch_no_activity', 'wrapper_disconnected'], + ['runtime_unhealthy', 'pre_dispatch', 'failed', 'pre_dispatch', 'wrapper_start_failed'], + ['accepted_overdue', 'pre_dispatch', 'failed', 'post_dispatch_no_activity', 'wrapper_no_output'], + ['invalid_model', 'pre_dispatch', 'failed', 'pre_dispatch', 'model_missing'], + // failed: a cancellation reason or arbitrary text is never an interruption stage. + ['queued_message_cancelled', 'pre_dispatch', 'failed', 'unknown', 'unclassified'], + ['interruption_unconfirmed', 'accepted', 'failed', 'unknown', 'unclassified'], + [undefined, 'pre_dispatch', 'failed', 'unknown', 'unclassified'], + ['some_wrapper_reason', 'accepted', 'failed', 'unknown', 'unclassified'], + // interrupted: the lifecycle status decides, regardless of the reason text. + ['queued_message_cancelled', 'pre_dispatch', 'interrupted', 'interruption', 'user_interrupt'], + ['interruption_unconfirmed', 'accepted', 'interrupted', 'interruption', 'user_interrupt'], + [undefined, 'pre_dispatch', 'interrupted', 'interruption', 'system_interrupt'], + ['missing_metadata', 'pre_dispatch', 'interrupted', 'interruption', 'system_interrupt'], + ['preparation_timeout', 'accepted', 'interrupted', 'interruption', 'system_interrupt'], + ['some_wrapper_reason', 'accepted', 'interrupted', 'interruption', 'system_interrupt'], +]; + +describe('classifyControlPlaneFailure', () => { + it.each(cases)( + 'maps %s at %s for %s to %s/%s', + (reason, dispatchState, status, failureStage, failureCode) => { + expect(classifyControlPlaneFailure(reason, dispatchState, status)).toEqual({ + stage: failureStage, + code: failureCode, + }); + } + ); + + it('only returns valid reporting classifications', () => { + for (const [reason, dispatchState, status] of cases) { + const classification = classifyControlPlaneFailure(reason, dispatchState, status); + expect(validPairs.has(`${classification.stage}:${classification.code}`)).toBe(true); + } + }); + + it('never reports an interruption stage for a failed run', () => { + for (const [reason, dispatchState] of [ + ['queued_message_cancelled', 'pre_dispatch'], + ['interruption_unconfirmed', 'accepted'], + [undefined, 'pre_dispatch'], + ['missing_metadata', 'pre_dispatch'], + ] as const) { + expect(classifyControlPlaneFailure(reason, dispatchState, 'failed').stage).not.toBe( + 'interruption' + ); + } + }); +}); diff --git a/services/cloud-agent-next/src/telemetry/control-plane-failure.ts b/services/cloud-agent-next/src/telemetry/control-plane-failure.ts new file mode 100644 index 0000000000..32d565b179 --- /dev/null +++ b/services/cloud-agent-next/src/telemetry/control-plane-failure.ts @@ -0,0 +1,87 @@ +import type { + CloudAgentFailureCode, + CloudAgentFailureStage, +} from '@kilocode/worker-utils/cloud-agent-failure'; + +/** + * The accepted-vs-pre-dispatch fact is captured at the committing transition. + * A wrapper outcome `reason` is arbitrary text, so anything unrecognized falls + * through to `unknown`/`unclassified` rather than being forced into + * `pre_dispatch`. + */ +export type ControlPlaneDispatchState = 'pre_dispatch' | 'accepted'; + +export type ControlPlaneFailureClassification = { + stage: CloudAgentFailureStage; + code: CloudAgentFailureCode; +}; + +const PRE_DISPATCH: ControlPlaneFailureClassification = { + stage: 'pre_dispatch', + code: 'wrapper_start_failed', +}; +const POST_DISPATCH_WRAPPER_DISCONNECTED: ControlPlaneFailureClassification = { + stage: 'post_dispatch_no_activity', + code: 'wrapper_disconnected', +}; +const PRE_DISPATCH_SANDBOX_CONNECT: ControlPlaneFailureClassification = { + stage: 'pre_dispatch', + code: 'sandbox_connect_failed', +}; +const INTERRUPTION_USER: ControlPlaneFailureClassification = { + stage: 'interruption', + code: 'user_interrupt', +}; +const INTERRUPTION_SYSTEM: ControlPlaneFailureClassification = { + stage: 'interruption', + code: 'system_interrupt', +}; +const UNKNOWN: ControlPlaneFailureClassification = { + stage: 'unknown', + code: 'unclassified', +}; + +export function classifyControlPlaneFailure( + reason: string | undefined, + dispatchState: ControlPlaneDispatchState, + status: 'failed' | 'interrupted' +): ControlPlaneFailureClassification { + if (status === 'interrupted') { + // An interrupted lifecycle is a cancellation, never a platform failure, + // even when a wrapper supplied arbitrary text as the reason. + return reason === 'queued_message_cancelled' || reason === 'interruption_unconfirmed' + ? INTERRUPTION_USER + : INTERRUPTION_SYSTEM; + } + switch (reason) { + case 'missing_metadata': + return { stage: 'pre_dispatch', code: 'session_metadata_missing' }; + case 'preparation_timeout': + case 'attach_exhausted': + return PRE_DISPATCH; + case 'prompt_exhausted': + return dispatchState === 'accepted' + ? POST_DISPATCH_WRAPPER_DISCONNECTED + : { stage: 'pre_dispatch', code: 'invalid_delivery_request' }; + case 'environment_failed': + return dispatchState === 'accepted' + ? POST_DISPATCH_WRAPPER_DISCONNECTED + : PRE_DISPATCH_SANDBOX_CONNECT; + case 'provider_unknown': + return PRE_DISPATCH_SANDBOX_CONNECT; + case 'runtime_unhealthy': + return dispatchState === 'accepted' + ? POST_DISPATCH_WRAPPER_DISCONNECTED + : PRE_DISPATCH; + case 'accepted_overdue': + return { stage: 'post_dispatch_no_activity', code: 'wrapper_no_output' }; + case 'invalid_model': + return { stage: 'pre_dispatch', code: 'model_missing' }; + case 'queued_message_cancelled': + case 'interruption_unconfirmed': + case undefined: + return UNKNOWN; + default: + return UNKNOWN; + } +} diff --git a/services/cloud-agent-next/src/telemetry/queue-reports.ts b/services/cloud-agent-next/src/telemetry/queue-reports.ts index 53b9402c4b..6764be5d4e 100644 --- a/services/cloud-agent-next/src/telemetry/queue-reports.ts +++ b/services/cloud-agent-next/src/telemetry/queue-reports.ts @@ -31,7 +31,7 @@ const INSUFFICIENT_CREDIT_TERMINAL_ERRORS = new Set([ 'insufficient credits: insufficient_funds', 'payment required', ]); -const FAILED_RUN_DIAGNOSTIC_MESSAGES: Partial< +export const FAILED_RUN_DIAGNOSTIC_MESSAGES: Partial< Record, string> > = { sandbox_connect_failed: 'Sandbox connection failed', @@ -128,6 +128,38 @@ async function trySendReport( } } +export type RunReportAnchor = { + kiloSessionId?: string; + initialMessageId?: string; + reportingCreatedAt?: string; +}; + +/** Assembles a validated queue report from already-decided run facts. */ +export function buildRunStateReport(params: { + cloudAgentSessionId: string; + anchor?: RunReportAnchor; + run: CloudAgentRunStateReport['run']; + occurredAt: number; +}): CloudAgentRunStateReport { + const { anchor } = params; + return { + version: 1, + type: 'run.state', + occurredAt: new Date(params.occurredAt).toISOString(), + session: { + cloudAgentSessionId: params.cloudAgentSessionId, + ...(anchor?.kiloSessionId === undefined ? {} : { kiloSessionId: anchor.kiloSessionId }), + ...(anchor?.initialMessageId === undefined + ? {} + : { initialMessageId: anchor.initialMessageId }), + ...(anchor?.reportingCreatedAt === undefined + ? {} + : { reportingCreatedAt: anchor.reportingCreatedAt }), + }, + run: params.run, + }; +} + export async function emitRunStateReport(params: { queue?: ReportQueue; cloudAgentSessionId: string; @@ -159,38 +191,37 @@ export async function emitRunStateReport(params: { : {}), }) : undefined; - const report: CloudAgentRunStateReport = { - version: 1, - type: 'run.state', - occurredAt: new Date(params.occurredAt ?? Date.now()).toISOString(), - session: { cloudAgentSessionId: params.cloudAgentSessionId }, - run: { - messageId: state.messageId, - status: state.status, - ...(state.wrapperRunId === undefined ? {} : { wrapperRunId: state.wrapperRunId }), - ...(state.queuedAt === undefined ? {} : { queuedAt: timestamp(state.queuedAt) }), - ...(observedDispatchAcceptedAt === undefined - ? {} - : { dispatchAcceptedAt: timestamp(observedDispatchAcceptedAt) }), - ...(state.agentActivityObservedAt === undefined - ? {} - : { agentActivityObservedAt: timestamp(state.agentActivityObservedAt) }), - ...(state.terminalAt === undefined ? {} : { terminalAt: timestamp(state.terminalAt) }), - ...(state.failureStage === undefined ? {} : { failureStage: state.failureStage }), - ...(failureCode === undefined ? {} : { failureCode }), - ...(state.failureCode === 'workspace_setup_failed' && - isWorkspaceFailureSubtype(state.failureSubtype) - ? { workspaceFailureSubtype: state.failureSubtype } - : {}), - ...(failureClassification === undefined - ? {} - : { - failureResponsibility: failureClassification.responsibility, - failureReason: failureClassification.reason, - }), - ...(diagnostic === undefined ? {} : { diagnostic }), - }, + const run: CloudAgentRunStateReport['run'] = { + messageId: state.messageId, + status: state.status, + ...(state.wrapperRunId === undefined ? {} : { wrapperRunId: state.wrapperRunId }), + ...(state.queuedAt === undefined ? {} : { queuedAt: timestamp(state.queuedAt) }), + ...(observedDispatchAcceptedAt === undefined + ? {} + : { dispatchAcceptedAt: timestamp(observedDispatchAcceptedAt) }), + ...(state.agentActivityObservedAt === undefined + ? {} + : { agentActivityObservedAt: timestamp(state.agentActivityObservedAt) }), + ...(state.terminalAt === undefined ? {} : { terminalAt: timestamp(state.terminalAt) }), + ...(state.failureStage === undefined ? {} : { failureStage: state.failureStage }), + ...(failureCode === undefined ? {} : { failureCode }), + ...(state.failureCode === 'workspace_setup_failed' && + isWorkspaceFailureSubtype(state.failureSubtype) + ? { workspaceFailureSubtype: state.failureSubtype } + : {}), + ...(failureClassification === undefined + ? {} + : { + failureResponsibility: failureClassification.responsibility, + failureReason: failureClassification.reason, + }), + ...(diagnostic === undefined ? {} : { diagnostic }), }; + const report = buildRunStateReport({ + cloudAgentSessionId: params.cloudAgentSessionId, + run, + occurredAt: params.occurredAt ?? Date.now(), + }); if (failureClassification !== undefined) { console.info('Cloud Agent failure classified', { metric: 'cloud_agent_failure_classified', diff --git a/services/cloud-agent-next/src/telemetry/report-consumer.test.ts b/services/cloud-agent-next/src/telemetry/report-consumer.test.ts index 81ed7bd032..e212717228 100644 --- a/services/cloud-agent-next/src/telemetry/report-consumer.test.ts +++ b/services/cloud-agent-next/src/telemetry/report-consumer.test.ts @@ -111,6 +111,28 @@ describe('Cloud Agent report consumer', () => { } ); + it('acks a report whose parent identity conflicts without retrying', async () => { + const error = vi.spyOn(console, 'error').mockImplementation(() => undefined); + const saveReport = vi.fn().mockResolvedValueOnce({ outcome: 'conflict' }); + vi.mocked(createCloudAgentReportStore).mockReturnValue({ saveReport } as never); + const message = makeMessage(report); + + await consumeCloudAgentReportBatch( + { messages: [message] } as unknown as MessageBatch, + env + ); + + expect(error).toHaveBeenCalledWith( + 'Dropping Cloud Agent run report with conflicting session parent identity', + { + cloudAgentSessionId: report.session.cloudAgentSessionId, + messageId: report.run.messageId, + } + ); + expect(message.ack).toHaveBeenCalledOnce(); + expect(message.retry).not.toHaveBeenCalled(); + }); + it('continues the batch after a missing parent and acknowledges a later successful redelivery', async () => { vi.spyOn(console, 'warn').mockImplementation(() => undefined); const saveReport = vi diff --git a/services/cloud-agent-next/src/telemetry/report-consumer.ts b/services/cloud-agent-next/src/telemetry/report-consumer.ts index 060b5703e0..dbb11d2cf4 100644 --- a/services/cloud-agent-next/src/telemetry/report-consumer.ts +++ b/services/cloud-agent-next/src/telemetry/report-consumer.ts @@ -107,6 +107,14 @@ export async function consumeCloudAgentReportBatch( message.retry(); continue; } + if (result.outcome === 'conflict') { + console.error('Dropping Cloud Agent run report with conflicting session parent identity', { + cloudAgentSessionId: parsed.data.session.cloudAgentSessionId, + messageId: parsed.data.run.messageId, + }); + message.ack(); + continue; + } message.ack(); } catch { console.error('Saving Cloud Agent report failed; message will retry', { 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 c1d4d997a1..8f6502019a 100644 --- a/services/cloud-agent-next/src/telemetry/report-store.test.ts +++ b/services/cloud-agent-next/src/telemetry/report-store.test.ts @@ -292,6 +292,113 @@ describe('cloud agent reporting store', () => { expect(fake.inserts).toHaveLength(0); }); + const anchor = { + kiloSessionId: 'ses_12345678901234567890123456', + initialMessageId: 'msg_anchor_initial', + reportingCreatedAt: occurredAt, + }; + const anchoredReport = { + version: 1, + type: 'run.state', + occurredAt, + session: { cloudAgentSessionId, ...anchor }, + run: { messageId: 'msg_anchor_run', status: 'queued', queuedAt: occurredAt }, + } satisfies CloudAgentRunStateReport; + + it('creates a retention-eligible parent from a trusted anchor then applies the run', async () => { + const fake = makeDb([ + [], + [ + { + createdAt: occurredAt, + kiloSessionId: anchor.kiloSessionId, + initialMessageId: anchor.initialMessageId, + }, + ], + [], + ]); + const store = createCloudAgentReportStore(fake.db as never); + + expect(await store.saveReport(anchoredReport, occurredAt)).toEqual({ outcome: 'applied' }); + expect(fake.inserts.find(call => call.table === cloud_agent_sessions)?.values).toEqual({ + cloud_agent_session_id: cloudAgentSessionId, + kilo_session_id: anchor.kiloSessionId, + initial_message_id: anchor.initialMessageId, + created_at: anchor.reportingCreatedAt, + }); + expect( + fake.inserts.find(call => call.table === cloud_agent_session_runs)?.values + ).toMatchObject({ + cloud_agent_session_id: cloudAgentSessionId, + message_id: 'msg_anchor_run', + status: 'queued', + }); + }); + + it('refuses to resurrect a parent for an anchor older than the retention cutoff', async () => { + const fake = makeDb([[]]); + const store = createCloudAgentReportStore(fake.db as never); + const expiredAnchor = { + ...anchoredReport, + session: { + cloudAgentSessionId, + ...anchor, + reportingCreatedAt: new Date( + Date.parse(occurredAt) - 91 * 24 * 60 * 60 * 1000 + ).toISOString(), + }, + } satisfies CloudAgentRunStateReport; + + expect(await store.saveReport(expiredAnchor, occurredAt)).toEqual({ + outcome: 'missing_parent', + }); + expect(fake.inserts).toHaveLength(0); + }); + + it('reports missing_parent when a concurrent parent still cannot be read back', async () => { + const fake = makeDb([[], [], []]); + const store = createCloudAgentReportStore(fake.db as never); + + expect(await store.saveReport(anchoredReport, occurredAt)).toEqual({ + outcome: 'missing_parent', + }); + expect(fake.inserts.some(call => call.table === cloud_agent_sessions)).toBe(true); + expect(fake.inserts.some(call => call.table === cloud_agent_session_runs)).toBe(false); + }); + + it('leaves an existing parent untouched when applying an anchored report', async () => { + const fake = makeDb([ + [ + { + createdAt: occurredAt, + kiloSessionId: anchor.kiloSessionId, + initialMessageId: anchor.initialMessageId, + }, + ], + [], + ]); + const store = createCloudAgentReportStore(fake.db as never); + + expect(await store.saveReport(anchoredReport, occurredAt)).toEqual({ outcome: 'applied' }); + expect(fake.inserts.some(call => call.table === cloud_agent_sessions)).toBe(false); + }); + + it('refuses an anchored report whose parent identity conflicts', async () => { + const fake = makeDb([ + [ + { + createdAt: occurredAt, + kiloSessionId: 'ses_zzzzzzzzzzzzzzzzzzzzzzzzzz', + initialMessageId: 'msg_other_initial', + }, + ], + ]); + const store = createCloudAgentReportStore(fake.db as never); + + expect(await store.saveReport(anchoredReport, occurredAt)).toEqual({ outcome: 'conflict' }); + expect(fake.inserts).toHaveLength(0); + }); + it('persists run milestones, typed failure and sanitized detail by natural composite key', async () => { const fake = makeDb([[{ createdAt: occurredAt }], []]); const store = createCloudAgentReportStore(fake.db as never); diff --git a/services/cloud-agent-next/src/telemetry/report-store.ts b/services/cloud-agent-next/src/telemetry/report-store.ts index ea6d415b8b..ee41727fca 100644 --- a/services/cloud-agent-next/src/telemetry/report-store.ts +++ b/services/cloud-agent-next/src/telemetry/report-store.ts @@ -68,7 +68,12 @@ const recordSessionFailureSchema = z type DatabaseTransaction = Parameters[0]>[0]; type MutationResult = { applied?: boolean }; -type SaveReportResult = { outcome: 'applied' | 'expired' | 'missing_parent' }; +type SaveReportResult = { outcome: 'applied' | 'expired' | 'missing_parent' | 'conflict' }; +type ReportingParent = { + createdAt: string; + kiloSessionId: string | null; + initialMessageId: string | null; +}; type StoredRunRow = { status: 'queued' | 'accepted' | 'completed' | 'failed' | 'interrupted'; wrapperRunId: string | null; @@ -90,6 +95,76 @@ function retentionCutoff(now: string): string { return cutoff.toISOString(); } +async function readReportingParent( + tx: DatabaseTransaction, + cloudAgentSessionId: string +): Promise { + const rows = await tx + .select({ + createdAt: cloud_agent_sessions.created_at, + kiloSessionId: cloud_agent_sessions.kilo_session_id, + initialMessageId: cloud_agent_sessions.initial_message_id, + }) + .from(cloud_agent_sessions) + .where(eq(cloud_agent_sessions.cloud_agent_session_id, cloudAgentSessionId)) + .limit(1); + return rows[0]; +} + +function parentConflictsWithAnchor( + parent: ReportingParent, + report: CloudAgentRunStateReport +): boolean { + const { kiloSessionId, initialMessageId } = report.session; + if (kiloSessionId === undefined || initialMessageId === undefined) return false; + return ( + (parent.kiloSessionId != null && parent.kiloSessionId !== kiloSessionId) || + (parent.initialMessageId != null && parent.initialMessageId !== initialMessageId) + ); +} + +/** + * Ensures a reporting parent for a control-plane session that never had an + * initial turn (worktree and empty chats). The trusted anchor is written at + * admission; the parent is inserted only inside the retention window and is + * never updated or refreshed. A concurrent writer may already have created it, + * so the insert tolerates conflicts and the caller re-reads. + */ +async function ensureReportingParent( + tx: DatabaseTransaction, + report: CloudAgentRunStateReport, + now: string +): Promise { + const { kiloSessionId, initialMessageId, reportingCreatedAt } = report.session; + if ( + kiloSessionId === undefined || + initialMessageId === undefined || + reportingCreatedAt === undefined + ) { + return undefined; + } + if (Date.parse(reportingCreatedAt) <= Date.parse(retentionCutoff(now))) return undefined; + + await tx + .insert(cloud_agent_sessions) + .values({ + cloud_agent_session_id: report.session.cloudAgentSessionId, + kilo_session_id: kiloSessionId, + initial_message_id: initialMessageId, + created_at: reportingCreatedAt, + }) + .onConflictDoNothing(); + + const ensured = await readReportingParent(tx, report.session.cloudAgentSessionId); + if (!ensured) { + console.error('Cloud Agent report could not ensure its session parent', { + cloudAgentSessionId: report.session.cloudAgentSessionId, + }); + return undefined; + } + return ensured; +} + async function lockReportingSession( tx: DatabaseTransaction, cloudAgentSessionId: string @@ -145,13 +220,18 @@ export function createCloudAgentReportStore(db: WorkerDb) { now: string ): Promise { const cloudAgentSessionId = report.session.cloudAgentSessionId; - const parentRows = await tx - .select({ createdAt: cloud_agent_sessions.created_at }) - .from(cloud_agent_sessions) - .where(eq(cloud_agent_sessions.cloud_agent_session_id, cloudAgentSessionId)) - .limit(1); - const parent = parentRows[0]; - if (!parent) return { outcome: 'missing_parent' }; + let parent = await readReportingParent(tx, cloudAgentSessionId); + if (!parent) { + parent = await ensureReportingParent(tx, report, now); + if (!parent) return { outcome: 'missing_parent' }; + } + if (parentConflictsWithAnchor(parent, report)) { + console.error('Cloud Agent report anchor conflicts with existing session parent', { + cloudAgentSessionId, + messageId: report.run.messageId, + }); + return { outcome: 'conflict' }; + } if (Date.parse(parent.createdAt) <= Date.parse(retentionCutoff(now))) { return { outcome: 'expired' }; } diff --git a/services/cloud-agent-next/test/integration/sandbox-control.test.ts b/services/cloud-agent-next/test/integration/sandbox-control.test.ts index f389758ba4..0dc2dd373d 100644 --- a/services/cloud-agent-next/test/integration/sandbox-control.test.ts +++ b/services/cloud-agent-next/test/integration/sandbox-control.test.ts @@ -10963,19 +10963,26 @@ describe('SandboxSession control-plane regressions', () => { ).resolves.toMatchObject({ success: true, messageId: followUpTurn.id }); expect(await state.storage.get('session_messages')).toEqual([ blocker, - createSessionMessageRecord({ - turn: initialTurn, - agent: { mode: 'code', model: 'test' }, - }), - createSessionMessageRecord({ - turn: { - type: 'command', - messageId: followUpTurn.id, - command: followUpTurn.command, - arguments: followUpTurn.arguments, - }, - agent: { mode: 'code', model: 'test' }, - }), + { + ...createSessionMessageRecord({ + turn: initialTurn, + agent: { mode: 'code', model: 'test' }, + }), + // Admission now persists a stable queue timestamp for reporting. + queuedAt: expect.any(Number), + }, + { + ...createSessionMessageRecord({ + turn: { + type: 'command', + messageId: followUpTurn.id, + command: followUpTurn.command, + arguments: followUpTurn.arguments, + }, + agent: { mode: 'code', model: 'test' }, + }), + queuedAt: expect.any(Number), + }, ]); }); }); @@ -12974,16 +12981,18 @@ describe('SandboxSession worktree admission', () => { finalization: submission.finalization, }) ).resolves.toMatchObject({ success: true, messageId }); - expectedMessages.push( - createSessionMessageRecord({ + expectedMessages.push({ + ...createSessionMessageRecord({ turn: { type: 'prompt', messageId, prompt: 'follow-up' }, agent: { mode: 'code', model: 'test-model' }, finalization: { autoCommit: submission.autoCommit, condenseOnComplete: submission.condenseOnComplete, }, - }) - ); + }), + // Admission now persists a stable queue timestamp for reporting. + queuedAt: expect.any(Number), + }); expect(state.storage.kv.get('session_messages')).toEqual(expectedMessages); expect(await instance.getMetadata()).toEqual(metadata); } diff --git a/services/cloud-agent-next/test/integration/sandbox-session-reports.test.ts b/services/cloud-agent-next/test/integration/sandbox-session-reports.test.ts new file mode 100644 index 0000000000..5200304360 --- /dev/null +++ b/services/cloud-agent-next/test/integration/sandbox-session-reports.test.ts @@ -0,0 +1,435 @@ +/** + * Integration tests for control-plane Cloud Agent run-state reporting. + * + * These drive the real `SandboxSession` Durable Object through admission, + * acceptance, terminal, and deletion commits with a captured + * `CLOUD_AGENT_REPORT_QUEUE`. They reach the DO transaction boundary and the + * durable report outbox; they do not reach the report consumer, PostgreSQL + * store, admin views, or the sandbox control plane (dispatch is suppressed + * because these sessions have no provisioned sandbox). + */ +import { abortAllDurableObjects, env, reset, runInDurableObject } from 'cloudflare:test'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import type { CloudAgentQueueReport } from '@kilocode/worker-utils/cloud-agent-queue-report'; +import type { SandboxSession } from '../../src/sandbox-session/SandboxSession.js'; +import type { + SessionMessageRecord, + SessionMessageTerminalSource, +} from '../../src/sandbox-session/session-message-queue.js'; +import { applyMessageOutcome } from '../../src/sandbox-session/session-message-queue.js'; +import { + REPORT_OUTBOX_PREFIX, + parsePendingRunReport, + readReportAnchor, +} from '../../src/sandbox-session/report-outbox.js'; + +const ownerId = 'report-owner'; +const kiloSessionId = 'ses_12345678901234567890123456'; +const agent = { mode: 'code' as const, model: 'anthropic/claude-sonnet-4' }; + +function sessionId(): string { + return `workspace_${crypto.randomUUID()}`; +} + +function sessionStub(id: string) { + return env.SANDBOX_SESSION.getByName(`${ownerId}:${id}`); +} + +function injectReportQueue(instance: SandboxSession, captured: CloudAgentQueueReport[]): void { + ( + instance as unknown as { env: { CLOUD_AGENT_REPORT_QUEUE: { send: unknown } } } + ).env.CLOUD_AGENT_REPORT_QUEUE = { + send: async (report: CloudAgentQueueReport) => { + captured.push(report); + }, + }; +} + +function suppressDispatch(instance: SandboxSession): void { + (instance as unknown as Record)['deliverQueuedMessage'] = async () => + undefined; +} + +function readMessages(state: DurableObjectState): SessionMessageRecord[] { + return state.storage.kv.get('session_messages') ?? []; +} + +function readObligation( + state: DurableObjectState, + messageId: string +): ReturnType { + return parsePendingRunReport( + state.storage.kv.get(`${REPORT_OUTBOX_PREFIX}${messageId}`) + ); +} + +async function register(instance: SandboxSession, id: string): Promise { + await instance.registerSession({ + identity: { sessionId: id, userId: ownerId }, + auth: { kiloSessionId, kilocodeToken: 'test-token' }, + agent, + }); +} + +function admit(instance: SandboxSession, messageId: string): Promise { + return instance.admitSubmittedMessage({ + userId: ownerId, + turn: { type: 'prompt', id: messageId, prompt: `prompt ${messageId}` }, + agent, + }); +} + +beforeEach(() => { + vi.spyOn(globalThis, 'fetch').mockImplementation(async () => Response.json({ valid: true })); +}); + +afterEach(async () => { + await reset(); + vi.restoreAllMocks(); +}); + +describe('control-plane run-state reporting', () => { + it('commits a queued obligation with the first-message anchor before any queue send', async () => { + const id = sessionId(); + const messageId = 'msg_018f1e2d3c4bQueuedReportAB'; + const captured: CloudAgentQueueReport[] = []; + + const result = await runInDurableObject(sessionStub(id), async (instance, state) => { + injectReportQueue(instance, captured); + suppressDispatch(instance); + await register(instance, id); + await admit(instance, messageId); + const first = readMessages(state).find(message => message.messageId === messageId); + const obligation = readObligation(state, messageId); + const anchor = readReportAnchor(state.storage); + const replay = await admit(instance, messageId); + const second = readMessages(state).find(message => message.messageId === messageId); + const obligationAfterReplay = readObligation(state, messageId); + return { first, second, obligation, obligationAfterReplay, anchor, replay }; + }); + + expect(captured).toEqual([]); + expect(result.first?.queuedAt).toEqual(expect.any(Number)); + expect(result.second?.queuedAt).toBe(result.first?.queuedAt); + expect(result.replay).toMatchObject({ success: true, outcome: 'queued' }); + expect(result.anchor).toEqual({ + version: 1, + kiloSessionId, + initialMessageId: messageId, + createdAt: expect.any(Number), + }); + expect(result.obligation?.obligationId).toEqual(expect.any(String)); + expect(result.obligationAfterReplay?.obligationId).toBe(result.obligation?.obligationId); + expect(result.obligation?.report.session).toEqual({ + cloudAgentSessionId: id, + kiloSessionId, + initialMessageId: messageId, + reportingCreatedAt: new Date(result.anchor?.createdAt ?? 0).toISOString(), + }); + expect(result.obligation?.report.run).toMatchObject({ + messageId, + status: 'queued', + queuedAt: new Date(result.first?.queuedAt ?? 0).toISOString(), + }); + expect(result.obligation?.report.run).not.toHaveProperty('dispatchAcceptedAt'); + expect(result.obligation?.report.run).not.toHaveProperty('terminalAt'); + }); + + it('never anchors an existing session using a later follow-up id', async () => { + const id = sessionId(); + const seededMessageId = 'msg_018f1e2d3c4bSeededExisting'; + const followUpId = 'msg_018f1e2d3c4bLaterFollowUpA'; + const captured: CloudAgentQueueReport[] = []; + + const result = await runInDurableObject(sessionStub(id), async (instance, state) => { + injectReportQueue(instance, captured); + suppressDispatch(instance); + await register(instance, id); + // Pre-existing message and no anchor: a legacy/unanchored session. + state.storage.kv.put('session_messages', [ + { messageId: seededMessageId, state: 'completed', terminalAt: Date.now() }, + ]); + await admit(instance, followUpId); + return { + anchor: readReportAnchor(state.storage), + obligation: readObligation(state, followUpId), + seeded: readMessages(state).some(message => message.messageId === seededMessageId), + }; + }); + + expect(result.anchor).toBeUndefined(); + expect(result.seeded).toBe(true); + expect(result.obligation?.report.session).toEqual({ cloudAgentSessionId: id }); + expect(result.obligation?.report.session).not.toHaveProperty('initialMessageId'); + }); + + it('aborts admission and rolls back the message when the report obligation cannot be written', async () => { + const id = sessionId(); + const messageId = 'msg_018f1e2d3c4bAtomicAbortAB'; + + const result = await runInDurableObject(sessionStub(id), async (instance, state) => { + suppressDispatch(instance); + await register(instance, id); + const original = (instance as unknown as { reportOutbox: unknown }).reportOutbox; + (instance as unknown as { reportOutbox: unknown }).reportOutbox = { + ...(original as object), + record: () => { + throw new Error('report obligation write failed'); + }, + }; + let succeeded = false; + let threw = false; + try { + const admission = (await admit(instance, messageId)) as { success?: boolean }; + succeeded = admission.success === true; + } catch { + threw = true; + } finally { + (instance as unknown as { reportOutbox: unknown }).reportOutbox = original; + } + return { succeeded, threw, messages: readMessages(state) }; + }); + + expect(result.succeeded).toBe(false); + expect(result.threw || !result.succeeded).toBe(true); + expect(result.messages.some(message => message.messageId === messageId)).toBe(false); + }); + + it('reports observed acceptance with no fabricated terminal facts', async () => { + const id = sessionId(); + const messageId = 'msg_018f1e2d3c4bAccepteedAB'; + const captured: CloudAgentQueueReport[] = []; + const acceptedAt = Date.now(); + + await runInDurableObject(sessionStub(id), async (instance, state) => { + injectReportQueue(instance, captured); + suppressDispatch(instance); + await register(instance, id); + await admit(instance, messageId); + const messages = readMessages(state).map(message => + message.messageId === messageId + ? { ...message, state: 'accepted' as const, acceptedAt } + : message + ); + instance['saveMessages'](messages); + await instance.alarm(); + }); + + const accepted = captured.find(report => report.run.status === 'accepted'); + expect(accepted?.run).toMatchObject({ + messageId, + status: 'accepted', + dispatchAcceptedAt: new Date(acceptedAt).toISOString(), + }); + expect(accepted?.run).not.toHaveProperty('terminalAt'); + expect(accepted?.run).not.toHaveProperty('failureStage'); + }); + + it('classifies a pre-dispatch failure with queued and terminal timestamps', async () => { + const id = sessionId(); + const messageId = 'msg_018f1e2d3c4bPredispatchAB'; + const captured: CloudAgentQueueReport[] = []; + + const queuedAt = await runInDurableObject(sessionStub(id), async (instance, state) => { + injectReportQueue(instance, captured); + suppressDispatch(instance); + await register(instance, id); + await admit(instance, messageId); + const queued = readMessages(state).find(message => message.messageId === messageId)?.queuedAt; + await instance.failWaitingMessages('missing_metadata'); + await instance.alarm(); + return queued; + }); + + const failed = captured.find(report => report.run.status === 'failed'); + expect(failed?.run).toMatchObject({ + messageId, + status: 'failed', + queuedAt: new Date(queuedAt ?? 0).toISOString(), + terminalAt: expect.any(String), + failureStage: 'pre_dispatch', + failureCode: 'session_metadata_missing', + failureResponsibility: 'platform', + failureReason: 'delivery', + diagnostic: { errorMessageRedacted: 'Session metadata is unavailable' }, + }); + expect(failed?.run).not.toHaveProperty('dispatchAcceptedAt'); + expect(failed?.run).not.toHaveProperty('agentActivityObservedAt'); + }); + + it('classifies a coordinator preparation timeout as platform runtime startup', async () => { + const id = sessionId(); + const messageId = 'msg_018f1e2d3c4bPrepTimeoutAB'; + const captured: CloudAgentQueueReport[] = []; + + await runInDurableObject(sessionStub(id), async (instance, state) => { + injectReportQueue(instance, captured); + suppressDispatch(instance); + await register(instance, id); + await admit(instance, messageId); + await instance.failWaitingMessages('preparation_timeout'); + await instance.alarm(); + }); + + const failed = captured.find(report => report.run.status === 'failed'); + expect(failed?.run).toMatchObject({ + messageId, + status: 'failed', + failureStage: 'pre_dispatch', + failureCode: 'wrapper_start_failed', + failureResponsibility: 'platform', + failureReason: 'runtime_startup', + }); + }); + + it('classifies post-acceptance failure without losing observed acceptance', async () => { + const id = sessionId(); + const messageId = 'msg_018f1e2d3c4bPostAcceptAB'; + const captured: CloudAgentQueueReport[] = []; + const acceptedAt = Date.now(); + + await runInDurableObject(sessionStub(id), async (instance, state) => { + injectReportQueue(instance, captured); + suppressDispatch(instance); + await register(instance, id); + await admit(instance, messageId); + instance['saveMessages']( + readMessages(state).map(message => + message.messageId === messageId + ? { ...message, state: 'accepted' as const, acceptedAt } + : message + ) + ); + await instance.failWaitingMessages('runtime_unhealthy'); + await instance.alarm(); + }); + + const failed = captured.find(report => report.run.status === 'failed'); + expect(failed?.run).toMatchObject({ + messageId, + status: 'failed', + dispatchAcceptedAt: new Date(acceptedAt).toISOString(), + failureStage: 'post_dispatch_no_activity', + failureCode: 'wrapper_disconnected', + }); + }); + + it.each(['wrapper_outcome', 'operation_result'] as const)( + 'does not treat an inferred acceptedAt as observed acceptance for a %s terminal-before-ACK', + async terminalSource => { + const id = sessionId(); + const messageId = 'msg_018f1e2d3c4bTermBeforeAck'; + const captured: CloudAgentQueueReport[] = []; + const wrapperInstanceId = 'wr_terminal_before_ack'; + + const applied = await runInDurableObject(sessionStub(id), async (instance, state) => { + injectReportQueue(instance, captured); + suppressDispatch(instance); + await register(instance, id); + await admit(instance, messageId); + // The wrapper runtime is fenced so the real outcome mapper applies. + const messages = readMessages(state).map(message => + message.messageId === messageId ? { ...message, wrapperInstanceId } : message + ); + const updated = applyMessageOutcome( + messages, + { messageId, status: 'failed', reason: 'missing_metadata' }, + wrapperInstanceId, + Date.now(), + terminalSource + ); + expect(updated).toBeDefined(); + expect( + updated?.find(message => message.messageId === messageId)?.acceptedAt + ).toEqual(expect.any(Number)); + state.storage.kv.put('session_messages', messages); + instance['saveMessages'](updated as SessionMessageRecord[]); + await instance.alarm(); + return true; + }); + + expect(applied).toBe(true); + const failed = captured.find(report => report.run.status === 'failed'); + // Arbitrary wrapper/operation text is not a known coordinator cause. + expect(failed?.run).toMatchObject({ + messageId, + status: 'failed', + failureStage: 'unknown', + failureCode: 'unclassified', + failureResponsibility: 'unknown', + failureReason: 'unclassified', + }); + expect(failed?.run).not.toHaveProperty('dispatchAcceptedAt'); + } + ); + + it('reports a cancelled queued message as an interruption without dispatch acceptance', async () => { + const id = sessionId(); + const messageId = 'msg_018f1e2d3c4bCancelReportAB'; + const captured: CloudAgentQueueReport[] = []; + + const queuedAt = await runInDurableObject(sessionStub(id), async (instance, state) => { + injectReportQueue(instance, captured); + suppressDispatch(instance); + await register(instance, id); + await admit(instance, messageId); + const queued = readMessages(state).find(message => message.messageId === messageId)?.queuedAt; + expect(await instance.cancelQueuedMessage(messageId)).toEqual({ dropped: true }); + await instance.alarm(); + return queued; + }); + + const interrupted = captured.find(report => report.run.status === 'interrupted'); + expect(interrupted?.run).toMatchObject({ + messageId, + status: 'interrupted', + queuedAt: new Date(queuedAt ?? 0).toISOString(), + terminalAt: expect.any(String), + failureStage: 'interruption', + failureCode: 'user_interrupt', + }); + expect(interrupted?.run).not.toHaveProperty('dispatchAcceptedAt'); + }); + + it('recovers a failed send after eviction and delivers it on the next alarm', async () => { + const id = sessionId(); + const messageId = 'msg_018f1e2d3c4bRecoveryABcDE'; + const captured: CloudAgentQueueReport[] = []; + + await runInDurableObject(sessionStub(id), async (instance, state) => { + ( + instance as unknown as { env: { CLOUD_AGENT_REPORT_QUEUE: { send: unknown } } } + ).env.CLOUD_AGENT_REPORT_QUEUE = { + send: async () => { + throw new Error('report queue unavailable'); + }, + }; + suppressDispatch(instance); + await register(instance, id); + await admit(instance, messageId); + await instance.alarm(); + + const pending = readObligation(state, messageId); + expect(pending?.attempts).toBe(1); + expect(await state.storage.getAlarm()).not.toBeNull(); + + // Simulate the retry interval elapsing without waiting 30 seconds. + state.storage.kv.put(`${REPORT_OUTBOX_PREFIX}${messageId}`, { + ...pending, + dueAt: 0, + }); + }); + + await abortAllDurableObjects(); + + await runInDurableObject(sessionStub(id), async (instance, state) => { + injectReportQueue(instance, captured); + suppressDispatch(instance); + await instance.alarm(); + expect(readObligation(state, messageId)).toBeUndefined(); + }); + + expect(captured).toHaveLength(1); + expect(captured[0].run).toMatchObject({ messageId, status: 'queued' }); + }); +}); diff --git a/services/cloud-agent-next/test/integration/worktree-deletion.test.ts b/services/cloud-agent-next/test/integration/worktree-deletion.test.ts index dcb0788490..43ad8b6191 100644 --- a/services/cloud-agent-next/test/integration/worktree-deletion.test.ts +++ b/services/cloud-agent-next/test/integration/worktree-deletion.test.ts @@ -3,6 +3,11 @@ import type { CloudAgentWorktreeId } from '@kilocode/session-ingest-contracts'; import { drizzle } from 'drizzle-orm/durable-sqlite'; import { afterEach, describe, it, expect, vi } from 'vitest'; import { getWorktreeWorkspacePath } from '../../src/workspace'; +import { CALLBACK_OUTBOX_PREFIX } from '../../src/sandbox-session/message-callbacks'; +import { + REPORT_OUTBOX_PREFIX, + parsePendingRunReport, +} from '../../src/sandbox-session/report-outbox'; import { events } from '../../src/db/sqlite-schema'; import { generateSandboxCredential, @@ -993,7 +998,9 @@ describe('worktree deletion in Durable Objects', () => { await expect(closed).resolves.toBe(1001); await runInDurableObject(stub, async (_instance, state) => { expect(await state.storage.get('session_messages')).toMatchObject([{ state: 'cancelled' }]); - expect(await state.storage.getAlarm()).toBeNull(); + // Deletion preserves the interrupted report obligation, so its delivery + // alarm is intentionally armed instead of removed. + expect(await state.storage.getAlarm()).not.toBeNull(); }); await stub.finishWorktreeDeletion(worktreeId); await expect(stub.getRuntimeLocation()).resolves.toBeNull(); @@ -1077,6 +1084,7 @@ describe('worktree deletion in Durable Objects', () => { const attach = vi.fn(); const request = vi.fn(); const original = instance['env'].SANDBOX_CONTROL; + const originalReportQueue = instance['env'].CLOUD_AGENT_REPORT_QUEUE; const validation = vi .spyOn(globalThis, 'fetch') .mockImplementation(async () => Response.json({ valid: true })); @@ -1097,6 +1105,13 @@ describe('worktree deletion in Durable Objects', () => { request, }), }, + // Keep the reporting obligation pending so deletion's preserved + // report-delivery alarm is observable instead of racing a live send. + CLOUD_AGENT_REPORT_QUEUE: { + send: async () => { + throw new Error('report queue unavailable'); + }, + }, }); try { await instance.registerSession(registration(sessionId, sandboxId)); @@ -1110,13 +1125,26 @@ describe('worktree deletion in Durable Objects', () => { expect(attach).not.toHaveBeenCalled(); expect(request).not.toHaveBeenCalled(); expect(await state.storage.get('session_messages')).toBeUndefined(); - expect(await state.storage.getAlarm()).toBeNull(); + // Deletion preserves report delivery state. The alarm must be armed + // solely for the interrupted report obligation, not callbacks. + expect([ + ...state.storage.kv.list({ prefix: CALLBACK_OUTBOX_PREFIX }), + ]).toHaveLength(0); + const reportEntries = [ + ...state.storage.kv.list({ prefix: REPORT_OUTBOX_PREFIX }), + ]; + expect(reportEntries).toHaveLength(1); + expect(parsePendingRunReport(reportEntries[0]?.[1])?.report.run.status).toBe('interrupted'); + expect(await state.storage.getAlarm()).not.toBeNull(); } finally { try { await instance.beginWorktreeDeletion(deletionInput); } finally { ready.resolve({ physical: 'running', connection: 'ready' }); - Object.assign(instance['env'], { SANDBOX_CONTROL: original }); + Object.assign(instance['env'], { + SANDBOX_CONTROL: original, + CLOUD_AGENT_REPORT_QUEUE: originalReportQueue, + }); validation.mockRestore(); } await instance.finishWorktreeDeletion(worktreeId); From b6b6372f68dda191d05e1bea572e9681cc411998 Mon Sep 17 00:00:00 2001 From: Evgeny Shurakov Date: Fri, 11 Sep 2026 12:42:37 +0200 Subject: [PATCH 2/2] fix(cloud-agent-next): classify accepted provider_unknown failures as post-dispatch An accepted run that failed as provider_unknown was reported as pre_dispatch/sandbox_connect_failed while also carrying dispatchAcceptedAt. Branch on the observed dispatch state like the sibling reasons, so an accepted run reports post_dispatch_no_activity/wrapper_disconnected. Also drop a tautological assertion in the reports integration test and apply oxfmt to the files CI flagged. --- .../src/cloud-agent-queue-report.ts | 8 ++- .../src/sandbox-session/SandboxSession.ts | 20 +++--- .../src/sandbox-session/report-outbox.test.ts | 12 ++-- .../telemetry/control-plane-failure.test.ts | 3 +- .../src/telemetry/control-plane-failure.ts | 6 +- .../sandbox-session-reports.test.ts | 71 ++++++++++--------- .../integration/worktree-deletion.test.ts | 4 +- 7 files changed, 66 insertions(+), 58 deletions(-) diff --git a/packages/worker-utils/src/cloud-agent-queue-report.ts b/packages/worker-utils/src/cloud-agent-queue-report.ts index 1c94665950..3819fc9e9d 100644 --- a/packages/worker-utils/src/cloud-agent-queue-report.ts +++ b/packages/worker-utils/src/cloud-agent-queue-report.ts @@ -68,9 +68,11 @@ const CloudAgentQueueSessionIdentitySchema = z }) .strict() .superRefine((session, ctx) => { - const present = [session.kiloSessionId, session.initialMessageId, session.reportingCreatedAt].filter( - value => value !== undefined - ).length; + const present = [ + session.kiloSessionId, + session.initialMessageId, + session.reportingCreatedAt, + ].filter(value => value !== undefined).length; if (present !== 0 && present !== 3) { ctx.addIssue({ code: 'custom', diff --git a/services/cloud-agent-next/src/sandbox-session/SandboxSession.ts b/services/cloud-agent-next/src/sandbox-session/SandboxSession.ts index b7f4b984ce..b1405f1642 100644 --- a/services/cloud-agent-next/src/sandbox-session/SandboxSession.ts +++ b/services/cloud-agent-next/src/sandbox-session/SandboxSession.ts @@ -201,10 +201,7 @@ import { type CloudAgentQueueReport, type CloudAgentRunStateReport, } from '@kilocode/worker-utils/cloud-agent-queue-report'; -import { - buildRunStateReport, - FAILED_RUN_DIAGNOSTIC_MESSAGES, -} from '../telemetry/queue-reports.js'; +import { buildRunStateReport, FAILED_RUN_DIAGNOSTIC_MESSAGES } from '../telemetry/queue-reports.js'; import { classifyControlPlaneFailure } from '../telemetry/control-plane-failure.js'; import { classifyCloudAgentFailure } from '@kilocode/worker-utils/cloud-agent-failure'; import { PENDING_SESSION_MESSAGE_LIMIT } from '../session/pending-messages.js'; @@ -1697,7 +1694,10 @@ export class SandboxSession extends DurableObject { if (this.reportOutbox.pendingCount() > 0) this.scheduleReportRepair(); this.deletedWorktreeId = worktreeId; for (const socket of this.ctx.getWebSockets()) socket.close(1001, 'Worktree deleted'); - if (this.messageCallbacks.pendingCallbackCount() === 0 && this.reportOutbox.pendingCount() === 0) + if ( + this.messageCallbacks.pendingCallbackCount() === 0 && + this.reportOutbox.pendingCount() === 0 + ) await this.ctx.storage.deleteAlarm(); if (!metadata) return null; return cloudAgentWorktreeLocationSchema.parse({ @@ -1817,7 +1817,11 @@ export class SandboxSession extends DurableObject { initialMessageId: firstMessageId, createdAt, }; - writeReportAnchor(this.ctx.storage, { kiloSessionId, initialMessageId: firstMessageId, createdAt }); + writeReportAnchor(this.ctx.storage, { + kiloSessionId, + initialMessageId: firstMessageId, + createdAt, + }); return anchor; } @@ -1875,9 +1879,7 @@ export class SandboxSession extends DurableObject { errorMessageRedacted: FAILED_RUN_DIAGNOSTIC_MESSAGES[classification.code] ?? 'Run failed without a classified cause', - errorExpiresAt: new Date( - message.terminalAt + DIAGNOSTIC_RETENTION_MS - ).toISOString(), + errorExpiresAt: new Date(message.terminalAt + DIAGNOSTIC_RETENTION_MS).toISOString(), }; } } diff --git a/services/cloud-agent-next/src/sandbox-session/report-outbox.test.ts b/services/cloud-agent-next/src/sandbox-session/report-outbox.test.ts index aee939319f..cdaeba32df 100644 --- a/services/cloud-agent-next/src/sandbox-session/report-outbox.test.ts +++ b/services/cloud-agent-next/src/sandbox-session/report-outbox.test.ts @@ -64,9 +64,11 @@ function failedReport(messageId: string): CloudAgentQueueReport { }; } -function createHarness(options: { - queue?: { send: (report: CloudAgentQueueReport) => Promise }; -} = {}) { +function createHarness( + options: { + queue?: { send: (report: CloudAgentQueueReport) => Promise }; + } = {} +) { const kv = memoryKv(); const outbox = createReportOutbox({ storage: { kv } as never, @@ -115,7 +117,9 @@ describe('createReportOutbox', () => { }, }); outbox.record(queuedReport('msg_one')); - const sentObligationId = parsePendingRunReport(kv.get(reportOutboxKey('msg_one')))?.obligationId; + const sentObligationId = parsePendingRunReport( + kv.get(reportOutboxKey('msg_one')) + )?.obligationId; await outbox.repair(); diff --git a/services/cloud-agent-next/src/telemetry/control-plane-failure.test.ts b/services/cloud-agent-next/src/telemetry/control-plane-failure.test.ts index 1678114af7..ae8c95e931 100644 --- a/services/cloud-agent-next/src/telemetry/control-plane-failure.test.ts +++ b/services/cloud-agent-next/src/telemetry/control-plane-failure.test.ts @@ -25,7 +25,8 @@ const cases: ReadonlyArray< ['prompt_exhausted', 'pre_dispatch', 'failed', 'pre_dispatch', 'invalid_delivery_request'], ['environment_failed', 'accepted', 'failed', 'post_dispatch_no_activity', 'wrapper_disconnected'], ['environment_failed', 'pre_dispatch', 'failed', 'pre_dispatch', 'sandbox_connect_failed'], - ['provider_unknown', 'accepted', 'failed', 'pre_dispatch', 'sandbox_connect_failed'], + ['provider_unknown', 'accepted', 'failed', 'post_dispatch_no_activity', 'wrapper_disconnected'], + ['provider_unknown', 'pre_dispatch', 'failed', 'pre_dispatch', 'sandbox_connect_failed'], ['runtime_unhealthy', 'accepted', 'failed', 'post_dispatch_no_activity', 'wrapper_disconnected'], ['runtime_unhealthy', 'pre_dispatch', 'failed', 'pre_dispatch', 'wrapper_start_failed'], ['accepted_overdue', 'pre_dispatch', 'failed', 'post_dispatch_no_activity', 'wrapper_no_output'], diff --git a/services/cloud-agent-next/src/telemetry/control-plane-failure.ts b/services/cloud-agent-next/src/telemetry/control-plane-failure.ts index 32d565b179..fe0d64090c 100644 --- a/services/cloud-agent-next/src/telemetry/control-plane-failure.ts +++ b/services/cloud-agent-next/src/telemetry/control-plane-failure.ts @@ -68,11 +68,11 @@ export function classifyControlPlaneFailure( ? POST_DISPATCH_WRAPPER_DISCONNECTED : PRE_DISPATCH_SANDBOX_CONNECT; case 'provider_unknown': - return PRE_DISPATCH_SANDBOX_CONNECT; - case 'runtime_unhealthy': return dispatchState === 'accepted' ? POST_DISPATCH_WRAPPER_DISCONNECTED - : PRE_DISPATCH; + : PRE_DISPATCH_SANDBOX_CONNECT; + case 'runtime_unhealthy': + return dispatchState === 'accepted' ? POST_DISPATCH_WRAPPER_DISCONNECTED : PRE_DISPATCH; case 'accepted_overdue': return { stage: 'post_dispatch_no_activity', code: 'wrapper_no_output' }; case 'invalid_model': diff --git a/services/cloud-agent-next/test/integration/sandbox-session-reports.test.ts b/services/cloud-agent-next/test/integration/sandbox-session-reports.test.ts index 5200304360..e1916773b2 100644 --- a/services/cloud-agent-next/test/integration/sandbox-session-reports.test.ts +++ b/services/cloud-agent-next/test/integration/sandbox-session-reports.test.ts @@ -46,8 +46,7 @@ function injectReportQueue(instance: SandboxSession, captured: CloudAgentQueueRe } function suppressDispatch(instance: SandboxSession): void { - (instance as unknown as Record)['deliverQueuedMessage'] = async () => - undefined; + (instance as unknown as Record)['deliverQueuedMessage'] = async () => undefined; } function readMessages(state: DurableObjectState): SessionMessageRecord[] { @@ -191,7 +190,6 @@ describe('control-plane run-state reporting', () => { }); expect(result.succeeded).toBe(false); - expect(result.threw || !result.succeeded).toBe(true); expect(result.messages.some(message => message.messageId === messageId)).toBe(false); }); @@ -282,37 +280,40 @@ describe('control-plane run-state reporting', () => { }); }); - it('classifies post-acceptance failure without losing observed acceptance', async () => { - const id = sessionId(); - const messageId = 'msg_018f1e2d3c4bPostAcceptAB'; - const captured: CloudAgentQueueReport[] = []; - const acceptedAt = Date.now(); + it.each(['runtime_unhealthy', 'provider_unknown'] as const)( + 'classifies post-acceptance failure without losing observed acceptance for %s', + async reason => { + const id = sessionId(); + const messageId = 'msg_018f1e2d3c4bPostAcceptAB'; + const captured: CloudAgentQueueReport[] = []; + const acceptedAt = Date.now(); - await runInDurableObject(sessionStub(id), async (instance, state) => { - injectReportQueue(instance, captured); - suppressDispatch(instance); - await register(instance, id); - await admit(instance, messageId); - instance['saveMessages']( - readMessages(state).map(message => - message.messageId === messageId - ? { ...message, state: 'accepted' as const, acceptedAt } - : message - ) - ); - await instance.failWaitingMessages('runtime_unhealthy'); - await instance.alarm(); - }); + await runInDurableObject(sessionStub(id), async (instance, state) => { + injectReportQueue(instance, captured); + suppressDispatch(instance); + await register(instance, id); + await admit(instance, messageId); + instance['saveMessages']( + readMessages(state).map(message => + message.messageId === messageId + ? { ...message, state: 'accepted' as const, acceptedAt } + : message + ) + ); + await instance.failWaitingMessages(reason); + await instance.alarm(); + }); - const failed = captured.find(report => report.run.status === 'failed'); - expect(failed?.run).toMatchObject({ - messageId, - status: 'failed', - dispatchAcceptedAt: new Date(acceptedAt).toISOString(), - failureStage: 'post_dispatch_no_activity', - failureCode: 'wrapper_disconnected', - }); - }); + const failed = captured.find(report => report.run.status === 'failed'); + expect(failed?.run).toMatchObject({ + messageId, + status: 'failed', + dispatchAcceptedAt: new Date(acceptedAt).toISOString(), + failureStage: 'post_dispatch_no_activity', + failureCode: 'wrapper_disconnected', + }); + } + ); it.each(['wrapper_outcome', 'operation_result'] as const)( 'does not treat an inferred acceptedAt as observed acceptance for a %s terminal-before-ACK', @@ -339,9 +340,9 @@ describe('control-plane run-state reporting', () => { terminalSource ); expect(updated).toBeDefined(); - expect( - updated?.find(message => message.messageId === messageId)?.acceptedAt - ).toEqual(expect.any(Number)); + expect(updated?.find(message => message.messageId === messageId)?.acceptedAt).toEqual( + expect.any(Number) + ); state.storage.kv.put('session_messages', messages); instance['saveMessages'](updated as SessionMessageRecord[]); await instance.alarm(); diff --git a/services/cloud-agent-next/test/integration/worktree-deletion.test.ts b/services/cloud-agent-next/test/integration/worktree-deletion.test.ts index 43ad8b6191..9ae9c3d5ee 100644 --- a/services/cloud-agent-next/test/integration/worktree-deletion.test.ts +++ b/services/cloud-agent-next/test/integration/worktree-deletion.test.ts @@ -1130,9 +1130,7 @@ describe('worktree deletion in Durable Objects', () => { expect([ ...state.storage.kv.list({ prefix: CALLBACK_OUTBOX_PREFIX }), ]).toHaveLength(0); - const reportEntries = [ - ...state.storage.kv.list({ prefix: REPORT_OUTBOX_PREFIX }), - ]; + const reportEntries = [...state.storage.kv.list({ prefix: REPORT_OUTBOX_PREFIX })]; expect(reportEntries).toHaveLength(1); expect(parsePendingRunReport(reportEntries[0]?.[1])?.report.run.status).toBe('interrupted'); expect(await state.storage.getAlarm()).not.toBeNull();