fix(cloud-agent-next): isolate scoped Stop across the service boundary

pnpm-workspace.yaml

@@ -42,7 +42,7 @@ catalog:
   ulid: 3.0.2
   vitest: 4.1.6
   workers-tagged-logger: 1.0.0
-  wrangler: 4.127.1
+  wrangler: 4.112.0
   zod: 4.4.3
 
 minimumReleaseAge: 6842

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

@@ -875,6 +875,31 @@ export class SandboxControl extends DurableObject<Env> {
         throw error;
       }
       const result = response.ok ? sessionAbortResultSchema.safeParse(response.result) : undefined;
+      if (response.ok && !result?.success) {
+        if (retirementAttempt) await this.nativeRuntimeRetirement.defer(retirementAttempt);
+        return errorResponse(
+          response.requestId,
+          'protocol_error',
+          'Invalid session abort result',
+          false
+        );
+      }
+      if (result?.success && result.data.cleanupScope === 'root') {
+        if (
+          !this.socketHandler.supportsScopedCleanupResult?.() ||
+          result.data.runtimeRetired === true
+        ) {
+          if (retirementAttempt) await this.nativeRuntimeRetirement.defer(retirementAttempt);
+          return errorResponse(
+            response.requestId,
+            'protocol_error',
+            'Invalid root-scoped session abort result',
+            false
+          );
+        }
+        if (retirementAttempt) await this.nativeRuntimeRetirement.release(retirementAttempt);
+        return response;
+      }
       if (
         result?.success &&
         result.data.runtimeRetired === true &&

services/cloud-agent-next/src/sandbox-control/frames.test.ts

@@ -69,6 +69,9 @@ describe('sandbox control frames', () => {
     expect(
       sandboxHelloResultSchema.safeParse({ ...helloResult(), handshakeComplete: false }).success
     ).toBe(false);
+    expect(
+      sandboxHelloResultSchema.parse(helloResult({ scopedCleanupResult: true })).capabilities
+    ).toMatchObject({ scopedCleanupResult: true });
   });
 
   it('accepts a valid request envelope', () => {

services/cloud-agent-next/src/sandbox-control/frames.ts

@@ -186,6 +186,7 @@ export function errorResponse(
 export function helloResult(capabilities?: {
   connectionRecovery?: boolean;
   eventReceipts?: boolean;
+  scopedCleanupResult?: boolean;
 }): SandboxHelloResult {
   return {
     protocolVersion: SANDBOX_CONTROL_PROTOCOL_VERSION,
@@ -197,6 +198,7 @@ export function helloResult(capabilities?: {
       nativeRuntimeRetirement: true,
       ...(capabilities?.connectionRecovery ? { connectionRecovery: true } : {}),
       ...(capabilities?.eventReceipts ? { eventReceipts: true } : {}),
+      ...(capabilities?.scopedCleanupResult ? { scopedCleanupResult: true } : {}),
     },
   };
 }

services/cloud-agent-next/src/sandbox-control/lifecycle.test.ts

@@ -2756,6 +2756,199 @@ describe('SandboxControl lifecycle boundaries', () => {
     ]);
   });
 
+  it('releases a negotiated unconfirmed root-scoped Stop without retiring the runtime', async () => {
+    const h = await harness();
+    await h.create();
+    const identity = await h.ready();
+    const nativeRuntimeId = '11111111-1111-4111-8111-111111111111';
+    const operationId = '33333333-3333-4333-8333-333333333333';
+    const [route] = await h.control.listRoutes();
+    if (!route) throw new Error('Missing route');
+    const routeB = {
+      ...route,
+      sessionId: 'workspace_22222222-2222-4222-8222-222222222222',
+      kiloSessionId: 'ses_22222222222222222222222222',
+    };
+    h.records.set('session_routes', [
+      { ...route, nativeRuntimeId },
+      { ...routeB, nativeRuntimeId },
+    ]);
+    h.socket.supportsNativeRuntimeRetirement = () => true;
+    h.socket.supportsScopedStopAbort = () => true;
+    h.socket.supportsScopedCleanupResult = () => true;
+    const rootResult = {
+      type: 'response' as const,
+      requestId: 'stop_root_unconfirmed',
+      ok: true as const,
+      result: {
+        status: 'unconfirmed' as const,
+        cleanupScope: 'root' as const,
+        quiescent: false,
+      },
+    };
+    h.sendRequest.mockImplementationOnce(async () => {
+      expect(h.records.get('native_runtime_retirements')).toEqual([
+        expect.objectContaining({ operationId, state: 'pending' }),
+      ]);
+      return rootResult;
+    });
+
+    await expect(
+      h.control.request({
+        operation: 'session.abort',
+        session: {
+          sessionId: route.sessionId,
+          kiloSessionId: route.kiloSessionId,
+          directory: route.directory,
+        },
+        payload: {
+          messageId: 'message_root_unconfirmed',
+          operationId,
+          cleanupDeadlineAt: Date.now() + 1_000,
+        },
+        expectedWrapperInstanceId: identity.wrapperInstanceId,
+      })
+    ).resolves.toEqual(rootResult);
+    expect(h.records.get('native_runtime_retirements')).toEqual([
+      expect.objectContaining({
+        operationId,
+        state: 'released',
+        disposition: 'operation_only',
+      }),
+    ]);
+    const routesAfterRootRelease = await h.control.listRoutes();
+    expect(routesAfterRootRelease).toHaveLength(2);
+    for (const currentRoute of routesAfterRootRelease) {
+      expect(currentRoute.nativeRuntimeId).toBe(nativeRuntimeId);
+      expect(currentRoute.retiringNativeRuntimeId).toBeUndefined();
+    }
+    expect(h.session.invalidateTerminalRuntime).not.toHaveBeenCalled();
+
+    const bAdmission = {
+      type: 'response' as const,
+      requestId: 'prompt_b_after_root_release',
+      ok: true as const,
+      result: { status: 'accepted' as const },
+    };
+    h.sendRequest.mockResolvedValueOnce(bAdmission);
+    await expect(
+      h.control.request({
+        operation: 'session.prompt',
+        session: {
+          sessionId: routeB.sessionId,
+          kiloSessionId: routeB.kiloSessionId,
+          directory: routeB.directory,
+        },
+        payload: {
+          messageId: 'message_b_after_root_release',
+          turn: { type: 'prompt', prompt: 'B remains live' },
+          agent: { mode: 'code', model: 'test' },
+        },
+        expectedWrapperInstanceId: identity.wrapperInstanceId,
+      })
+    ).resolves.toEqual(bAdmission);
+
+    const requestsBeforeMaintenance = h.sendRequest.mock.calls.length;
+    await h.fireAlarm();
+    expect(h.sendRequest.mock.calls.length).toBe(requestsBeforeMaintenance);
+  });
+
+  it('releases a known superseded operation-only result through the operation-only disposition', async () => {
+    const h = await harness();
+    await h.create();
+    const identity = await h.ready();
+    const nativeRuntimeId = '11111111-1111-4111-8111-111111111111';
+    const operationId = '44444444-4444-4444-8444-444444444444';
+    const [route] = await h.control.listRoutes();
+    if (!route) throw new Error('Missing route');
+    const routeB = {
+      ...route,
+      sessionId: 'workspace_22222222-2222-4222-8222-222222222222',
+      kiloSessionId: 'ses_22222222222222222222222222',
+    };
+    h.records.set('session_routes', [
+      { ...route, nativeRuntimeId },
+      { ...routeB, nativeRuntimeId },
+    ]);
+    h.socket.supportsNativeRuntimeRetirement = () => true;
+    h.socket.supportsScopedStopAbort = () => true;
+    h.socket.supportsScopedCleanupResult = () => true;
+    const rootResult = {
+      type: 'response' as const,
+      requestId: 'stop_root',
+      ok: true as const,
+      result: {
+        status: 'already_idle' as const,
+        quiescent: false,
+      },
+    };
+    h.sendRequest.mockImplementationOnce(async () => {
+      expect(h.records.get('native_runtime_retirements')).toEqual([
+        expect.objectContaining({ operationId, state: 'pending' }),
+      ]);
+      return rootResult;
+    });
+
+    await expect(
+      h.control.request({
+        operation: 'session.abort',
+        session: {
+          sessionId: route.sessionId,
+          kiloSessionId: route.kiloSessionId,
+          directory: route.directory,
+        },
+        payload: {
+          messageId: 'message_root',
+          operationId,
+          cleanupDeadlineAt: Date.now() + 1_000,
+        },
+        expectedWrapperInstanceId: identity.wrapperInstanceId,
+      })
+    ).resolves.toEqual(rootResult);
+    expect(h.records.get('native_runtime_retirements')).toEqual([
+      expect.objectContaining({
+        operationId,
+        state: 'released',
+        disposition: 'operation_only',
+      }),
+    ]);
+    const routesAfterRootRelease = await h.control.listRoutes();
+    expect(routesAfterRootRelease).toHaveLength(2);
+    for (const currentRoute of routesAfterRootRelease) {
+      expect(currentRoute.nativeRuntimeId).toBe(nativeRuntimeId);
+      expect(currentRoute.retiringNativeRuntimeId).toBeUndefined();
+    }
+    expect(h.session.invalidateTerminalRuntime).not.toHaveBeenCalled();
+
+    const bAdmission = {
+      type: 'response' as const,
+      requestId: 'prompt_b',
+      ok: true as const,
+      result: { status: 'accepted' as const },
+    };
+    h.sendRequest.mockResolvedValueOnce(bAdmission);
+    await expect(
+      h.control.request({
+        operation: 'session.prompt',
+        session: {
+          sessionId: routeB.sessionId,
+          kiloSessionId: routeB.kiloSessionId,
+          directory: routeB.directory,
+        },
+        payload: {
+          messageId: 'message_b',
+          turn: { type: 'prompt', prompt: 'B remains live' },
+          agent: { mode: 'code', model: 'test' },
+        },
+        expectedWrapperInstanceId: identity.wrapperInstanceId,
+      })
+    ).resolves.toEqual(bAdmission);
+
+    const requestsBeforeMaintenance = h.sendRequest.mock.calls.length;
+    await h.fireAlarm();
+    expect(h.sendRequest.mock.calls.length).toBe(requestsBeforeMaintenance);
+  });
+
   it('keeps an operation-only Stop replay from blocking or changing a fresh native retirement', async () => {
     const h = await harness();
     await h.create();

services/cloud-agent-next/src/sandbox-control/scoped-stop-maintenance.test.ts

@@ -4,6 +4,10 @@ import {
   parseScopedStopMaintenance,
   stopAbortWirePayload,
 } from './scoped-stop-maintenance.js';
+import {
+  sessionAbortPayloadSchema,
+  sessionAbortResultSchema,
+} from '../shared/sandbox-control-protocol.js';
 
 const OPERATION_ID = '33333333-3333-4333-8333-333333333333';
 
@@ -43,4 +47,40 @@ describe('scoped Stop maintenance', () => {
     );
     expect(hasScopedStopMaintenanceFields({ cleanupDeadlineAt: 11_000 })).toBe(true);
   });
+
+  it('rejects root-scoped results that claim physical quiescence or retirement', () => {
+    expect(
+      sessionAbortResultSchema.safeParse({
+        status: 'aborted',
+        quiescent: true,
+        cleanupScope: 'root',
+      }).success
+    ).toBe(false);
+    expect(
+      sessionAbortResultSchema.safeParse({
+        status: 'aborted',
+        quiescent: false,
+        cleanupScope: 'root',
+        runtimeRetired: true,
+      }).success
+    ).toBe(false);
+    expect(
+      sessionAbortResultSchema.safeParse({
+        status: 'unconfirmed',
+        quiescent: false,
+        cleanupScope: 'root',
+      }).success
+    ).toBe(true);
+  });
+
+  it('does not accept request-side cleanup scope from a Stop caller', () => {
+    expect(
+      sessionAbortPayloadSchema.safeParse({
+        messageId: 'a',
+        operationId: '11111111-1111-4111-8111-111111111111',
+        cleanupDeadlineAt: Date.now() + 1_000,
+        cleanupScope: 'root',
+      }).success
+    ).toBe(false);
+  });
 });

services/cloud-agent-next/src/sandbox-control/socket.test.ts

@@ -64,7 +64,11 @@ function helloFrame(
   providerInstanceId: string,
   wrapperInstanceId?: string,
   requestId = 'req_hello',
-  capabilities?: { nativeRuntimeRetirement?: boolean; workingBranches?: boolean }
+  capabilities?: {
+    nativeRuntimeRetirement?: boolean;
+    scopedCleanupResult?: boolean;
+    workingBranches?: boolean;
+  }
 ): string {
   return JSON.stringify({
     type: 'request',
@@ -396,6 +400,23 @@ describe('sandbox control socket handler', () => {
     expect(handler.supportsWorkingBranches?.()).toBe(true);
   });
 
+  it('grants scoped cleanup results only to a reader that offers the capability', async () => {
+    const incoming = createFakeWebSocket();
+    const handler = createSandboxControlSocketHandler(createFakeState([incoming]), 'sbx_test');
+
+    await handler.handleMessage(
+      asWs(incoming),
+      helloFrame('inst_1', WRAPPER_INSTANCE_ID, 'req_scoped_cleanup', {
+        scopedCleanupResult: true,
+      })
+    );
+
+    expect(handler.supportsScopedCleanupResult?.()).toBe(true);
+    expect(incoming.send).toHaveBeenCalledWith(
+      expect.stringContaining('"scopedCleanupResult":true')
+    );
+  });
+
   it('rejects duplicate hellos without replacing the current connection', async () => {
     const current = createFakeWebSocket({
       handshakeComplete: true,

services/cloud-agent-next/src/sandbox-control/socket.ts

@@ -124,6 +124,7 @@ export type SandboxControlSocketHandler = {
   hasHandshakenSocket(): boolean;
   supportsOperationResults(): boolean;
   supportsScopedStopAbort(): boolean;
+  supportsScopedCleanupResult?(): boolean;
   supportsNativeRuntimeRetirement(): boolean;
   supportsWorkingBranches?(): boolean;
   supportsConnectionRecovery(): boolean;
@@ -363,6 +364,14 @@ export function createSandboxControlSocketHandler(
       );
     },
 
+    supportsScopedCleanupResult(): boolean {
+      const current = currentHandshakenSocket(state);
+      return (
+        current !== null &&
+        readAttachment(current.socket)?.capabilities?.scopedCleanupResult === true
+      );
+    },
+
     supportsNativeRuntimeRetirement(): boolean {
       const current = currentHandshakenSocket(state);
       return (
@@ -602,6 +611,7 @@ export function createSandboxControlSocketHandler(
             helloResult({
               connectionRecovery: payload.capabilities?.connectionRecovery === true,
               eventReceipts: payload.capabilities?.eventReceipts === true,
+              scopedCleanupResult: payload.capabilities?.scopedCleanupResult === true,
             })
           )
         );