fix(cloud-agent-next): reconcile completed kilo turns on wrapper death

services/cloud-agent-next/AGENTS.md

@@ -120,7 +120,7 @@ This pattern blocks API endpoints from running for external contributors who don
 - New pending-message writes use a versioned record containing one nested immutable `SessionMessageIntent`, delivery retry state, and callback snapshot; flat pending rows are decoded only inside `src/session/pending-messages.ts`.
 - `SessionMessageState` owns lifecycle/outbox status, terminal effect accounting, and a named immutable `admissionSnapshot` only for post-pending replay validation and recovery; predecessor records normalize into partial `legacyAdmissionConstraints` and never fabricate missing immutable input. Terminal and accepted/sent effects are repairable from pending/alarm replay and events use deterministic uniqueness.
 - Wrapper handoff is currently at-least-once under ambiguous delivery failures: the wrapper forwards prompt/command submissions directly to Kilo and does not query Kilo to suppress or recover duplicate `messageId` submissions. Duplicate prompt/command processing is an accepted edge-case trade-off until Kilo provides an atomic submit-or-return-existing contract.
-- Once accepted work has no pending residue and its fenced wrapper runtime/socket is gone, current DO/Worker interfaces have no bounded authoritative Kilo terminal query. Disconnect or liveness expiry therefore terminalizes remaining accepted work as wrapper failure without redispatch; adding an authoritative Kilo recovery contract is separate lifecycle capability work.
+- When accepted work has no pending residue and its fenced wrapper runtime/socket is gone, disconnect or liveness expiry first reconciles each accepted message against the DO's stored kilocode events (`getAssistantMessageForUserMessage`): positive terminal evidence (assistant `time.completed` or a terminal assistant error) settles the message as `idle_reconciliation`; anything else terminalizes as wrapper failure without redispatch. There is still no live authoritative Kilo terminal query for redispatch; adding one remains separate lifecycle capability work.
 - Callback delivery retry policy is paired with `wrangler.jsonc`: `CALLBACK_DELIVERY_MAX_ATTEMPTS` includes the initial attempt, and each Cloud Agent Next callback queue consumer must configure `max_retries` for the remaining redeliveries.
 - Queue/drain emits unfenced `MessageDeliveryRequest`; only `AgentRuntime` may allocate/reuse current identity and construct `FencedWrapperDispatchRequest` with complete `WrapperRunFence` for downstream dispatch.
 - Session creation selects an explicit `ProfileResolutionPolicy` at the handler boundary. Implicit repository/default profile resolution is limited to the closed set of approved session origins; omitted, unknown, and non-approved automation origins fail closed unless they supply an explicit profile id.

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

@@ -989,6 +989,147 @@ describe('WrapperSupervisor', () => {
     });
   });
 
+  it('reconciles a completed assistant reply when the wrapper dies after finalizing', async () => {
+    const pingDeadlineAt = 92_000;
+    const noOutputDeadlineAt = 332_000;
+    const assistantMessageId = 'ase_wrapper_death_reconcile';
+    const harness = createHarness(
+      [
+        liveRuntimeState({
+          pingDeadlineAt,
+          noOutputDeadlineAt,
+          finalizingWrapperRunId: WRAPPER_RUN_ID,
+        }),
+        OWNED_WRAPPER_LEASE,
+      ],
+      {
+        getAssistantMessageForUserMessage: () =>
+          ({
+            info: {
+              id: assistantMessageId,
+              role: 'assistant',
+              time: { created: 90_000, completed: 90_500 },
+            },
+            parts: [],
+          }) as unknown as LatestAssistantMessage,
+      }
+    );
+    await putSessionMessageState(harness.storage, {
+      ...acceptedMessage(),
+      callbackRequired: true,
+      callbackTarget: { url: 'https://example.com/wrapper-death-reconcile' },
+    });
+
+    await harness.supervisor.runMaintenance(pingDeadlineAt);
+
+    await expect(getSessionMessageState(harness.storage, MESSAGE_ID)).resolves.toMatchObject({
+      status: 'completed',
+      completionSource: 'idle_reconciliation',
+      assistantMessageId,
+    });
+    expect(harness.events.map(event => event.streamEventType)).toEqual(['cloud.message.completed']);
+    expect(harness.callbackJobs).toHaveLength(1);
+    expect(harness.callbackJobs[0].payload).toMatchObject({
+      messageId: MESSAGE_ID,
+      status: 'completed',
+    });
+  });
+
+  it('still fails accepted work when the stored assistant reply never completed', async () => {
+    const pingDeadlineAt = 92_000;
+    const noOutputDeadlineAt = 332_000;
+    const harness = createHarness(
+      [liveRuntimeState({ pingDeadlineAt, noOutputDeadlineAt }), OWNED_WRAPPER_LEASE],
+      {
+        getAssistantMessageForUserMessage: () =>
+          ({
+            info: { id: 'ase_in_flight', role: 'assistant', time: { created: 90_000 } },
+            parts: [],
+          }) as unknown as LatestAssistantMessage,
+      }
+    );
+    await putSessionMessageState(harness.storage, acceptedMessage());
+
+    await harness.supervisor.runMaintenance(pingDeadlineAt);
+
+    await expect(getSessionMessageState(harness.storage, MESSAGE_ID)).resolves.toMatchObject({
+      status: 'failed',
+      failureReason: 'wrapper_failure',
+      error: 'Wrapper did not respond to liveness ping',
+      completionSource: 'wrapper_failure',
+      failureCode: 'wrapper_ping_timeout',
+    });
+  });
+
+  it('reconciles a terminal assistant error instead of masking it as a wrapper failure', async () => {
+    const pingDeadlineAt = 92_000;
+    const noOutputDeadlineAt = 332_000;
+    const harness = createHarness(
+      [liveRuntimeState({ pingDeadlineAt, noOutputDeadlineAt }), OWNED_WRAPPER_LEASE],
+      {
+        getAssistantMessageForUserMessage: () =>
+          ({
+            info: {
+              id: 'ase_terminal_error',
+              role: 'assistant',
+              error: { data: { message: 'Payment required: insufficient credits' } },
+            },
+            parts: [],
+          }) as unknown as LatestAssistantMessage,
+      }
+    );
+    await putSessionMessageState(harness.storage, acceptedMessage());
+
+    await harness.supervisor.runMaintenance(pingDeadlineAt);
+
+    await expect(getSessionMessageState(harness.storage, MESSAGE_ID)).resolves.toMatchObject({
+      status: 'failed',
+      failureReason: 'assistant_error',
+      completionSource: 'idle_reconciliation',
+      failureCode: 'payment_required',
+    });
+    expect(harness.events.map(event => event.streamEventType)).toEqual(['cloud.message.failed']);
+  });
+
+  it('reconciles a completed assistant reply when disconnect grace expires after wrapper death', async () => {
+    const assistantMessageId = 'ase_disconnect_reconcile';
+    const harness = createHarness(
+      [liveRuntimeState({ finalizingWrapperRunId: WRAPPER_RUN_ID }), OWNED_WRAPPER_LEASE],
+      {
+        getAssistantMessageForUserMessage: () =>
+          ({
+            info: {
+              id: assistantMessageId,
+              role: 'assistant',
+              time: { created: 90_000, completed: 90_500 },
+            },
+            parts: [],
+          }) as unknown as LatestAssistantMessage,
+      }
+    );
+    await putSessionMessageState(harness.storage, acceptedMessage());
+    await harness.supervisor.onDisconnected({
+      disconnected: {
+        wrapperRunId: WRAPPER_RUN_ID,
+        wrapperGeneration: 4,
+        wrapperConnectionId: WRAPPER_CONNECTION_ID,
+      },
+      wsCloseCode: 1006,
+      wsCloseReason: 'socket closed while finalizing',
+    });
+
+    const grace = await harness.storage.get<{ disconnectedAt: number }>('disconnect_grace');
+    if (!grace) throw new Error('Expected disconnect grace to be persisted');
+    await harness.supervisor.runMaintenance(grace.disconnectedAt + 10_001);
+
+    await expect(getSessionMessageState(harness.storage, MESSAGE_ID)).resolves.toMatchObject({
+      status: 'completed',
+      completionSource: 'idle_reconciliation',
+      assistantMessageId,
+    });
+    expect(harness.events.map(event => event.streamEventType)).toEqual(['cloud.message.completed']);
+  });
+
   it('defers liveness failure while disconnect grace is active for the current connection', async () => {
     const pingDeadlineAt = 92_000;
     const noOutputDeadlineAt = 332_000;

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

@@ -236,6 +236,48 @@ function getAssistantErrorMessage(error: unknown): string | undefined {
   return 'Assistant message failed';
 }
 
+function assistantErrorTerminalizeParams(assistantError: string): TerminalizeParams {
+  const assistantFailure = classifyAssistantFailure(assistantError);
+  return {
+    kind: 'failed',
+    reason: 'assistant_error',
+    error: assistantError,
+    completionSource: 'idle_reconciliation',
+    failureStage: 'agent_activity',
+    failureCode: assistantFailure.terminalCode ?? 'assistant_error',
+    assistantFailureReason: assistantFailure.reason,
+    providerOwnership: assistantFailure.providerOwnership,
+    safeFailureMessage: assistantFailure.safeMessage,
+  };
+}
+
+function hasAssistantCompletionMarker(info: LatestAssistantMessage['info']): boolean {
+  const time = info.time;
+  if (typeof time !== 'object' || time === null || !('completed' in time)) return false;
+  return typeof time.completed === 'number';
+}
+
+/**
+ * Wrapper-death reconciliation needs positive terminal evidence: unlike the
+ * sealed-batch path, no wrapper `complete` vouches for turn finality, so bare
+ * assistant-message presence is not enough. Require the same completion
+ * marker the wrapper itself uses to arm finalization (the wrapper's
+ * isAssistantCompletionSignal): a completed timestamp or a terminal error.
+ */
+function projectWrapperDeathReconciliation(
+  assistantMessage: LatestAssistantMessage | null
+): TerminalizeParams | null {
+  if (!assistantMessage) return null;
+  const assistantError = getAssistantErrorMessage(assistantMessage.info.error);
+  if (assistantError !== undefined) return assistantErrorTerminalizeParams(assistantError);
+  if (!hasAssistantCompletionMarker(assistantMessage.info)) return null;
+  return {
+    kind: 'completed',
+    assistantMessageId: assistantMessage.info.id,
+    completionSource: 'idle_reconciliation',
+  };
+}
+
 function getWrapperInterruptionFailureCode(
   interruptionSource: WrapperTerminalEvent['interruptionSource'],
   error: string | undefined
@@ -624,6 +666,49 @@ export function createWrapperSupervisor(
     }
   }
 
+  /**
+   * When the wrapper dies before its terminal report, the DO's event store
+   * still holds every kilocode event that arrived over the FIFO ingest
+   * channel — including, by ingest ordering, the completed assistant state
+   * whenever the wrapper had already started finalizing. Settle from that
+   * positive terminal evidence when present; otherwise fall back to
+   * wrapper-failure handling.
+   */
+  async function terminalizeAcceptedMessagesForDeadWrapper(
+    acceptedMessages: SessionMessageState[],
+    fallbackParams: (message: SessionMessageState) => TerminalizeParams
+  ): Promise<void> {
+    const metadata = await getMetadata();
+    const kiloSessionId = metadata?.auth.kiloSessionId;
+    let reconciledCount = 0;
+    for (const message of acceptedMessages) {
+      const reconciled =
+        metadata && kiloSessionId
+          ? projectWrapperDeathReconciliation(
+              getAssistantMessageForUserMessage(
+                metadata.identity.sessionId,
+                kiloSessionId,
+                message.messageId
+              )
+            )
+          : null;
+      if (reconciled) reconciledCount += 1;
+      await messageSettlementOutbox.terminalizeSessionMessageOnce(
+        message.messageId,
+        reconciled ?? fallbackParams(message)
+      );
+    }
+    if (reconciledCount > 0) {
+      logger
+        .withFields({
+          sessionId: getSessionIdForLogs(),
+          reconciledCount,
+          fallbackCount: acceptedMessages.length - reconciledCount,
+        })
+        .warn('Settled accepted wrapper work from stored assistant events after wrapper death');
+    }
+  }
+
   async function handleUnhealthyWrapper(
     state: WrapperRuntimeState,
     error: string,
@@ -641,17 +726,17 @@ export function createWrapperSupervisor(
     await requestPhysicalWrapperStop('unhealthy-wrapper');
 
     const acceptedMessages = await listNonTerminalAcceptedMessages(storage, state.wrapperRunId);
-    for (const message of acceptedMessages) {
+    await terminalizeAcceptedMessagesForDeadWrapper(acceptedMessages, message => {
       const activityObserved = message.agentActivityObservedAt !== undefined;
-      await messageSettlementOutbox.terminalizeSessionMessageOnce(message.messageId, {
+      return {
         kind: 'failed',
         reason: 'wrapper_failure',
         error,
         completionSource: 'wrapper_failure',
         failureStage: activityObserved ? 'agent_activity' : 'post_dispatch_no_activity',
         failureCode: activityObserved ? 'wrapper_error_after_activity' : failureCode,
-      });
-    }
+      };
+    });
     await messageSettlementOutbox.releaseWrapperTerminalWaitForIdleBatch();
     if (isWrapperRunFinalizing(state) && state.wrapperRunId) {
       await messageSettlementOutbox.finalizeTerminalWrapperRunCallbackIfReady(state.wrapperRunId);
@@ -720,17 +805,17 @@ export function createWrapperSupervisor(
       .warn('Grace period expired - failing supervised wrapper work');
     await requestPhysicalWrapperStop('unhealthy-wrapper');
     await storage.delete(DISCONNECT_GRACE_KEY);
-    for (const message of acceptedMessages) {
+    await terminalizeAcceptedMessagesForDeadWrapper(acceptedMessages, message => {
       const activityObserved = message.agentActivityObservedAt !== undefined;
-      await messageSettlementOutbox.terminalizeSessionMessageOnce(message.messageId, {
+      return {
         kind: 'failed',
         reason: 'wrapper_disconnected',
         error: 'Wrapper disconnected',
         completionSource: 'wrapper_failure',
         failureStage: activityObserved ? 'agent_activity' : 'post_dispatch_no_activity',
         failureCode: activityObserved ? 'wrapper_error_after_activity' : 'wrapper_disconnected',
-      });
-    }
+      };
+    });
     await clearWrapperRuntimeIdentity(
       storage,
       {
@@ -971,21 +1056,10 @@ export function createWrapperSupervisor(
         : null;
       const assistantError = getAssistantErrorMessage(assistantMessage?.info.error);
       if (assistantError !== undefined) {
-        const assistantFailure = classifyAssistantFailure(assistantError);
         projectedSettlements.push({
           message,
           observeCorrelatedActivity: true,
-          params: {
-            kind: 'failed',
-            reason: 'assistant_error',
-            error: assistantError,
-            completionSource: 'idle_reconciliation',
-            failureStage: 'agent_activity',
-            failureCode: assistantFailure.terminalCode ?? 'assistant_error',
-            assistantFailureReason: assistantFailure.reason,
-            providerOwnership: assistantFailure.providerOwnership,
-            safeFailureMessage: assistantFailure.safeMessage,
-          },
+          params: assistantErrorTerminalizeParams(assistantError),
         });
       } else if (assistantMessage) {
         projectedSettlements.push({