fix(eval): require a server admission token before preserve-detach

Transport in-flight is not Host admission. preserve_environment now
calls hosted.execution.admit and detaches only after that token
returns. A frame-written interrupt without the token settles the
Host instead of claiming execution continues.

Cancelled relay cleanup funnels every finalize exception through
settle-or-destroy before rethrowing, including execution and
persist failures during host abort.

Fixes #3150

Author: 1625567290

packages/cli/src/__tests__/runtime-host-operator-command.test.ts

@@ -138,6 +138,7 @@ describe('Runtime Host operator commands', () => {
         'access.credential.issue',
         'access.credential.revoke',
         'host.upgrade.prepare',
+        'hosted.execution.admit',
         'hosted.execution.cancel',
         'hosted.execution.start',
       ],

packages/eval/harbor/relay_agent.py

@@ -172,24 +172,16 @@ async def run(self, instruction: str, environment: Any, context: Any) -> None:
             current = asyncio.current_task()
             if current is not None and hasattr(current, "uncancel"):
                 current.uncancel()
-            try:
-                if request is not None and execution is not None:
-                    await self._finalize_cancelled_execution(
-                        environment,
-                        cwd,
-                        scope_path,
-                        execution,
-                        request,
-                        writer,
-                        execution_reported,
-                    )
-            except asyncio.CancelledError:
-                if request is not None and execution is not None:
-                    with contextlib.suppress(Exception):
-                        await _settle_or_destroy(
-                            environment, cwd, scope_path, execution, self._teardown_timeout
-                        )
-                raise
+            if request is not None and execution is not None:
+                await self._cleanup_cancelled_execution(
+                    environment,
+                    cwd,
+                    scope_path,
+                    execution,
+                    request,
+                    writer,
+                    execution_reported,
+                )
             raise
         except RelayTransportClosed:
             if request is not None and execution is not None:
@@ -227,6 +219,36 @@ async def run(self, instruction: str, environment: Any, context: Any) -> None:
                 writer.close()
                 await asyncio.wait_for(writer.wait_closed(), timeout=1)
 
+    async def _cleanup_cancelled_execution(
+        self,
+        environment: Any,
+        cwd: str,
+        scope_path: str,
+        execution: asyncio.Task[Any],
+        request: dict[str, Any],
+        writer: Any,
+        execution_reported: bool,
+    ) -> None:
+        try:
+            await self._finalize_cancelled_execution(
+                environment,
+                cwd,
+                scope_path,
+                execution,
+                request,
+                writer,
+                execution_reported,
+            )
+        except BaseException:
+            with contextlib.suppress(Exception):
+                await asyncio.wait_for(
+                    _settle_or_destroy(
+                        environment, cwd, scope_path, execution, self._teardown_timeout
+                    ),
+                    timeout=self._teardown_timeout,
+                )
+            raise
+
     async def _finalize_cancelled_execution(
         self,
         environment: Any,

packages/eval/harbor/test_relay_lifecycle.py

@@ -83,6 +83,17 @@ async def exec(self, command, cwd=None, timeout_sec=None):
         return SimpleNamespace(return_code=0, stdout="", stderr="")
 
 
+class DummyWriter:
+    def is_closing(self):
+        return False
+
+    def write(self, _value):
+        return None
+
+    async def drain(self):
+        return None
+
+
 class ClosedWriter:
     def is_closing(self):
         return False
@@ -188,6 +199,20 @@ async def exec(self, command, cwd=None, timeout_sec=None):
         return await super().exec(command, cwd=cwd, timeout_sec=timeout_sec)
 
 
+class ExplodingSubjectEnvironment:
+    def __init__(self):
+        self.stopped = False
+
+    async def exec(self, command, cwd=None, timeout_sec=None):
+        return SimpleNamespace(return_code=0, stdout="", stderr="")
+
+    async def stop(self, delete=False):
+        self.stopped = delete
+
+    async def upload_file(self, source, target):
+        return None
+
+
 class SlowLeaderStopEnvironment(IgnoringLeaderEnvironment):
     def __init__(self):
         super().__init__()
@@ -218,6 +243,19 @@ async def exec(self, command, cwd=None, timeout_sec=None):
         return await super().exec(command, cwd=cwd, timeout_sec=timeout_sec)
 
 
+class PersistFailureEnvironment(LiveScopeEnvironment):
+    def __init__(self):
+        super().__init__()
+        self.stopped = False
+
+    async def stop(self, delete=False):
+        self.stopped = delete
+
+    async def upload_file(self, source, target):
+        if str(target).endswith("maka-subject.stdout.txt"):
+            raise RuntimeError("persist failed")
+
+
 class TransportLossEnvironment(SimultaneousEnvironment):
     async def exec(self, command, cwd=None, timeout_sec=None):
         if "kill -TERM" in command:
@@ -734,6 +772,87 @@ async def accept(reader, writer):
             server.close()
             await server.wait_closed()
 
+    async def test_execution_exception_during_host_abort_still_destroys(self):
+        relay = load_relay()
+        environment = ExplodingSubjectEnvironment()
+
+        async def boom():
+            raise RuntimeError("subject execution exploded")
+
+        execution = asyncio.create_task(boom())
+        await asyncio.sleep(0)
+        relay.request_host_teardown()
+        agent = relay.RelayAgent(
+            logs_dir=Path(tempfile.gettempdir()),
+            relay_host="127.0.0.1",
+            relay_port=1,
+            relay_token="token",
+            teardown_timeout_ms=1_000,
+        )
+        with self.assertRaisesRegex(RuntimeError, "subject execution exploded"):
+            await agent._cleanup_cancelled_execution(
+                environment,
+                "/",
+                "/tmp/missing",
+                execution,
+                {"resultToken": "0" * 32, "captureStdout": True},
+                DummyWriter(),
+                False,
+            )
+        self.assertTrue(environment.stopped)
+
+    async def test_persistence_failure_during_host_abort_still_destroys(self):
+        relay = load_relay()
+        environment = PersistFailureEnvironment()
+        token = f"host-persist-{os.getpid()}"
+        connected = asyncio.get_running_loop().create_future()
+
+        async def accept(reader, writer):
+            connected.set_result((reader, writer))
+
+        server = await asyncio.start_server(accept, "127.0.0.1", 0)
+        port = server.sockets[0].getsockname()[1]
+        agent = relay.RelayAgent(
+            logs_dir=Path(tempfile.gettempdir()),
+            relay_host="127.0.0.1",
+            relay_port=port,
+            relay_token=token,
+            teardown_timeout_ms=1_000,
+        )
+        running = asyncio.create_task(agent.run("solve", environment, None))
+        reader, writer = await connected
+        try:
+            await reader.readline()
+            writer.write(
+                (
+                    __import__("json").dumps(
+                        {
+                            "token": token,
+                            "kind": "execute",
+                            "command": "/bin/true",
+                            "args": [],
+                            "credentials": {},
+                            "resultToken": "0" * 32,
+                        }
+                    )
+                    + "\n"
+                ).encode()
+            )
+            await writer.drain()
+            environment.release.set()
+            await environment.finished.wait()
+            relay.request_host_teardown()
+            running.cancel()
+            with self.assertRaisesRegex(RuntimeError, "persist failed"):
+                await asyncio.wait_for(running, timeout=1)
+            self.assertTrue(
+                any(_is_teardown(command) for command in environment.commands),
+            )
+        finally:
+            writer.close()
+            server.close()
+            await server.wait_closed()
+
     async def test_framework_timeout_stops_the_subject_without_the_process_group(self):
         relay = load_relay()
         environment = TimeoutScopeEnvironment()

packages/runtime-host/src/__tests__/hosted-execution-client.test.ts

@@ -94,18 +94,21 @@ test('post-connect cancellation reports the owned Host settlement outcome', asyn
 
 test('environment-preserving abort detaches without cancelling or settling the Host', async () => {
   const abort = new AbortController();
-  const started = deferred();
+  const admitted = deferred();
   const closed = deferred();
   const events: string[] = [];
   const connected = ownedHost({
     request: async (operation: string) => {
       events.push(operation);
+      if (operation === 'hosted.execution.admit') {
+        admitted.resolve();
+        return { executionId: ID, admissionToken: ADMISSION_TOKEN };
+      }
       if (operation !== 'hosted.execution.start') {
         throw new Error(`Unexpected operation ${operation}`);
       }
-      started.resolve();
       await closed.promise;
-      throw admittedInterrupt();
+      throw dispatchedInterrupt();
     },
   });
   connected.connection.close = async () => {
@@ -124,29 +127,34 @@ test('environment-preserving abort detaches without cancelling or settling the H
     { ...input(abort.signal), abortPolicy: 'preserve_environment' },
     { connectOwnedRuntimeHost: async () => connected as never },
   );
-  await started.promise;
+  await admitted.promise;
   abort.abort();
   const result = await execution;
 
   assert.equal(result.kind, 'indeterminate');
   assert.equal(result.failureReason, 'Hosted execution continues for environment verification');
-  assert.deepEqual(events, ['hosted.execution.start', 'close', 'release']);
+  assert.deepEqual(events, [
+    'hosted.execution.admit',
+    'close',
+    'hosted.execution.start',
+    'release',
+  ]);
 });
 
-test('pre-admission preserve abort does not claim execution continues', async () => {
+test('frame-written admit interruption is not a server admission', async () => {
   const abort = new AbortController();
   const started = deferred();
   const closed = deferred();
   const events: string[] = [];
   const connected = ownedHost({
     request: async (operation: string) => {
       events.push(operation);
-      if (operation !== 'hosted.execution.start') {
+      if (operation !== 'hosted.execution.admit') {
         throw new Error(`Unexpected operation ${operation}`);
       }
       started.resolve();
       await closed.promise;
-      throw queuedInterrupt();
+      throw dispatchedInterrupt();
     },
   });
   connected.connection.close = async () => {
@@ -172,7 +180,7 @@ test('pre-admission preserve abort does not claim execution continues', async ()
 
   assert.equal(result.kind, 'indeterminate');
   assert.equal(result.failureReason, 'Hosted execution was cancelled');
-  assert.deepEqual(events, ['hosted.execution.start', 'close', 'settle']);
+  assert.deepEqual(events, ['hosted.execution.admit', 'close', 'settle']);
 });
 
 test('environment-preserving abort before start does not claim execution continues', async () => {
@@ -322,6 +330,7 @@ test('explicit target reconnect cancellation preserves the cancelled outcome', a
 
 const ID = '00000000-0000-4000-8000-000000000001';
 const CONNECTION_ID = '00000000-0000-4000-8000-000000000002';
+const ADMISSION_TOKEN = '00000000-0000-4000-8000-0000000000ad';
 
 function input(signal?: AbortSignal) {
   return {
@@ -458,24 +467,15 @@ function ownedHost(connection: Record<string, unknown>, clean = false) {
   };
 }
 
-function admittedInterrupt() {
+function dispatchedInterrupt() {
   return new RuntimeHostRequestInterruptedError(
-    'hosted.execution.start',
+    'hosted.execution.admit',
     'command',
     'dispatched',
     'connection_lost',
   );
 }
 
-function queuedInterrupt() {
-  return new RuntimeHostRequestInterruptedError(
-    'hosted.execution.start',
-    'command',
-    'not_dispatched',
-    'connection_lost',
-  );
-}
-
 function deferred() {
   let resolve!: () => void;
   const promise = new Promise<void>((accept) => {

packages/runtime-host/src/__tests__/hosted-execution-coordinator.test.ts

@@ -24,6 +24,32 @@ test('cancel before start prevents the hosted execution from running', async ()
   assert.equal(runs, 0);
 });
 
+test('admit returns a server-owned token before start waits for settlement', async () => {
+  const release = deferred();
+  const coordinator = new HostHostedExecutionCoordinator(
+    async (input) => {
+      await release.promise;
+      return settled(input.executionId, 'completed');
+    },
+    () => {},
+  );
+
+  const admitted = await coordinator.handlers['hosted.execution.admit'](input(), context());
+  assert.equal(admitted.ok, true);
+  if (!admitted.ok) return;
+  assert.equal(admitted.result.executionId, ID);
+  assert.match(admitted.result.admissionToken, /^[0-9a-f-]{36}$/i);
+
+  const waiting = coordinator.handlers['hosted.execution.start'](input(), context());
+  const again = await coordinator.handlers['hosted.execution.admit'](input(), context());
+  assert.equal(again.ok, true);
+  if (again.ok) assert.equal(again.result.admissionToken, admitted.result.admissionToken);
+  release.resolve();
+  const started = await waiting;
+  assert.equal(started.ok, true);
+  if (started.ok) assert.equal(started.result.kind, 'settled');
+});
+
 test('cancelling a settled subject reclaims its verification environment', async () => {
   let drains = 0;
   const coordinator = new HostHostedExecutionCoordinator(
@@ -79,3 +105,11 @@ function context() {
     acquireResidency: () => ({ release() {} }),
   };
 }
+
+function deferred() {
+  let resolve!: () => void;
+  const promise = new Promise<void>((accept) => {
+    resolve = accept;
+  });
+  return { promise, resolve };
+}

packages/runtime-host/src/__tests__/protocol.test.ts

@@ -46,6 +46,12 @@ describe('Runtime Host bootstrap protocol', () => {
     assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 22);
   });
 
+  test('publishes a new compatibility epoch for hosted.execution.admit', () => {
+    // Epoch 25 has start and cancel only. Mixed-version peers must fail
+    // during handshake instead of sending an unknown admit command.
+    assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 25);
+  });
+
   test('rejects the legacy connection update result in the current compatibility epoch', () => {
     assert.throws(
       () =>