fix(cloud-agent): settle accepted messages on wrapper complete to prevent webhook hang (#3680)

* fix(cloud-agent-next): settle accepted messages on wrapper complete to prevent webhook hang
* test(cloud-agent): update integration tests for wrapper complete reconciliation

Author: St0rmz1

services/cloud-agent-next/src/session/wrapper-runtime-state.ts

@@ -451,6 +451,15 @@ export async function recordRootSessionIdle(
  * this, the deadline would be cleared forever after the first event, and a
  * wrapper whose kilo-server SSE subscription silently stalls would remain
  * live without failing its accepted messages.
+ *
+ * This intentionally does NOT clear the idle-reconciliation fields
+ * (`lastWrapperIdleAt`/`idleReconcileAfter`/`wrapperIdleDeadlineAt`). Those are
+ * only ever armed by `recordRootSessionIdle` once the root session goes idle,
+ * so any wrapper output observed afterwards is post-completion infrastructure
+ * work (autocommit, condense, log upload) — not a new agent turn. Clearing the
+ * idle deadline here would disarm the reconciler that finalizes the in-flight
+ * message, stranding it non-terminal and hanging the callback. A genuinely new
+ * turn clears idle via `recordWrapperAcceptedMessage` instead.
  */
 export async function recordMeaningfulWrapperOutput(
   storage: DurableObjectStorage,
@@ -465,9 +474,6 @@ export async function recordMeaningfulWrapperOutput(
     lastWrapperMessageAt: now,
     noOutputDeadlineAt,
     nextPingAt: current.pingDeadlineAt === undefined ? nextPingAt : undefined,
-    lastWrapperIdleAt: undefined,
-    idleReconcileAfter: undefined,
-    wrapperIdleDeadlineAt: undefined,
   }));
 }
 

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

@@ -16,6 +16,7 @@ import {
   type WrapperSupervisorStorage,
 } from './wrapper-supervisor.js';
 import { getWrapperLease, getWrapperRuntimeState } from './wrapper-runtime-state.js';
+import type { LatestAssistantMessage } from './types.js';
 
 vi.mock('@cloudflare/sandbox', () => ({
   getSandbox: vi.fn(),
@@ -103,8 +104,15 @@ function createHarness(
   options?: {
     metadata?: SessionMetadata;
     storageHooks?: { beforeList?: (prefix: string) => Promise<void> };
+    getAssistantMessageForUserMessage?: (
+      sessionId: string,
+      kiloSessionId: string,
+      parentMessageId: string
+    ) => LatestAssistantMessage | null;
   }
 ) {
+  const getAssistantMessageForUserMessage =
+    options?.getAssistantMessageForUserMessage ?? (() => null);
   const storage = createMemoryStorage(initialEntries, options?.storageHooks);
   const events: MessageEvent[] = [];
   const callbackJobs: CallbackJob[] = [];
@@ -144,7 +152,7 @@ function createHarness(
     messageSettlementOutbox: settlementOutbox,
     sessionMessageQueue: { requestPendingDrainIfNeeded },
     getMetadata: async () => currentMetadata,
-    getAssistantMessageForUserMessage: () => null,
+    getAssistantMessageForUserMessage,
     hasActiveIngestConnection: async () => false,
     clearInterruptRequest: async () => {},
     stopWrappers,
@@ -631,7 +639,7 @@ describe('WrapperSupervisor', () => {
     await expect(getSessionMessageState(harness.storage, MESSAGE_ID)).resolves.toMatchObject({
       status: 'failed',
       failureReason: 'missing_assistant_reply',
-      error: 'No assistant reply found after idle timeout',
+      error: 'No assistant reply found during reconciliation',
       completionSource: 'idle_reconciliation',
       failureStage: 'post_dispatch_no_activity',
       failureCode: 'missing_assistant_reply',
@@ -679,6 +687,66 @@ describe('WrapperSupervisor', () => {
     });
   });
 
+  it('settles a still-accepted message and enqueues its callback when the wrapper completes without a terminal assistant event', async () => {
+    // Reproduces the webhook hang: the final assistant message.updated never
+    // carried time.completed (so assistant_message_event never settled it), and
+    // post-idle autocommit refreshed liveness while clearing the idle-reconcile
+    // deadline (idle fields absent below). The race-free `complete` event must
+    // still terminalize the message and release the gated callback.
+    const assistantMessageId = 'ase_complete_reconcile';
+    const harness = createHarness([liveRuntimeState(), OWNED_WRAPPER_LEASE], {
+      getAssistantMessageForUserMessage: () =>
+        ({
+          info: { id: assistantMessageId, role: 'assistant' },
+          parts: [],
+        }) as unknown as LatestAssistantMessage,
+    });
+    await putSessionMessageState(harness.storage, {
+      ...acceptedMessage(),
+      callbackRequired: true,
+      callbackTarget: { url: 'https://example.com/complete-reconcile' },
+    });
+
+    await harness.supervisor.onTerminalEvent({
+      wrapperRunId: WRAPPER_RUN_ID,
+      status: 'completed',
+    });
+
+    await expect(getSessionMessageState(harness.storage, MESSAGE_ID)).resolves.toMatchObject({
+      status: 'completed',
+      completionSource: 'idle_reconciliation',
+      assistantMessageId,
+    });
+    expect(harness.callbackJobs).toHaveLength(1);
+    expect(harness.callbackJobs[0].payload).toMatchObject({
+      messageId: MESSAGE_ID,
+      status: 'completed',
+    });
+    expect(harness.requestPendingDrainIfNeeded).toHaveBeenCalledOnce();
+  });
+
+  it('keeps the idle-reconcile deadline armed when post-completion output is observed', async () => {
+    // Layer 2: autocommit/condense events after session.idle must refresh
+    // liveness without disarming the reconciler that finalizes the in-flight
+    // message. Before the fix, observeMeaningfulOutput cleared these fields.
+    const harness = createHarness([
+      liveRuntimeState({
+        lastWrapperIdleAt: 1_000,
+        idleReconcileAfter: 9_000,
+        wrapperIdleDeadlineAt: 50_000,
+      }),
+    ]);
+
+    await harness.supervisor.observeMeaningfulOutput(4, WRAPPER_CONNECTION_ID, 2_000);
+
+    await expect(getWrapperRuntimeState(harness.storage)).resolves.toMatchObject({
+      lastWrapperIdleAt: 1_000,
+      idleReconcileAfter: 9_000,
+      wrapperIdleDeadlineAt: 50_000,
+      lastWrapperMessageAt: 2_000,
+    });
+  });
+
   it('retains successful idle ownership with a bounded physical warm deadline', async () => {
     const harness = createHarness([liveRuntimeState(), OWNED_WRAPPER_LEASE]);
 

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

@@ -645,41 +645,37 @@ export function createWrapperSupervisor(
     return false;
   }
 
-  async function checkIdleReconciliation(now: number): Promise<void> {
-    const metadata = await getMetadata();
-    if (!metadata) return;
-
-    const state = await getWrapperRuntimeState(storage);
-    if (!state.wrapperRunId) return;
-
-    const acceptedMessages = await listNonTerminalAcceptedMessages(storage, state.wrapperRunId);
-    if (acceptedMessages.length === 0) {
-      if (
-        state.wrapperConnectionId &&
-        (state.lastWrapperIdleAt !== undefined || state.idleReconcileAfter !== undefined)
-      ) {
-        await clearWrapperIdleState(storage, state.wrapperGeneration, state.wrapperConnectionId);
-      }
-      return;
-    }
-
-    if (state.idleReconcileAfter !== undefined) {
-      if (now < state.idleReconcileAfter) return;
-    } else {
-      const hasRecentOutput =
-        state.lastWrapperMessageAt !== undefined &&
-        now - state.lastWrapperMessageAt < WRAPPER_NO_OUTPUT_TIMEOUT_MS;
-      if (hasRecentOutput) return;
-    }
-
+  /**
+   * Terminalize the wrapper run's still-accepted messages against the latest
+   * assistant reply, mirroring how a finished turn would have settled them.
+   *
+   * Shared by two triggers:
+   *  - `idle`: the idle-reconciliation grace deadline elapsed.
+   *  - `wrapper_complete`: the wrapper emitted its terminal `complete` event,
+   *    which is the race-free "fully done" signal (it fires only after all
+   *    post-completion work). This is the authoritative backstop for the case
+   *    where the assistant `message.updated` arrived without `time.completed`,
+   *    so neither the assistant-event nor the idle path settled the message.
+   *
+   * `terminalizeSessionMessageOnce` is idempotent, so re-running here after a
+   * partial settlement is safe.
+   */
+  async function reconcileAcceptedMessages(
+    now: number,
+    state: WrapperRuntimeState,
+    metadata: SessionMetadata,
+    acceptedMessages: Awaited<ReturnType<typeof listNonTerminalAcceptedMessages>>,
+    trigger: 'idle' | 'wrapper_complete'
+  ): Promise<void> {
     logger
       .withFields({
         sessionId: metadata.identity.sessionId,
         wrapperRunId: state.wrapperRunId,
         acceptedMessageCount: acceptedMessages.length,
         hasKiloSessionId: metadata.auth.kiloSessionId !== undefined,
+        trigger,
       })
-      .info('Idle reconciliation processing accepted messages');
+      .info('Reconciling accepted messages');
 
     let failedTerminalObserved = !metadata.auth.kiloSessionId;
     if (!failedTerminalObserved && metadata.auth.kiloSessionId) {
@@ -704,7 +700,7 @@ export function createWrapperSupervisor(
         await messageSettlementOutbox.terminalizeSessionMessageOnce(message.messageId, {
           kind: 'failed',
           reason: 'missing_assistant_reply',
-          error: 'No assistant reply found after idle timeout',
+          error: 'No assistant reply found during reconciliation',
           completionSource: 'idle_reconciliation',
           failureStage: 'post_dispatch_no_activity',
           failureCode: 'missing_assistant_reply',
@@ -722,7 +718,7 @@ export function createWrapperSupervisor(
         await messageSettlementOutbox.terminalizeSessionMessageOnce(message.messageId, {
           kind: 'failed',
           reason: 'missing_assistant_reply',
-          error: 'No assistant reply found after idle timeout',
+          error: 'No assistant reply found during reconciliation',
           completionSource: 'idle_reconciliation',
           failureStage: 'post_dispatch_no_activity',
           failureCode: 'missing_assistant_reply',
@@ -768,8 +764,39 @@ export function createWrapperSupervisor(
         sessionId: metadata.identity.sessionId,
         wrapperRunId: state.wrapperRunId,
         acceptedMessageCount: acceptedMessages.length,
+        trigger,
       })
-      .info('Idle reconciliation pass completed');
+      .info('Reconciliation pass completed');
+  }
+
+  async function checkIdleReconciliation(now: number): Promise<void> {
+    const metadata = await getMetadata();
+    if (!metadata) return;
+
+    const state = await getWrapperRuntimeState(storage);
+    if (!state.wrapperRunId) return;
+
+    const acceptedMessages = await listNonTerminalAcceptedMessages(storage, state.wrapperRunId);
+    if (acceptedMessages.length === 0) {
+      if (
+        state.wrapperConnectionId &&
+        (state.lastWrapperIdleAt !== undefined || state.idleReconcileAfter !== undefined)
+      ) {
+        await clearWrapperIdleState(storage, state.wrapperGeneration, state.wrapperConnectionId);
+      }
+      return;
+    }
+
+    if (state.idleReconcileAfter !== undefined) {
+      if (now < state.idleReconcileAfter) return;
+    } else {
+      const hasRecentOutput =
+        state.lastWrapperMessageAt !== undefined &&
+        now - state.lastWrapperMessageAt < WRAPPER_NO_OUTPUT_TIMEOUT_MS;
+      if (hasRecentOutput) return;
+    }
+
+    await reconcileAcceptedMessages(now, state, metadata, acceptedMessages, 'idle');
   }
 
   async function checkKeepWarmCleanup(now: number): Promise<void> {
@@ -968,6 +995,28 @@ export function createWrapperSupervisor(
       if (acceptedMessages.length === 0) {
         await retainPhysicalWrapperWarm(Date.now());
         await clearInterruptRequest();
+      } else {
+        // `complete` is the race-free "fully done" signal: it is emitted only
+        // after all post-completion work. If messages are still accepted here,
+        // the assistant-event and idle-reconcile paths both missed them (e.g.
+        // the final assistant `message.updated` lacked `time.completed`, and
+        // post-idle autocommit refreshed liveness). Settle them now rather than
+        // leaving the callback gated indefinitely.
+        const metadata = await getMetadata();
+        if (metadata) {
+          await reconcileAcceptedMessages(
+            Date.now(),
+            state,
+            metadata,
+            acceptedMessages,
+            'wrapper_complete'
+          );
+        } else {
+          logger
+            .withFields({ sessionId, wrapperRunId, acceptedMessageCount: acceptedMessages.length })
+            .warn('Wrapper complete with accepted messages but no session metadata to reconcile');
+        }
+        await clearInterruptRequest();
       }
     } else {
       await clearWrapperRuntimeIdentity(storage, {

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

@@ -20,6 +20,7 @@ import {
   recordWrapperAcceptedMessage,
 } from '../../../src/session/wrapper-runtime-state.js';
 import {
+  getSessionMessageState,
   listNonTerminalAcceptedMessages,
   putSessionMessageState,
   type SessionMessageState,
@@ -294,7 +295,7 @@ describe('handleWrapperTerminalEvent — new-path identity and message preservat
     );
   });
 
-  it('wrapper complete does not clear wrapper runtime identity when accepted messages remain', async () => {
+  it('wrapper complete reconciles still-accepted messages instead of stranding them', async () => {
     const userId = 'user_wrapper_complete_identity';
     const sessionId = 'agent_wrapper_complete_identity';
     const doId = env.CLOUD_AGENT_SESSION.idFromName(`${userId}:${sessionId}`);
@@ -315,9 +316,10 @@ describe('handleWrapperTerminalEvent — new-path identity and message preservat
       const { state: wrapperState } = await allocateWrapperRuntimeState(instance.ctx.storage);
       const { wrapperRunId, wrapperConnectionId } = wrapperState;
 
+      const messageId = 'msg_018f1e2d3c4bWrpCmpAbCdEfGh';
       // Store an accepted (non-terminal) session message state
       const acceptedMessage: SessionMessageState = {
-        messageId: 'msg_018f1e2d3c4bWrpCmpAbCdEfGh',
+        messageId,
         status: 'accepted',
         prompt: 'hello',
         createdAt: Date.now(),
@@ -326,7 +328,11 @@ describe('handleWrapperTerminalEvent — new-path identity and message preservat
       };
       await putSessionMessageState(instance.ctx.storage, acceptedMessage);
 
-      // Fire wrapper complete — with accepted non-terminal messages present
+      // Fire wrapper complete with an accepted non-terminal message present.
+      // `complete` is the race-free terminal signal, so the message must be
+      // settled here rather than left stranded (which would hang the callback).
+      // This session has no kiloSessionId, so no assistant reply can be found
+      // and the message settles as failed (missing_assistant_reply).
       await instance.handleWrapperTerminalEvent({
         wrapperRunId: wrapperRunId!,
         status: 'completed',
@@ -337,20 +343,21 @@ describe('handleWrapperTerminalEvent — new-path identity and message preservat
         instance.ctx.storage,
         wrapperRunId!
       );
+      const settledMessage = await getSessionMessageState(instance.ctx.storage, messageId);
 
-      return { wrapperRuntimeState, wrapperConnectionId, acceptedMessages };
+      return { wrapperRuntimeState, wrapperConnectionId, acceptedMessages, settledMessage };
     });
 
-    // Identity must NOT be cleared while accepted work remains
-    expect(result.wrapperRuntimeState.wrapperConnectionId).toBe(result.wrapperConnectionId);
-    // Accepted message must still be non-terminal
-    expect(result.acceptedMessages).toHaveLength(1);
-    expect(result.acceptedMessages[0]?.status).toBe('accepted');
+    // The message is settled rather than left accepted.
+    expect(result.acceptedMessages).toHaveLength(0);
+    expect(result.settledMessage).toMatchObject({
+      status: 'failed',
+      failureCode: 'missing_assistant_reply',
+      completionSource: 'idle_reconciliation',
+    });
+    // Identity is released once the run has no remaining accepted work.
+    expect(result.wrapperRuntimeState.wrapperConnectionId).toBeUndefined();
   });
-
-  // NOTE: Per Phase 6 (keep-warm cleanup), wrapper `complete` will eventually NOT clear
-  // identity even when no accepted messages remain — keep-warm alarm cleanup owns that.
-  // The current behavior (clearing when idle) is interim and will be superseded by Phase 6.
 });
 
 describe('new-path liveness without executionId', () => {