feat(shells): give every background shell an owner, a deadline, and a fenced exit

Cursor background shells now belong to the session that spawned them:
the transport threads its stable id through both dispatch paths, stdin
and termination check ownership, and a cross-session write is refused
without naming the owner. Admission runs through the shared gate with
the lease held in the registry entry until the child's close is actually
confirmed — a failed kill quarantines the entry instead of forgetting
the process. Idle and absolute deadlines bound every shell's life, the
transport's async close awaits its session's cleanup, and shutdown
installs a synchronous spawn fence before draining so a queued frame
can never start a shell the drain will miss.

Author: lidge-jun

src/adapters/cursor/live-transport.ts

@@ -74,6 +74,10 @@ import {
   requestedCursorToolUseCount,
 } from "./tool-definitions";
 import type { CursorNativeToolDeps } from "./native-exec-tools";
+import {
+  terminateBackgroundShellsForSession,
+  type BackgroundShellTerminationReport,
+} from "./native-exec-shell";
 import type { CursorClientMessage, CursorRunRequest, CursorServerMessage } from "./types";
 import type { CursorTransport, CursorTransportFactoryInput } from "./transport";
 
@@ -418,6 +422,7 @@ class LiveCursorTransport implements CursorTransport {
   private mcpPrepared?: Promise<void>;
   private releaseMcpObservation?: () => void;
   private blobRequestScope?: CursorBlobRequestScopeToken;
+  private shellCleanup?: Promise<BackgroundShellTerminationReport>;
   // Per-turn diagnostic counters/timestamps when provider debug is on (`ocx debug provider on`). Stamped in open(), cleared on
   // close; safe to read after a stream failure because open() owns the only writer before run().
   private turnStartedAt = 0;
@@ -437,7 +442,11 @@ class LiveCursorTransport implements CursorTransport {
     this.activeClientToolFinalizeGraceMs = this.clientToolFinalizeGraceMs;
     // Desktop (computer-use / record-screen) executors are available even with no MCP servers.
     this.desktopDeps = desktopDepsFromConfig(input.provider.desktopExecutor);
-    this.execContext = { ...this.desktopDeps, unsafeAllowNativeLocalExec: effectiveCursorNativeExecAllow(input.provider, input.requestDeclaresFullAccess === true) };
+    this.execContext = {
+      ...this.desktopDeps,
+      sessionId: this.sessionId,
+      unsafeAllowNativeLocalExec: effectiveCursorNativeExecAllow(input.provider, input.requestDeclaresFullAccess === true),
+    };
     const servers = resolveMcpServers(input.provider);
     if (servers.length > 0) {
       this.mcpManager = new CursorMcpManager(servers, {
@@ -470,6 +479,7 @@ class LiveCursorTransport implements CursorTransport {
             ...this.desktopDeps,
             ...mcpDepsFromManager(this.mcpManager!),
             mcpToolDefs,
+            sessionId: this.sessionId,
             unsafeAllowNativeLocalExec: effectiveCursorNativeExecAllow(this.input.provider, this.input.requestDeclaresFullAccess === true),
           };
         } catch (err) {
@@ -648,7 +658,11 @@ class LiveCursorTransport implements CursorTransport {
     }
   }
 
-  close(): void {
+  private startShellCleanup(): Promise<BackgroundShellTerminationReport> {
+    return this.shellCleanup ??= terminateBackgroundShellsForSession(this.sessionId);
+  }
+
+  async close(): Promise<void> {
     if (this.heartbeat) clearInterval(this.heartbeat);
     this.clearPendingFinalize();
     this.clearFirstFrameTimer();
@@ -658,6 +672,7 @@ class LiveCursorTransport implements CursorTransport {
     this.releaseMcpObservation?.();
     this.releaseMcpObservation = undefined;
     void this.mcpManager?.dispose();
+    await this.startShellCleanup();
   }
 
   private cancelCursorRun(): void {
@@ -675,6 +690,7 @@ class LiveCursorTransport implements CursorTransport {
     this.releaseMcpObservation?.();
     this.releaseMcpObservation = undefined;
     void this.mcpManager?.dispose();
+    void this.startShellCleanup().catch(() => { /* close() observes the same cleanup promise */ });
   }
 
   private releaseBlobRequestScope(): void {
@@ -850,16 +866,17 @@ class LiveCursorTransport implements CursorTransport {
         return;
       }
       const previousPayloadBytes = connectBufferedPayloadBytes(pending);
-      // Transfer the retained-byte lease from the aggregate pending buffer to decoded frame and
-      // residual owners. Charging both copies transiently would cut the 32 MiB frame contract in half.
-      this.releaseTransportBytes(previousPayloadBytes);
+      // The decoder materializes payload and residual copies. Keep the aggregate pending owner
+      // charged until every replacement has been admitted and committed; a failed admission must
+      // leave that predecessor lease intact for deterministic cleanup.
       const decoded = decodeAvailableConnectFrames(
         pending,
         CURSOR_MAX_EFFECTIVE_CONNECT_PAYLOAD_BYTES,
         availableSlots,
         reservePayloadCopy,
       );
       pending = decoded.remainder;
+      this.releaseTransportBytes(previousPayloadBytes);
       for (const frame of decoded.frames) {
         this.pendingTransportFrames += 1;
         this.updateTransportFlowControl();
@@ -882,18 +899,26 @@ class LiveCursorTransport implements CursorTransport {
         debugProviderDiagnostic("cursor", "first-frame", { latencyMs: this.firstFrameAt - this.turnStartedAt });
       }
       const bytes = typeof chunk === "string" ? new TextEncoder().encode(chunk) : new Uint8Array(chunk.buffer, chunk.byteOffset, chunk.byteLength);
+      let incomingPayloadBytes = 0;
+      let incomingCharged = false;
+      let replacement: ReturnType<typeof reservePayloadCopy> | undefined;
       try {
         const previousPendingPayloadBytes = connectBufferedPayloadBytes(pending);
+        const nextPendingPayloadBytes = connectBufferedPayloadBytesAcross(pending, bytes);
+        incomingPayloadBytes = Math.max(0, nextPendingPayloadBytes - previousPendingPayloadBytes);
+        this.reserveTransportBytes(incomingPayloadBytes);
+        incomingCharged = true;
+        replacement = reservePayloadCopy(nextPendingPayloadBytes);
         const nextPending = concatBytes(pending, bytes);
-        const nextPendingPayloadBytes = connectBufferedPayloadBytes(nextPending);
-        // The incoming HTTP/2 chunk is externally owned. Replace the prior pending lease with the
-        // concatenated buffer lease instead of charging old + new simultaneously.
-        this.releaseTransportBytes(previousPendingPayloadBytes);
-        const replacement = reservePayloadCopy(nextPendingPayloadBytes);
         pending = nextPending;
         replacement.commitRetained();
+        replacement = undefined;
+        this.releaseTransportBytes(previousPendingPayloadBytes + incomingPayloadBytes);
+        incomingCharged = false;
         drainPendingFrames();
       } catch (err) {
+        replacement?.release();
+        if (incomingCharged) this.releaseTransportBytes(incomingPayloadBytes);
         failAndClear(err instanceof Error ? err : new Error(String(err)));
       }
     });
@@ -1135,6 +1160,27 @@ function connectBufferedPayloadBytes(input: Uint8Array): number {
   return payloadBytes;
 }
 
+/** Payload-byte count for a virtual concatenation, without allocating the concatenated buffer. */
+function connectBufferedPayloadBytesAcross(first: Uint8Array, second: Uint8Array): number {
+  const totalBytes = first.byteLength + second.byteLength;
+  const byteAt = (index: number): number => index < first.byteLength
+    ? first[index]!
+    : second[index - first.byteLength]!;
+  let offset = 0;
+  let payloadBytes = 0;
+  while (totalBytes - offset >= 5) {
+    const length = (byteAt(offset + 1) * 0x1000000)
+      + (byteAt(offset + 2) << 16)
+      + (byteAt(offset + 3) << 8)
+      + byteAt(offset + 4);
+    const available = Math.min(length, totalBytes - offset - 5);
+    payloadBytes += available;
+    if (available < length) break;
+    offset += 5 + length;
+  }
+  return payloadBytes;
+}
+
 /** Host-only label for Cursor transport diagnostics — never leaks path/query/credentials. */
 function cursorHostLabel(baseUrl: string): string {
   try {