fix(cloud-agent-next): recover stalled ingest reconnects and stuck preparing UI

packages/cloud-agent-sdk/src/service-state.test.ts

@@ -1738,6 +1738,44 @@ describe('createServiceState', () => {
       });
     });
 
+    it('terminal delivery failure resolves a stale preparing status', () => {
+      const state = createServiceState(makeConfig());
+
+      state.process({
+        type: 'cloud.status',
+        cloudStatus: { type: 'preparing', message: 'Setting up environment...' },
+      });
+      state.process({
+        type: 'cloud.message.failed',
+        messageId: 'm1',
+        error: 'Environment preparation failed',
+        reason: 'exhausted',
+      });
+
+      expect(state.getCloudStatus()).toEqual({
+        type: 'error',
+        message: 'Environment preparation failed',
+      });
+    });
+
+    it('an interrupt during preparation clears the preparing status', () => {
+      const state = createServiceState(makeConfig());
+
+      state.process({
+        type: 'cloud.status',
+        cloudStatus: { type: 'preparing', message: 'Setting up environment...' },
+      });
+      state.process({
+        type: 'cloud.message.failed',
+        messageId: 'm1',
+        error: 'The message was interrupted',
+        reason: 'interrupted',
+      });
+
+      expect(state.getCloudStatus()).toBeNull();
+      expect(state.getStatus()).toEqual({ type: 'interrupted' });
+    });
+
     it('cloud.message.failed with reason=interrupted settles the session', () => {
       const state = createServiceState(makeConfig());
 

packages/cloud-agent-sdk/src/service-state.ts

@@ -591,6 +591,14 @@ function createServiceState(config: ServiceStateConfig): ServiceState {
       ...(event.attempts !== undefined ? { attempts: event.attempts } : {}),
     };
     pendingMessages.set(event.messageId, deliveryState);
+    // A preparation failure can arrive as a terminal message-delivery event
+    // without a separate preparing event. Do not leave the composer showing
+    // "Setting up environment" forever in that case. An interrupt is the user
+    // cancelling, not a failure, so it clears the stale status instead of
+    // raising an error banner.
+    if (cloudStatus?.type === 'preparing') {
+      cloudStatus = event.reason === 'interrupted' ? null : { type: 'error', message: event.error };
+    }
     if (event.reason === 'interrupted') {
       activity = { type: 'idle' };
       status = { type: 'interrupted' };

services/cloud-agent-next/src/persistence/CloudAgentSession.ts

@@ -3551,7 +3551,7 @@ export class CloudAgentSession extends DurableObject<WorkerEnv> {
         { ...plan, preparation: { attemptId: recorder.attemptId } },
         {
           onProgress: (step, message) => {
-            recorder.onProgress(step, message);
+            if (!recorder.onProgress(step, message)) return;
             this.broadcastVolatileEvent({
               executionId: eventSourceId,
               sessionId,
@@ -3567,6 +3567,7 @@ export class CloudAgentSession extends DurableObject<WorkerEnv> {
             if (!readyResult.success) {
               throw new Error(readyResult.error ?? 'Failed to record session readiness');
             }
+            recorder.finalize({ status: 'completed' });
           },
           onAccepted: delivery => this.recordRuntimeAcceptedMessage(plan, delivery),
         }

services/cloud-agent-next/src/session/preparation-progress.test.ts

@@ -143,6 +143,19 @@ describe('createPreparationProgressRecorder', () => {
     expect(eventQueries.findByEntityPrefix('preparation/attempt/')).toEqual([]);
   });
 
+  it('ignores progress received after the attempt was finalized', () => {
+    const eventQueries = createMemoryEventQueries();
+    const broadcasts: StoredEvent[] = [];
+    const recorder = createRecorder(eventQueries, broadcasts);
+
+    recorder.onProgress('sandbox_provision', 'Provisioning sandbox…');
+    recorder.finalize({ status: 'failed', safeError: 'Environment preparation failed' });
+    broadcasts.length = 0;
+
+    expect(recorder.onProgress('cloning', 'Cloning repository…')).toBe(false);
+    expect(broadcasts).toEqual([]);
+  });
+
   it('finalize settles an attempt the wrapper continued but never terminated', () => {
     const eventQueries = createMemoryEventQueries();
     const broadcasts: StoredEvent[] = [];

services/cloud-agent-next/src/session/preparation-progress.ts

@@ -19,7 +19,7 @@ import {
 export type PreparationProgressRecorder = {
   readonly attemptId: string;
   /** Translate a legacy (step, message) progress callback into v2 events. */
-  onProgress(step: string, message: string): void;
+  onProgress(step: string, message: string): boolean;
   /**
    * Drive the attempt to a terminal state if it is still running. A no-op
    * when no preparation happened or the wrapper already finished the attempt.
@@ -72,7 +72,7 @@ export function createPreparationProgressRecorder(options: {
     if (materializePreparationEvent(eventQueries, stored, data)) broadcast(stored);
   }
 
-  function onProgress(step: string, message: string): void {
+  function onProgress(step: string, message: string): boolean {
     const key = step as PreparingStep;
     if (!readPreparationAttempt(eventQueries, attemptId)) {
       emit('workspace_setup', 'Preparing environment', { action: 'attempt_started' });
@@ -91,10 +91,12 @@ export function createPreparationProgressRecorder(options: {
       activeStep = { id: stepId, key };
     }
     emit(key, message, { action: 'step_progress', stepId, detail: message });
+    return readPreparationAttempt(eventQueries, attemptId)?.status === 'running';
   }
 
   function finalize(outcome: PreparationOutcome): void {
     activeStep = undefined;
+    if (!readPreparationAttempt(eventQueries, attemptId)) return;
     for (const event of finalizePreparationAttempt(eventQueries, attemptId, {
       ...outcome,
       timestamp: now(),

services/cloud-agent-next/test/integration/session/execute-directly-failure.test.ts

@@ -158,6 +158,134 @@ describe('executeDirectly failure handling', () => {
     expect(result.wrapperRuntimeState.noOutputDeadlineAt).toBeGreaterThan(result.staleDeadline);
   });
 
+  /**
+   * A held delivery (the previous wrapper batch is still finalizing) never
+   * reaches preparation: the message stays queued and is retried moments later.
+   * Settling a preparation attempt from that outcome would flash a spurious
+   * "Environment preparation failed" card between the two tries.
+   */
+  async function drainWithRuntimeResult(
+    keySuffix: string,
+    result: { success: false; code: string; error: string },
+    options: { emitProgress?: boolean; reportWorkspaceReady?: boolean } = {}
+  ) {
+    const userId = `user_exec_direct_${keySuffix}`;
+    const sessionId = `agent_exec_direct_${keySuffix}`;
+    const stub = env.CLOUD_AGENT_SESSION.get(
+      env.CLOUD_AGENT_SESSION.idFromName(`${userId}:${sessionId}`)
+    );
+
+    return runInDurableObject(stub, async (instance, state) => {
+      (instance as any).agentRuntime = {
+        send: async (_plan: unknown, hooks: any) => {
+          if (options.emitProgress) {
+            hooks?.onProgress?.('sandbox_provision', 'Provisioning sandbox…');
+          }
+          if (options.reportWorkspaceReady) {
+            await hooks?.onWorkspaceReady?.({
+              workspacePath: `/workspace/${userId}/sessions/${sessionId}`,
+              sandboxId: 'usr-123456789abc',
+              sessionHome: `/home/${sessionId}`,
+              branchName: `session/${sessionId}`,
+              kiloSessionId: 'bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb',
+            });
+          }
+          return result;
+        },
+        requestSnapshot: async () => {},
+        interruptWrapper: async () => ({ commandSent: false }),
+        sendPing: () => {},
+        keepSandboxAlive: async () => {},
+      };
+
+      await registerReadySession(instance, {
+        sessionId,
+        userId,
+        orgId: `org_exec_direct_${keySuffix}`,
+        kiloSessionId: 'bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb',
+        prompt: 'initial prompt',
+        mode: 'code',
+        model: 'test-model',
+        kilocodeToken: `token-${keySuffix}`,
+      });
+
+      await instance.admitSubmittedMessage(
+        queueUserMessageInput({
+          userId,
+          prompt: 'do some work',
+          messageId: 'msg_018f1e2d3c4bHeldMsgAbCdEfG',
+        })
+      );
+      await instance.alarm();
+
+      const db = drizzle(state.storage, { logger: false });
+      const eventQueries = createEventQueries(db, state.storage.sql);
+      const attemptRows = eventQueries.findByEntityPrefix('preparation/attempt/');
+      return {
+        pending: await listPendingSessionMessages(instance.ctx.storage),
+        preparationEvents: attemptRows,
+        attemptStatuses: attemptRows
+          .map(row => parsePreparationAttemptStatus(row.payload))
+          .filter((status): status is string => status !== null),
+      };
+    });
+  }
+
+  it('a held delivery leaves no preparation attempt behind', async () => {
+    const result = await drainWithRuntimeResult('held', {
+      success: false,
+      code: 'WRAPPER_FINALIZING',
+      error: 'Wrapper batch is finalizing',
+    });
+
+    expect(result.preparationEvents).toEqual([]);
+    // Held, not failed: the message is still queued for the next drain.
+    expect(result.pending.map(message => message.messageId)).toEqual([
+      'msg_018f1e2d3c4bHeldMsgAbCdEfG',
+    ]);
+  });
+
+  it('a hold raised after preparation started terminalizes that attempt', async () => {
+    const result = await drainWithRuntimeResult(
+      'held-after-progress',
+      { success: false, code: 'WRAPPER_FINALIZING', error: 'Wrapper batch is finalizing' },
+      { emitProgress: true }
+    );
+
+    // The attempt exists (progress was observed) and must not be left running,
+    // or clients stay in the preparing state forever.
+    expect(result.attemptStatuses).toEqual(['failed']);
+  });
+
+  function parsePreparationAttemptStatus(payload: unknown): string | null {
+    const parsed = JSON.parse(String(payload)) as { attempt?: { status?: string } };
+    return parsed.attempt?.status ?? null;
+  }
+
+  it('a terminal delivery failure before preparation leaves no attempt behind', async () => {
+    const result = await drainWithRuntimeResult('terminal', {
+      success: false,
+      code: 'SANDBOX_CAPABILITY_UNAVAILABLE',
+      error: 'Sandbox capability unavailable',
+    });
+
+    expect(result.preparationEvents).toEqual([]);
+  });
+
+  it('a delivery failure after workspace readiness leaves the attempt completed', async () => {
+    const result = await drainWithRuntimeResult(
+      'failure-after-ready',
+      {
+        success: false,
+        code: 'WRAPPER_START_FAILED',
+        error: 'Prompt dispatch failed',
+      },
+      { emitProgress: true, reportWorkspaceReady: true }
+    );
+
+    expect(result.attemptStatuses).toEqual(['completed']);
+  });
+
   it('queued flush pre-start failure retries cleanly with the original execution and message ids', async () => {
     const userId = 'user_exec_direct_fail';
     const sessionId = 'agent_exec_direct_fail';