fix(runtime): preserve cancellation recovery across harnesses

README.md

@@ -117,6 +117,7 @@ Consumers ignore previews after the first final chunk and preserve each thought'
 Final content is acknowledged through the shared event publisher before the Run terminal, keeping large thoughts outside the terminal batch limit.
 
 Each successful Run exports an immutable, verified native bundle under `.state/native-checkpoints/<runId>` and includes its descriptor in `run.completed`.
+OpenAI cancellation before the first successful native checkpoint starts a fresh thread for the next input; cancellation after a successful checkpoint resumes the established thread.
 In mosoo, the host backs up that bundle before atomically committing the checkpoint descriptor and canonical completion receipt.
 The next Run waits for terminal acknowledgement, backend settlement, required cleanup, and input-command acknowledgement.
 A cleanup failure blocks handoff while preserving the committed outcome.

src/runtimes/acp/acp-agent-process.ts

@@ -4,10 +4,10 @@ import { once } from "node:events";
 import { mkdir } from "node:fs/promises";
 import type { Readable, Writable } from "node:stream";
 
+import type { AgentDriverContext } from "../../core/agent-driver-backend";
 import { summarizePath } from "../../observability/driver-debug";
 import type { DriverStartInput } from "../../protocol/start";
 import { raceWithAbort, settlePromiseWithTimeout } from "../../utils/async";
-import type { AgentDriverContext } from "../../core/agent-driver-backend";
 import {
   bindSpawnedProcess,
   createProcessTreeEnvironment,
@@ -50,6 +50,7 @@ export async function startAcpAgentProcess(
   supervisionOptions: AcpAgentProcessSupervisionOptions = {},
 ): Promise<{
   readonly process: AcpAgentProcess;
+  readonly releaseStartupAbort: () => void;
   readonly ready: Promise<void>;
 }> {
   const command = supervisionOptions.command ?? readFallbackCommand();
@@ -89,14 +90,22 @@ export async function startAcpAgentProcess(
     watchdog: null,
   };
   agentProcessSupervision.set(agentProcess, supervision);
+  let startupAbortReleased = false;
   const onAbort = () => signalAcpAgentProcess(agentProcess, "SIGKILL");
+  const releaseStartupAbort = () => {
+    if (startupAbortReleased) {
+      return;
+    }
+    startupAbortReleased = true;
+    signal.removeEventListener("abort", onAbort);
+  };
   signal.addEventListener("abort", onAbort, { once: true });
   const exited = Promise.withResolvers<void>();
   const onExit = () => {
     agentProcess.off("exit", onExit);
     agentProcess.off("close", onExit);
     signalLinuxProcessMarker(supervision.marker, "SIGKILL");
-    signal.removeEventListener("abort", onAbort);
+    releaseStartupAbort();
     exited.resolve();
   };
   agentProcess.once("exit", onExit);
@@ -136,7 +145,7 @@ export async function startAcpAgentProcess(
   );
   void ready.catch(() => {});
 
-  return { process: agentProcess, ready };
+  return { process: agentProcess, ready, releaseStartupAbort };
 }
 
 async function waitForAcpAgentProcessReady(

src/runtimes/acp/acp-client-request-handler.ts

@@ -19,8 +19,8 @@ import type {
   WriteTextFileResponse,
 } from "@agentclientprotocol/sdk";
 
-import type { DriverEventInput } from "../../protocol/events";
 import type { AgentDriverContext } from "../../core/agent-driver-backend";
+import type { DriverEventInput } from "../../protocol/events";
 import {
   AcpTurnStateLimitError,
   type AcpAssistantTranscriptState,
@@ -41,6 +41,7 @@ interface AcpClientRequestHandlerOptions {
   nativeSessionId(): string | null;
   onUpdateFailure(error: Error): void;
   push(context: AgentDriverContext, reason: string, events: DriverEventInput[]): Promise<void>;
+  pushSession: AcpClientRequestHandlerOptions["push"];
   readonly turnEvents: AcpAssistantTranscriptState;
 }
 
@@ -63,6 +64,7 @@ export class AcpClientRequestHandler {
   readonly #pathScope: AcpPathScope;
   readonly #pendingPermissions = new Set<Promise<unknown>>();
   readonly #push: AcpClientRequestHandlerOptions["push"];
+  readonly #pushSession: AcpClientRequestHandlerOptions["push"];
   #permissionIngressClosed = false;
   #stopping = false;
   #transcriptTail: Promise<void> = Promise.resolve();
@@ -77,6 +79,7 @@ export class AcpClientRequestHandler {
     this.#nativeSessionId = options.nativeSessionId;
     this.#onUpdateFailure = options.onUpdateFailure;
     this.#push = options.push;
+    this.#pushSession = options.pushSession;
     this.#turnEvents = options.turnEvents;
     this.#pathScope = new AcpPathScope({
       allowedRoots: options.allowedRoots,
@@ -456,7 +459,8 @@ export class AcpClientRequestHandler {
         return;
       }
 
-      await this.#push(context, "driver.acp.session.update", events);
+      const push = isTurnScopedSessionUpdate(params) ? this.#push : this.#pushSession;
+      await push(context, "driver.acp.session.update", events);
     });
   }
 

src/runtimes/acp/acp-driver-backend.ts

@@ -1,3 +1,6 @@
+import { dirname, join } from "node:path";
+import { Readable, Writable } from "node:stream";
+
 import {
   client as createAcpClient,
   methods as acpMethods,
@@ -10,18 +13,16 @@ import type {
   ClientContext,
   InitializeResponse,
 } from "@agentclientprotocol/sdk";
-import { Readable, Writable } from "node:stream";
-import { dirname, join } from "node:path";
 
+import type { AgentDriverBackend, AgentDriverContext } from "../../core/agent-driver-backend";
+import { AGENT_DRIVER_VERSION } from "../../core/version";
 import { summarizePath, summarizePathCollection } from "../../observability/driver-debug";
 import type { DriverEventInput } from "../../protocol/events";
 import type { RunId } from "../../protocol/id";
 import type { DriverRuntime } from "../../protocol/runtime";
 import type { DriverStartInput } from "../../protocol/start";
 import type { RuntimeCommandInput } from "../../runtime-command";
 import { raceWithAbort, settlePromiseWithTimeout } from "../../utils/async";
-import { AGENT_DRIVER_VERSION } from "../../core/version";
-import type { AgentDriverBackend, AgentDriverContext } from "../../core/agent-driver-backend";
 import { DriverEventPublisher } from "../driver-event-publisher";
 import {
   buildRuntimeBootstrapText,
@@ -33,7 +34,6 @@ import {
 import { startAcpAgentProcess, stopAcpAgentProcess } from "./acp-agent-process";
 import type { AcpAgentProcess } from "./acp-agent-process";
 import { AcpClientRequestHandler } from "./acp-client-request-handler";
-import { limitAcpInput } from "./acp-input-limit";
 import {
   ACP_PROTOCOL_VERSION,
   appendOpenCodeInstruction,
@@ -49,11 +49,11 @@ import {
   supportsSessionLoad,
   supportsSessionResume,
 } from "./acp-configuration";
+import { limitAcpInput } from "./acp-input-limit";
 import { toAuthEvent, toInitializeEvents, toSessionReadyEvents } from "./acp-session-events";
 import { setupAcpSession } from "./acp-session-setup";
 import { withAcpStartupStage } from "./acp-startup";
 import { AcpTurnController } from "./acp-turn-controller";
-
 import {
   exportOpenCodeCheckpoint,
   openCodeDataPath,
@@ -130,6 +130,8 @@ export class AcpDriverBackend implements AgentDriverBackend {
       nativeSessionId: () => this.#nativeSessionId,
       onUpdateFailure: (error) => this.#connection?.close(error),
       push: async (context, reason, events) => this.#push(context, reason, events),
+      pushSession: async (context, reason, events) =>
+        this.#eventPublisher.pushSession(context, reason, events),
       turnEvents: this.#turnController.events,
     });
   }
@@ -462,6 +464,7 @@ export class AcpDriverBackend implements AgentDriverBackend {
       }
       connection.signal.throwIfAborted();
       connectionReady = true;
+      startedProcess.releaseStartupAbort();
 
       return setup;
     } catch (error) {
@@ -613,10 +616,25 @@ export class AcpDriverBackend implements AgentDriverBackend {
       await this.#joinStop();
       return;
     }
-    if (resumeSignal.aborted) {
+    if (!providerPromptAdmitted) {
       return;
     }
-    if (!providerPromptAdmitted) {
+
+    if (resumeSignal.aborted) {
+      const transport = this.#connection;
+      const agentProcess = this.#agentProcess;
+      if (transport !== null) {
+        this.#connection = null;
+        transport.close(new Error("ACP cancelled turn stopping provider process."));
+      }
+      if (agentProcess !== null) {
+        await this.#stopOwnedAgent(
+          context,
+          agentProcess,
+          "turn.cancelled.stop",
+          Date.now() + ACP_RECYCLE_BUDGET_MS,
+        );
+      }
       return;
     }
 

src/runtimes/openai/app-server-driver-backend.ts

@@ -1,3 +1,4 @@
+import type { AgentDriverBackend, AgentDriverContext } from "../../core/agent-driver-backend";
 import { isDriverFullAccess } from "../../core/driver-permission-policy";
 import { pushLosslessEvents } from "../../core/driver-runtime-io";
 import {
@@ -10,24 +11,24 @@ import {
   toDurationMs,
 } from "../../core/driver-runtime-timing";
 import { summarizePath, summarizeRuntimeCommandInput } from "../../observability/driver-debug";
+import type { RuntimeRunCompletedPayload } from "../../protocol/events";
 import { driverIdTimeMs, type RunId } from "../../protocol/id";
+import type { NativeCheckpoint } from "../../protocol/native-checkpoint";
 import type { DriverRuntime } from "../../protocol/runtime";
 import type { DriverStartInput } from "../../protocol/start";
 import type { RuntimeCommandInput } from "../../runtime-command";
 import { raceWithAbort, settlePromiseWithTimeout } from "../../utils/async";
-import type { AgentDriverBackend, AgentDriverContext } from "../../core/agent-driver-backend";
 import { DriverEventPublisher } from "../driver-event-publisher";
 import { createRuntimeSourceEventId } from "../runtime-public-id";
 import {
   buildNativeRuntimeSystemPrompt,
   computeRuntimeBootstrapDigest,
   writeSkillBootstrapArtifacts,
 } from "../skill-bootstrap";
-import { OpenAiAppServerClient } from "./app-server-client";
 import { openAiAgentTasksClosedEvent } from "./app-server-agent-task-events";
+import { OpenAiAppServerClient } from "./app-server-client";
 import { MOSOO_OPENAI_RUNTIME_SANDBOX_MODE } from "./app-server-env";
 import { OpenAiAppServerEventBridge } from "./app-server-event-bridge";
-import { createOpenAiNativeCheckpoint, restoreOpenAiNativeCheckpoint } from "./native-checkpoint";
 import { toOpenAiProtocolError } from "./app-server-event-mapping";
 import type {
   ApprovalPolicy,
@@ -36,6 +37,7 @@ import type {
   ThreadStartResponse,
   TurnStatus,
 } from "./app-server-protocol";
+import { createOpenAiNativeCheckpoint, restoreOpenAiNativeCheckpoint } from "./native-checkpoint";
 
 /**
  * App-server approval policy derived from the driver permission policy.
@@ -122,6 +124,7 @@ export class OpenAiAppServerDriverBackend implements AgentDriverBackend {
   #pendingTurnStartServerRequests: Promise<void> | null = null;
   #pendingTurnStartUpdates: Promise<void> | null = null;
   #pendingCancellationSettlement: Promise<void> | null = null;
+  #nativeCheckpoint: NativeCheckpoint | null;
   #restartThreadId: string | null = null;
   #stopping = false;
   #threadId: string | null = null;
@@ -154,13 +157,24 @@ export class OpenAiAppServerDriverBackend implements AgentDriverBackend {
     push: async (context, reason, events) => this.#eventPublisher.push(context, reason, events),
     pushSession: async (context, reason, events) =>
       this.#eventPublisher.pushSession(context, reason, events),
-    pushTerminal: async (context, reason, closures, terminal, cancellationSignal) =>
-      this.#eventPublisher.pushTerminal(context, reason, closures, terminal, cancellationSignal),
+    pushTerminal: async (context, reason, closures, terminal, cancellationSignal) => {
+      await this.#eventPublisher.pushTerminal(
+        context,
+        reason,
+        closures,
+        terminal,
+        cancellationSignal,
+      );
+      if (terminal.kind === "run.completed") {
+        this.#nativeCheckpoint = (terminal.payload as RuntimeRunCompletedPayload).checkpoint;
+      }
+    },
     requireThreadId: () => this.#requireThreadId(),
   });
 
   constructor(payload: DriverStartInput) {
     this.#payload = payload;
+    this.#nativeCheckpoint = payload.execution.session.nativeCheckpoint;
   }
 
   async start(context: AgentDriverContext, signal: AbortSignal): Promise<void> {
@@ -183,6 +197,7 @@ export class OpenAiAppServerDriverBackend implements AgentDriverBackend {
       restoreOpenAiNativeCheckpoint(this.#payload, signal),
     );
     const nativeResumeThreadId = readResumeThreadId(this.#payload);
+    this.#nativeCheckpoint = this.#payload.execution.session.nativeCheckpoint;
     // Resume emits historical usage before its response; bind the verified native identity first.
     this.#threadId = nativeResumeThreadId;
     const client = this.#createClient(context);
@@ -388,13 +403,43 @@ export class OpenAiAppServerDriverBackend implements AgentDriverBackend {
 
     const client = this.#createClient(context);
     const startupCancellation = new AbortController();
-    const resumeThreadId = this.#restartThreadId ?? this.#threadId;
+    const resumeThreadId =
+      this.#nativeCheckpoint !== null ? (this.#restartThreadId ?? this.#threadId) : null;
+    const checkpoint = this.#nativeCheckpoint;
+    if (resumeThreadId !== null) {
+      if (checkpoint === null) {
+        throw new Error("OpenAI native checkpoint is unavailable for thread resume.");
+      }
+      if (
+        checkpoint.nativeRef.runtimeId !== this.runtime ||
+        checkpoint.nativeRef.kind !== "openai_thread_id" ||
+        checkpoint.nativeRef.value !== resumeThreadId
+      ) {
+        throw new Error("OpenAI native checkpoint does not match the resumed thread.");
+      }
+    }
     this.#threadId = resumeThreadId;
     this.#client = client;
     this.#clientStartupCancellation = startupCancellation;
     this.#clientStopRequested = false;
 
     try {
+      if (checkpoint !== null && resumeThreadId !== null) {
+        await restoreOpenAiNativeCheckpoint(
+          {
+            ...this.#payload,
+            execution: {
+              ...this.#payload.execution,
+              session: {
+                ...this.#payload.execution.session,
+                nativeCheckpoint: checkpoint,
+                nativeResumeRef: checkpoint.nativeRef,
+              },
+            },
+          },
+          startupCancellation.signal,
+        );
+      }
       await client.start(startupCancellation.signal);
       const threadResult = await this.#startThread(
         client,
@@ -1015,7 +1060,7 @@ export class OpenAiAppServerDriverBackend implements AgentDriverBackend {
   ): Promise<void> {
     const isCurrentClient = this.#client === client;
     if (isCurrentClient) {
-      this.#restartThreadId = threadId;
+      this.#restartThreadId = this.#nativeCheckpoint !== null ? threadId : null;
       this.#clientStopRequested = true;
     }
     this.#clientStartupCancellation?.abort(new DriverTurnCancelledError(reason));

src/runtimes/openai/app-server-item-events.ts

@@ -56,6 +56,10 @@ function toOpenAiMessageSnapshotEvents(input: {
   readonly sourcePrefix: string;
   readonly text: string;
 }): DriverEventInput[] {
+  // Native tool turns can close an assistant item without text. Its lifecycle
+  // still closes, but there is no text snapshot to publish.
+  if (input.text.length === 0) return [];
+
   const createAddedEvent = (content: string): DriverEventInput => ({
     delivery: "lossless",
     kind: "message.added",