CloudAgentNext - broadcast cloud.status ready after terminalization and broaden wrapper terminal error detection

packages/db/src/schema.ts

@@ -5533,6 +5533,7 @@ export type CloudAgentSessionRunFailureCode =
   | 'wrapper_error_after_activity'
   | 'missing_assistant_reply'
   | 'payment_required'
+  | 'provider_error'
   | 'user_interrupt'
   | 'container_shutdown'
   | 'system_interrupt'

packages/worker-utils/src/cloud-agent-failure.ts

@@ -28,6 +28,7 @@ export const CLOUD_AGENT_FAILURE_CODES = [
   'wrapper_error_after_activity',
   'missing_assistant_reply',
   'payment_required',
+  'provider_error',
   'user_interrupt',
   'container_shutdown',
   'system_interrupt',

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

@@ -715,6 +715,15 @@ export class CloudAgentSession extends DurableObject<WorkerEnv> {
             : recordSharedSandboxFailover(this.env.SHARED_SANDBOX_OVERRIDES, routeKey),
         requestAlarmAtOrBefore: deadline => this.scheduleAlarmAtOrBefore(deadline),
         getSessionIdForLogs: () => this.sessionId,
+        onSessionTerminalized: async () => {
+          this.broadcastVolatileEvent({
+            executionId: '' as EventSourceId,
+            sessionId: this.sessionId ?? '',
+            streamEventType: 'cloud.status',
+            payload: JSON.stringify({ cloudStatus: { type: 'ready' } }),
+            timestamp: Date.now(),
+          });
+        },
       });
     }
 
@@ -3259,8 +3268,19 @@ export class CloudAgentSession extends DurableObject<WorkerEnv> {
   }
 
   async handleWrapperTerminalEvent(params: WrapperTerminalEvent): Promise<void> {
-    await this.resolveSessionId();
-    await this.getWrapperSupervisor().onTerminalEvent(params);
+    const sessionId = await this.resolveSessionId();
+    const accepted = await this.getWrapperSupervisor().onTerminalEvent(params);
+
+    if (accepted && (params.status === 'failed' || params.status === 'interrupted')) {
+      this.broadcastVolatileEvent({
+        executionId: '' as EventSourceId,
+        sessionId: sessionId ?? '',
+        streamEventType: 'cloud.status',
+        payload: JSON.stringify({ cloudStatus: { type: 'ready' } }),
+        timestamp: Date.now(),
+      });
+    }
+
     const metadata = await this.getMetadata();
     if (!metadata) return;
     if (!isCodeReviewEphemeralSandboxId(metadata.workspace?.sandboxId)) return;

services/cloud-agent-next/src/session/safe-failure-projection.ts

@@ -39,6 +39,7 @@ const GENERIC_FAILURE_MESSAGES = {
   wrapper_error_after_activity: 'Agent wrapper failed while processing the message',
   missing_assistant_reply: 'No assistant reply was produced',
   payment_required: 'Assistant request failed: insufficient credits',
+  provider_error: 'Assistant provider returned an invalid response',
   user_interrupt: 'The message was interrupted by the user',
   container_shutdown: 'The agent container shut down',
   system_interrupt: 'The message was interrupted',

services/cloud-agent-next/src/session/terminal-error-projector.ts

@@ -9,6 +9,7 @@ const NON_RETRYABLE_FAILURE_CODES = new Set<SessionMessageFailureCode>([
   'session_metadata_missing',
   'model_missing',
   'payment_required',
+  'provider_error',
   'user_interrupt',
 ]);
 

services/cloud-agent-next/src/session/wrapper-supervisor.test.ts

@@ -122,6 +122,7 @@ function createHarness(
       wrapperRunId: string
     ) => Promise<void>;
     recordSharedSandboxFailover?: (routeKey: SandboxId) => Promise<void>;
+    onSessionTerminalized?: () => Promise<void>;
   }
 ) {
   const getAssistantMessageForUserMessage =
@@ -158,6 +159,7 @@ function createHarness(
   const recordSharedSandboxFailover = vi.fn(
     options?.recordSharedSandboxFailover ?? (async () => {})
   );
+  const onSessionTerminalized = vi.fn(options?.onSessionTerminalized ?? (async () => {}));
   const supervisor = createWrapperSupervisor({
     storage,
     agentRuntime: {
@@ -179,6 +181,7 @@ function createHarness(
       requestedAlarms.push(deadline);
     },
     getSessionIdForLogs: () => currentMetadata.identity.sessionId,
+    onSessionTerminalized,
   });
 
   return {
@@ -191,6 +194,7 @@ function createHarness(
     requestedAlarms,
     requestPendingDrainIfNeeded,
     recordSharedSandboxFailover,
+    onSessionTerminalized,
     settlementOutbox,
     supervisor,
   };
@@ -1099,6 +1103,153 @@ describe('WrapperSupervisor', () => {
     });
   });
 
+  it('calls onSessionTerminalized after disconnect grace terminalizes accepted work', async () => {
+    const harness = createHarness([
+      liveRuntimeState({ pingDeadlineAt: 92_000, noOutputDeadlineAt: 332_000 }),
+      OWNED_WRAPPER_LEASE,
+      disconnectGraceForCurrentConnection(90_000),
+    ]);
+    await putSessionMessageState(harness.storage, acceptedMessage());
+
+    await harness.supervisor.runMaintenance(100_001);
+
+    await expect(getSessionMessageState(harness.storage, MESSAGE_ID)).resolves.toMatchObject({
+      status: 'failed',
+      failureCode: 'wrapper_disconnected',
+    });
+    expect(harness.onSessionTerminalized).toHaveBeenCalledTimes(1);
+  });
+
+  it('calls onSessionTerminalized after liveness expiry terminalizes accepted work', async () => {
+    const acceptedAt = 2_000;
+    const noOutputDeadlineAt = acceptedAt + WRAPPER_NO_OUTPUT_TIMEOUT_MS;
+    const harness = createHarness([
+      liveRuntimeState({ noOutputDeadlineAt, nextPingAt: noOutputDeadlineAt + 1 }),
+      OWNED_WRAPPER_LEASE,
+    ]);
+    await putSessionMessageState(harness.storage, acceptedMessage());
+
+    await harness.supervisor.runMaintenance(noOutputDeadlineAt);
+
+    await expect(getSessionMessageState(harness.storage, MESSAGE_ID)).resolves.toMatchObject({
+      status: 'failed',
+      failureCode: 'wrapper_no_output',
+    });
+    expect(harness.onSessionTerminalized).toHaveBeenCalledTimes(1);
+  });
+
+  it('preserves supervision when terminalization persistence fails', async () => {
+    const acceptedAt = 2_000;
+    const noOutputDeadlineAt = acceptedAt + WRAPPER_NO_OUTPUT_TIMEOUT_MS;
+    const harness = createHarness([
+      liveRuntimeState({ noOutputDeadlineAt, nextPingAt: noOutputDeadlineAt + 1 }),
+      OWNED_WRAPPER_LEASE,
+    ]);
+    await putSessionMessageState(harness.storage, acceptedMessage());
+
+    const originalPut = harness.storage.put.bind(harness.storage);
+    harness.storage.put = (async (key: string, value: unknown) => {
+      if (key.startsWith('session_message')) {
+        throw new Error('terminalize storage failure');
+      }
+      await originalPut(key, value);
+    }) as typeof harness.storage.put;
+
+    await expect(harness.supervisor.runMaintenance(noOutputDeadlineAt)).rejects.toThrow(
+      'terminalize storage failure'
+    );
+
+    // Readiness is not published because the durable transition failed
+    expect(harness.onSessionTerminalized).not.toHaveBeenCalled();
+    // Supervision is preserved — the wrapper runtime identity is not cleared
+    const runtimeState = await getWrapperRuntimeState(harness.storage);
+    expect(runtimeState.wrapperConnectionId).toBeDefined();
+    // The message remains accepted so maintenance can retry terminalization
+    await expect(getSessionMessageState(harness.storage, MESSAGE_ID)).resolves.toMatchObject({
+      status: 'accepted',
+    });
+  });
+
+  it('publishes readiness when terminal effects fail after the durable transition', async () => {
+    const acceptedAt = 2_000;
+    const noOutputDeadlineAt = acceptedAt + WRAPPER_NO_OUTPUT_TIMEOUT_MS;
+    const harness = createHarness([
+      liveRuntimeState({ noOutputDeadlineAt, nextPingAt: noOutputDeadlineAt + 1 }),
+      OWNED_WRAPPER_LEASE,
+    ]);
+    await putSessionMessageState(harness.storage, acceptedMessage());
+
+    // Let the durable terminal transition succeed (first session_message:
+    // state write), then fail on the subsequent terminal-effects write so the
+    // effect path throws while the message is already terminal.
+    let sessionMessageStatePutCount = 0;
+    const originalPut = harness.storage.put.bind(harness.storage);
+    harness.storage.put = (async (key: string, value: unknown) => {
+      if (key.startsWith('session_message:')) {
+        sessionMessageStatePutCount++;
+        if (sessionMessageStatePutCount > 1) {
+          throw new Error('terminal effects storage failure');
+        }
+      }
+      await originalPut(key, value);
+    }) as typeof harness.storage.put;
+
+    await harness.supervisor.runMaintenance(noOutputDeadlineAt);
+
+    // Readiness is published because the durable transition succeeded
+    expect(harness.onSessionTerminalized).toHaveBeenCalledTimes(1);
+    // The message is terminal despite the effect failure
+    await expect(getSessionMessageState(harness.storage, MESSAGE_ID)).resolves.toMatchObject({
+      status: 'failed',
+    });
+  });
+
+  it('does not call onSessionTerminalized when disconnect grace expires for a stale wrapper run', async () => {
+    const harness = createHarness([
+      liveRuntimeState({
+        wrapperRunId: 'wr_newer_current',
+        wrapperGeneration: 5,
+        wrapperConnectionId: 'conn_newer',
+      }),
+      OWNED_WRAPPER_LEASE,
+      disconnectGraceForCurrentConnection(90_000, {
+        wrapperRunId: 'wr_stale_run',
+        wrapperGeneration: 4,
+        wrapperConnectionId: WRAPPER_CONNECTION_ID,
+      }),
+    ]);
+    await putSessionMessageState(harness.storage, {
+      ...acceptedMessage(),
+      wrapperRunId: 'wr_newer_current',
+    });
+
+    await harness.supervisor.runMaintenance(100_001);
+
+    await expect(harness.storage.get('disconnect_grace')).resolves.toBeUndefined();
+    expect(harness.onSessionTerminalized).not.toHaveBeenCalled();
+  });
+
+  it('calls onSessionTerminalized when liveness expiry abandons a finalizing run with no accepted messages', async () => {
+    const harness = createHarness([
+      liveRuntimeState({
+        finalizingWrapperRunId: WRAPPER_RUN_ID,
+        noOutputDeadlineAt: 9_000,
+        nextPingAt: 30_000,
+      }),
+      OWNED_WRAPPER_LEASE,
+    ]);
+    await putSessionMessageState(harness.storage, {
+      ...acceptedMessage(),
+      status: 'completed',
+      terminalAt: 3_000,
+      completionSource: 'assistant_message_event',
+    });
+
+    await harness.supervisor.runMaintenance(10_000);
+
+    expect(harness.onSessionTerminalized).toHaveBeenCalledTimes(1);
+  });
+
   it('includes the disconnect grace expiry deadline even when the ping deadline is earlier', async () => {
     const harness = createHarness([
       liveRuntimeState({ pingDeadlineAt: 92_000, noOutputDeadlineAt: 332_000 }),

services/cloud-agent-next/src/session/wrapper-supervisor.ts

@@ -13,6 +13,8 @@ import { classifyAssistantFailureMessage } from './safe-failure-projection.js';
 import { countPendingSessionMessages, type SessionQueueStorage } from './pending-messages.js';
 import type { SessionMessageQueue } from './session-message-queue.js';
 import {
+  getSessionMessageState,
+  isTerminalMessageState,
   listMessagesForWrapperRun,
   listNonTerminalAcceptedMessages,
   type SessionMessageState,
@@ -135,7 +137,7 @@ export type WrapperSupervisor = {
   ): Promise<void>;
   observeFinalizing(wrapperRunId: string): Promise<void>;
   onDisconnected(input: WrapperDisconnectedInput): Promise<void>;
-  onTerminalEvent(params: WrapperTerminalEvent): Promise<void>;
+  onTerminalEvent(params: WrapperTerminalEvent): Promise<boolean>;
   requestPhysicalWrapperStop(reason: WrapperStopReason, target?: WrapperStopTarget): Promise<void>;
   clearDisconnectGrace(): Promise<void>;
   runMaintenance(now: number): Promise<void>;
@@ -178,6 +180,13 @@ export type WrapperSupervisorDependencies = {
   recordSharedSandboxFailover: (routeKey: SandboxId) => Promise<void>;
   requestAlarmAtOrBefore?: (deadline: number) => Promise<void>;
   getSessionIdForLogs: () => string | undefined;
+  /**
+   * Invoked after a disconnect-grace or liveness path terminalizes accepted
+   * work, so the DO can broadcast a `cloud.status: ready` event and re-enable
+   * client input. Not invoked for the normal terminal-event path, which
+   * broadcasts readiness from {@link handleWrapperTerminalEvent} directly.
+   */
+  onSessionTerminalized?: () => Promise<void>;
 };
 
 function matchesDisconnectGraceFence(
@@ -362,6 +371,7 @@ export function createWrapperSupervisor(
     recordSharedSandboxFailover,
     requestAlarmAtOrBefore,
     getSessionIdForLogs,
+    onSessionTerminalized,
   } = dependencies;
 
   async function readDisconnectGrace(): Promise<DisconnectGraceState | undefined> {
@@ -633,15 +643,46 @@ export function createWrapperSupervisor(
     const acceptedMessages = await listNonTerminalAcceptedMessages(storage, state.wrapperRunId);
     for (const message of acceptedMessages) {
       const activityObserved = message.agentActivityObservedAt !== undefined;
-      await messageSettlementOutbox.terminalizeSessionMessageOnce(message.messageId, {
-        kind: 'failed',
-        reason: 'wrapper_failure',
-        error,
-        completionSource: 'wrapper_failure',
-        failureStage: activityObserved ? 'agent_activity' : 'post_dispatch_no_activity',
-        failureCode: activityObserved ? 'wrapper_error_after_activity' : failureCode,
-      });
+      try {
+        await messageSettlementOutbox.terminalizeSessionMessageOnce(message.messageId, {
+          kind: 'failed',
+          reason: 'wrapper_failure',
+          error,
+          completionSource: 'wrapper_failure',
+          failureStage: activityObserved ? 'agent_activity' : 'post_dispatch_no_activity',
+          failureCode: activityObserved ? 'wrapper_error_after_activity' : failureCode,
+        });
+      } catch (terminalizeError) {
+        // terminalizeSessionMessageOnly only becomes fallible after the
+        // durable transition commits (event/callback effects). A throw while
+        // the message is still non-terminal means the durable write itself
+        // failed; re-throw so the wrapper runtime identity is preserved and
+        // maintenance can retry instead of orphaning still-accepted work.
+        let durableTransitionSucceeded = false;
+        try {
+          const persistedState = await getSessionMessageState(storage, message.messageId);
+          durableTransitionSucceeded =
+            persistedState !== undefined && isTerminalMessageState(persistedState);
+        } catch {
+          // Re-read failed — assume the durable transition did not succeed.
+        }
+        if (!durableTransitionSucceeded) {
+          throw terminalizeError;
+        }
+        logger
+          .withFields({
+            sessionId: getSessionIdForLogs(),
+            wrapperRunId: state.wrapperRunId,
+            messageId: message.messageId,
+            error:
+              terminalizeError instanceof Error
+                ? terminalizeError.message
+                : String(terminalizeError),
+          })
+          .warn('Failed to apply terminal effects during unhealthy wrapper handling');
+      }
     }
+    await onSessionTerminalized?.();
     await messageSettlementOutbox.releaseWrapperTerminalWaitForIdleBatch();
     if (isWrapperRunFinalizing(state) && state.wrapperRunId) {
       await messageSettlementOutbox.finalizeTerminalWrapperRunCallbackIfReady(state.wrapperRunId);
@@ -721,6 +762,7 @@ export function createWrapperSupervisor(
         failureCode: activityObserved ? 'wrapper_error_after_activity' : 'wrapper_disconnected',
       });
     }
+    await onSessionTerminalized?.();
     await clearWrapperRuntimeIdentity(
       storage,
       {
@@ -1290,7 +1332,7 @@ export function createWrapperSupervisor(
     );
   }
 
-  async function onTerminalEvent(params: WrapperTerminalEvent): Promise<void> {
+  async function onTerminalEvent(params: WrapperTerminalEvent): Promise<boolean> {
     const {
       wrapperRunId,
       status,
@@ -1313,7 +1355,7 @@ export function createWrapperSupervisor(
       logger
         .withFields({ sessionId, wrapperRunId, status })
         .warn('Ignoring non-current wrapper terminal event');
-      return;
+      return false;
     }
 
     logger
@@ -1461,6 +1503,7 @@ export function createWrapperSupervisor(
     ) {
       await sessionMessageQueue.requestPendingDrainIfNeeded();
     }
+    return true;
   }
 
   async function runMaintenance(now: number): Promise<void> {

services/cloud-agent-next/src/shared/protocol.ts

@@ -53,7 +53,11 @@ export type IngestEvent = {
   data: unknown;
 };
 
-export const WrapperTerminalFailureCodes = ['payment_required', 'model_missing'] as const;
+export const WrapperTerminalFailureCodes = [
+  'payment_required',
+  'model_missing',
+  'provider_error',
+] as const;
 export type WrapperTerminalFailureCode = (typeof WrapperTerminalFailureCodes)[number];
 
 /**